HADOOP-10564. Add username to native RPCv9 client (cmccabe)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HADOOP-10388@1595028 13f79535-47bb-0310-9956-ffa450edef68

hadoop-native-core/CMakeLists.txt

@@ -55,6 +55,7 @@ set(COMMON_SRCS
     common/net.c
     common/string.c
     common/test.c
+    common/user.c
 )
 set(COMMON_DEPS
     pthread

hadoop-native-core/common/user.c

@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "common/user.h"
+
+#include <errno.h>
+#include <pwd.h>
+#include <stdarg.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#define GETPWUID_R_BUFLEN_MIN 4096
+#define GETPWUID_R_BUFLEN_MAX (8 * 1024 * 1024)
+
+/**
+ * Get the user name associated with a given numeric user ID.
+ *
+ * The POSIX APIs to do this require a little more code than you might expect.
+ * We have to supply a buffer to the getpwuid_r call.  But how big should it
+ * be?  We can call sysconf to get a "suggested size," but this may or may not
+ * be big enough for our particular user.  Also, on some platforms, sysconf
+ * returns -1 for this.  So we have a loop which keeps doubling the size of the
+ * buffer.  As a sanity check against buggy libraries, we give up when we reach
+ * some ridiculous buffer size.
+ *
+ * @param uid           UID to look up
+ * @param out           (out param) on success, the user name.
+ *
+ * @return              0 on success; error code on failure.
+ */
+static int uid_to_string(uid_t uid, char **out)
+{
+    int ret;
+    struct passwd pwd, *result;
+    char *buf = NULL, *nbuf;
+    size_t buflen, nbuflen;
+
+    buflen = sysconf(_SC_GETPW_R_SIZE_MAX);
+    if (buflen < GETPWUID_R_BUFLEN_MIN) {
+        buflen = GETPWUID_R_BUFLEN_MIN;
+    }
+    while (1) {
+        nbuf = realloc(buf, buflen);
+        if (!nbuf) {
+            ret = ENOMEM;
+            goto done;
+        }
+        buf = nbuf;
+        ret = getpwuid_r(uid, &pwd, buf, buflen, &result);
+        if (ret == 0)
+            break;
+        if (ret != ERANGE) {
+            fprintf(stderr, "geteuid_string: getpwuid_r(%lld) failed "
+                    "with error %d\n", (long long)uid, ret);
+            goto done;
+        }
+        nbuflen = buflen *2;
+        if (nbuflen > GETPWUID_R_BUFLEN_MAX) {
+            nbuflen = GETPWUID_R_BUFLEN_MAX;
+        }
+        if (buflen == nbuflen) {
+            fprintf(stderr, "geteuid_string: getpwuid_r(%lld) still gets "
+                    "ERANGE with buflen %zd\n", (long long)uid, buflen);
+            ret = ERANGE;
+            goto done;
+        }
+        buflen = nbuflen;
+    }
+    if (!result) {
+        ret = ENOENT;
+        fprintf(stderr, "geteuid_string: getpwuid_r(%lld): no name "
+                "found for current effective user id.\n", (long long)uid);
+        goto done;
+    }
+    *out = strdup(result->pw_name);
+    if (!*out) {
+        ret = ENOMEM;
+        goto done;
+    }
+
+done:
+    free(buf);
+    return ret;
+}
+
+int geteuid_string(char **out)
+{
+    return uid_to_string(geteuid(), out);
+}
+
+// vim: ts=4:sw=4:tw=79:et

hadoop-native-core/common/user.h

@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef HADOOP_CORE_COMMON_USER_H
+#define HADOOP_CORE_COMMON_USER_H
+
+/**
+ * Get the current effective user name.
+ *
+ * @param out           (out param) The name of the current user.
+ *                      A malloc'ed string which must be freed.
+ *
+ * @return              0 on success; error code otherwise.
+ */
+int geteuid_string(char **out);
+
+#endif
+
+// vim: ts=4:sw=4:tw=79:et

hadoop-native-core/hdfs/namenode-rpc-unit.c

@@ -18,6 +18,7 @@
 
 #include "common/hadoop_err.h"
 #include "common/test.h"
+#include "common/user.h"
 #include "protobuf/ClientNamenodeProtocol.call.h"
 #include "rpc/messenger.h"
 #include "rpc/proxy.h"
@@ -31,12 +32,14 @@
 
 struct options {
     struct sockaddr_in remote;
+    char *username;
 };
 
 static void options_from_env(struct options *opts)
 {
     const char *ip_str;
     const char *port_str;
+    const char *username;
     int res, port;
 
     ip_str = getenv("HDFS_IP");
@@ -58,6 +61,19 @@ static void options_from_env(struct options *opts)
                 ip_str, port, uv_strerror(res));
         exit(EXIT_FAILURE);
     }
+    username = getenv("HDFS_USERNAME");
+    if (username) {
+        opts->username = strdup(username);
+        if (!opts->username)
+            abort();
+        fprintf(stderr, "using HDFS username %s\n", username);
+    } else {
+        res = geteuid_string(&opts->username);
+        if (res) {
+            fprintf(stderr, "geteuid_string failed with error %d\n", res);
+            abort();
+        }
+    }
 }
 
 void set_replication_cb(SetReplicationResponseProto *resp,
@@ -87,9 +103,8 @@ void set_replication_cb(SetReplicationResponseProto *resp,
 int main(void)
 {
     struct hrpc_messenger_builder *msgr_bld;
-    struct hrpc_proxy_builder *proxy_bld;
-    struct hrpc_proxy *proxy;
     struct hrpc_messenger *msgr;
+    struct hrpc_proxy proxy;
     struct options opts;
     uv_sem_t sem;
 
@@ -98,26 +113,24 @@ int main(void)
     msgr_bld = hrpc_messenger_builder_alloc();
     EXPECT_NONNULL(msgr_bld);
     EXPECT_NO_HADOOP_ERR(hrpc_messenger_create(msgr_bld, &msgr));
-    proxy_bld = hrpc_proxy_builder_alloc(msgr);
-    EXPECT_NONNULL(proxy_bld);
-    hrpc_proxy_builder_set_remote(proxy_bld, &opts.remote);
-    hrpc_proxy_builder_set_protocol(proxy_bld,
-                "org.apache.hadoop.hdfs.protocol.ClientProtocol");
-    EXPECT_NO_HADOOP_ERR(hrpc_proxy_create(proxy_bld, &proxy));
+
+    hrpc_proxy_init(&proxy, msgr, &opts.remote,
+            "org.apache.hadoop.hdfs.protocol.ClientProtocol",
+            opts.username);
     EXPECT_INT_ZERO(uv_sem_init(&sem, 0));
     {
         SetReplicationRequestProto req = SET_REPLICATION_REQUEST_PROTO__INIT;
         req.src = "/foo2";
         req.replication = 2;
-        cnn_async_set_replication(proxy, &req, set_replication_cb, &sem);
+        cnn_async_set_replication(&proxy, &req, set_replication_cb, &sem);
     }
     uv_sem_wait(&sem);
 
-    hrpc_proxy_free(proxy);
     hrpc_messenger_shutdown(msgr);
     hrpc_messenger_free(msgr);
     uv_sem_destroy(&sem);
 
+    free(opts.username);
     return EXIT_SUCCESS;
 }
 

hadoop-native-core/rpc/call.h

@@ -20,12 +20,14 @@
 #define HADOOP_CORE_RPC_CALL_H
 
 #include "common/queue.h"
-#include "rpc/proxy.h" // for hrpc_raw_cb_t 
 
 #include <netinet/in.h> // for struct sockaddr_in
 #include <stdint.h> // for int32_t
 #include <uv.h> // for uv_buf_t
 
+struct hadoop_err;
+struct hrpc_response;
+
 /**
  * The Hadoop call.
  *
@@ -33,7 +35,8 @@
  * include.
  */
 
-struct hrpc_call;
+typedef void (*hrpc_raw_cb_t)(struct hrpc_response *,
+    struct hadoop_err *, void *);
 
 /**
  * A Hadoop RPC call.
@@ -61,14 +64,19 @@ struct hrpc_call {
     void *cb_data;
 
     /**
-     * Malloc'ed payload to send.  Malloced.
+     * Malloc'ed payload to send.
      */
     uv_buf_t payload;
 
     /**
-     * The Java protocol we're using.  Malloced.
+     * String describing the protocol this call is using.
+     */
+    const char *protocol;
+
+    /**
+     * String describing the username this call is using. 
      */
-    char *protocol;
+    const char *username;
 
     /**
      * Nonzero if the call is currently active.  Must be set using atomic

hadoop-native-core/rpc/conn.c

@@ -24,10 +24,12 @@
 #include "rpc/call.h"
 #include "rpc/conn.h"
 #include "rpc/messenger.h"
+#include "rpc/proxy.h"
 #include "rpc/reactor.h"
 #include "rpc/varint.h"
 
 #include <errno.h>
+#include <inttypes.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -58,6 +60,8 @@ static const uint8_t FRAME[] = {
 
 #define FRAME_LEN sizeof(FRAME)
 
+#define CONNECTION_CONTEXT_CALL_ID (-3)
+
 static void conn_read_cb(uv_stream_t *stream, ssize_t nread,
                          const uv_buf_t* buf);
 static void hrpc_conn_read_alloc(uv_handle_t *handle,
@@ -98,6 +102,7 @@ static struct hadoop_err *conn_setup_ipc_header(struct hrpc_conn *conn)
     struct hrpc_conn_writer *writer = &conn->writer;
     IpcConnectionContextProto ipc_ctx = IPC_CONNECTION_CONTEXT_PROTO__INIT;
     RpcRequestHeaderProto rpc_req_header = RPC_REQUEST_HEADER_PROTO__INIT;
+    UserInformationProto userinfo = USER_INFORMATION_PROTO__INIT;
     int32_t cset_len, buf_len;
     int32_t ipc_ctx_len, rpc_req_header_len, off = 0;
     uint8_t *buf;
@@ -107,14 +112,15 @@ static struct hadoop_err *conn_setup_ipc_header(struct hrpc_conn *conn)
     rpc_req_header.rpckind = RPC_PROTOCOL_BUFFER;
     rpc_req_header.has_rpcop = 1;
     rpc_req_header.rpcop = RPC_FINAL_PACKET;
-    rpc_req_header.callid = -3; // ???
+    rpc_req_header.callid = CONNECTION_CONTEXT_CALL_ID;
     rpc_req_header.clientid.data = conn->client_id.buf;
     rpc_req_header.clientid.len = HRPC_CLIENT_ID_LEN;
     rpc_req_header.has_retrycount = 0;
     rpc_req_header_len =
         rpc_request_header_proto__get_packed_size(&rpc_req_header);
-
-    ipc_ctx.userinfo = NULL;
+    userinfo.effectiveuser = conn->username;
+    userinfo.realuser = NULL;
+    ipc_ctx.userinfo = &userinfo;
     ipc_ctx.protocol = conn->protocol;
     ipc_ctx_len = ipc_connection_context_proto__get_packed_size(&ipc_ctx);
 
@@ -156,7 +162,7 @@ static struct hadoop_err *hrpc_conn_setup_payload(struct hrpc_conn *conn)
     rpc_req_header.rpckind = RPC_PROTOCOL_BUFFER;
     rpc_req_header.has_rpcop = 1;
     rpc_req_header.rpcop = RPC_FINAL_PACKET;
-    rpc_req_header.callid = 0; // ???
+    rpc_req_header.callid = conn->call_id;
     rpc_req_header.clientid.data = conn->client_id.buf;
     rpc_req_header.clientid.len = HRPC_CLIENT_ID_LEN;
     rpc_req_header.has_retrycount = 0;
@@ -168,7 +174,7 @@ static struct hadoop_err *hrpc_conn_setup_payload(struct hrpc_conn *conn)
     total_len64 += call->payload.len;
     if (total_len64 > MAX_READER_BODY_LEN) {
         err = hadoop_lerr_alloc(EINVAL, "hrpc_conn_setup_payload: "
-                "can't send a payload of length %"PRId64".  The maximum "
+                "can't send a payload of length %" PRId64 ".  The maximum "
                 "payload length is %d", total_len64, MAX_READER_BODY_LEN);
         return err;
     }
@@ -255,6 +261,12 @@ static void conn_start_outbound(struct hrpc_conn *conn)
     struct hrpc_conn_writer *writer = &conn->writer;
     int res;
 
+    // Get next call ID to use.
+    if (conn->call_id >= 0x7fffffff) {
+        conn->call_id = 1;
+    } else {
+        conn->call_id++;
+    }
     writer->state = HRPC_CONN_WRITE_IPC_HEADER;
     err = conn_setup_ipc_header(conn);
     if (err) {
@@ -317,7 +329,8 @@ struct hadoop_err *hrpc_conn_create_outbound(struct hrpc_reactor *reactor,
     conn->call = call;
     conn->remote = call->remote;
     conn->protocol = strdup(call->protocol);
-    if (!conn->protocol) {
+    conn->username = strdup(call->username);
+    if ((!conn->protocol) || (!conn->username)) {
         err = hadoop_lerr_alloc(ENOMEM, "hrpc_conn_create_outbound: OOM");
         goto done;
     }
@@ -344,6 +357,7 @@ done:
     if (err) {
         if (conn) {
             free(conn->protocol);
+            free(conn->username);
             if (tcp_init) {
                 uv_close((uv_handle_t*)&conn->stream, NULL);
             }
@@ -357,7 +371,7 @@ done:
 
 int hrpc_conn_compare(const struct hrpc_conn *a, const struct hrpc_conn *b)
 {
-    int proto_cmp, a_active, b_active;
+    int proto_cmp, username_cmp, a_active, b_active;
 
     // Big-endian versus little-endian doesn't matter here.
     // We just want a consistent ordering on the same machine.
@@ -373,6 +387,10 @@ int hrpc_conn_compare(const struct hrpc_conn *a, const struct hrpc_conn *b)
     proto_cmp = strcmp(a->protocol, b->protocol);
     if (proto_cmp != 0)
         return proto_cmp;
+    // Compare username.
+    username_cmp = strcmp(a->username, b->username);
+    if (username_cmp != 0)
+        return username_cmp;
     // Make the inactive connections sort before the active ones.
     a_active = !!a->call;
     b_active = !!b->call;
@@ -390,7 +408,8 @@ int hrpc_conn_compare(const struct hrpc_conn *a, const struct hrpc_conn *b)
 }
 
 int hrpc_conn_usable(const struct hrpc_conn *conn,
-                      const struct sockaddr_in *addr, const char *protocol)
+            const struct sockaddr_in *addr,
+            const char *protocol, const char *username)
 {
     if (conn->remote.sin_addr.s_addr != addr->sin_addr.s_addr)
         return 0;
@@ -398,6 +417,8 @@ int hrpc_conn_usable(const struct hrpc_conn *conn,
         return 0;
     else if (strcmp(conn->protocol, protocol))
         return 0;
+    else if (strcmp(conn->username, username))
+        return 0;
     return 1;
 }
 
@@ -420,14 +441,14 @@ static struct hadoop_err *conn_deliver_resp(struct hrpc_conn *conn,
     // Check if the server sent us a bogus payload_len value.
     if (payload_len < 0) {
         return hadoop_lerr_alloc(EIO, "conn_deliver_resp: "
-            "server's payload_len was %"PRId32", but negative payload "
+            "server's payload_len was %" PRId32 ", but negative payload "
             "lengths are not valid.", payload_len);
     }
     payload_end = off;
     payload_end += payload_len;
     if (payload_end > reader->body_len) {
         return hadoop_lerr_alloc(EIO, "conn_deliver_resp: "
-            "server's payload_len was %"PRId64", but there are only %d "
+            "server's payload_len was %" PRId64 ", but there are only %d "
             "bytes left in the body buffer.", payload_end, reader->body_len);
     }
     // Reset the connection's read state.  We'll hold on to the response buffer
@@ -461,13 +482,13 @@ static struct hadoop_err *conn_process_response(struct hrpc_conn *conn)
     }
     if (resp_header_len <= 0) {
         err = hadoop_lerr_alloc(EIO, "conn_process_response: server sent "
-                "invalid resp_header_len of %"PRId32, resp_header_len);
+                "invalid resp_header_len of %" PRId32, resp_header_len);
         goto done;
     }
     rem = reader->body_len - off;
     if (resp_header_len > rem) {
         err = hadoop_lerr_alloc(EIO, "conn_process_response: server sent "
-                "resp_header_len of %"PRId32", but there were only %"PRId32
+                "resp_header_len of %" PRId32 ", but there were only %" PRId32
                 " bytes left in the response.", resp_header_len, rem);
         goto done;
     }
@@ -479,6 +500,14 @@ static struct hadoop_err *conn_process_response(struct hrpc_conn *conn)
         goto done;
     }
     off += resp_header_len;
+    if (resp_header->callid != conn->call_id) {
+        // We currently only send one request at a time.  So when we get a
+        // response, we expect it to be for the request we just sent.
+        err = hadoop_lerr_alloc(EIO, "conn_process_response: incorrect call "
+                "id in response.  Expected %" PRId32 ", got %" PRId32 ".",
+                conn->call_id, resp_header->callid);
+        goto done;
+    }
     if (resp_header->status != RPC_STATUS_PROTO__SUCCESS) {
         // TODO: keep connection open if we got an ERROR rather than a FATAL.
         err = hadoop_lerr_alloc(EIO, "conn_process_response: error %s: %s",
@@ -546,8 +575,8 @@ static void hrpc_conn_read_alloc(uv_handle_t *handle,
         rem = reader->body_len - reader->off;
         if (rem <= 0) {
             conn_log_warn(conn, "hrpc_conn_read_alloc: we're in state "
-                    "HRPC_CONN_READ_BODY with reader->body_len = %"PRId32", but "
-                    "reader->off = %"PRId32"\n", reader->body_len, reader->off);
+                    "HRPC_CONN_READ_BODY with reader->body_len = %" PRId32 ", but "
+                    "reader->off = %" PRId32 "\n", reader->body_len, reader->off);
             buf->base = NULL;
             buf->len = 0;
             return;
@@ -589,7 +618,7 @@ static void conn_read_cb(uv_stream_t *stream, ssize_t nread,
         reader->off += nread;
         if (reader->off < READLEN_BUF_LEN) {
             conn_log_debug(conn, "conn_read_cb: got partial read of "
-                           "body_len.  reader->off = %"PRId32"\n",
+                           "body_len.  reader->off = %" PRId32 "\n",
                            reader->off);
             return;
         }
@@ -597,12 +626,12 @@ static void conn_read_cb(uv_stream_t *stream, ssize_t nread,
         if ((reader->body_len <= 0) ||
                 (reader->body_len > MAX_READER_BODY_LEN)) {
             hrpc_conn_destroy(conn, hadoop_lerr_alloc(EIO, 
-                "conn_read_cb: got an invalid body length of %"PRId32"\n", 
+                "conn_read_cb: got an invalid body length of %" PRId32 "\n", 
                 reader->body_len));
             return;
         }
         conn_log_debug(conn, "conn_read_cb: got body length of "
-                       "%"PRId32".  Transitioning to HRPC_CONN_READ_BODY.\n",
+                       "%" PRId32 ".  Transitioning to HRPC_CONN_READ_BODY.\n",
                        reader->body_len);
         reader->off = 0;
         reader->state = HRPC_CONN_READ_BODY;
@@ -610,15 +639,15 @@ static void conn_read_cb(uv_stream_t *stream, ssize_t nread,
         if (!reader->body) {
             hrpc_conn_destroy(conn, hadoop_lerr_alloc(ENOMEM,
                     "hrpc_conn_read_alloc: failed to allocate "
-                    "%"PRId32" bytes.\n", reader->body_len));
+                    "%" PRId32 " bytes.\n", reader->body_len));
         }
         break;
     case HRPC_CONN_READ_BODY:
         reader->off += nread;
         if (reader->off < reader->body_len) {
             conn_log_debug(conn, "conn_read_cb: got partial read of "
-                           "body.  reader->off = %"PRId32" out of %"PRId32"\n",
-                           reader->off, reader->body_len);
+                           "body.  reader->off = %" PRId32 " out of %"
+                           PRId32 "\n", reader->off, reader->body_len);
             return;
         }
         err = conn_process_response(conn);
@@ -663,6 +692,7 @@ void hrpc_conn_destroy(struct hrpc_conn *conn, struct hadoop_err *err)
     free_write_bufs(conn);
     conn->writer.state = HRPC_CONN_WRITE_CLOSED;
     free(conn->protocol);
+    free(conn->username);
     uv_close((uv_handle_t*)&conn->stream, conn_free);
 }
 

hadoop-native-core/rpc/conn.h

@@ -133,6 +133,11 @@ struct hrpc_conn {
      */
     char *protocol;
 
+    /**
+     * The Hadoop username we're connected as.  Malloc'ed.
+     */
+    char *username;
+
     /**
      * The client ID we used when establishing the connection.
      */
@@ -143,6 +148,11 @@ struct hrpc_conn {
      */
     uv_connect_t conn_req;
 
+    /**
+     * The current call ID.
+     */
+    uint32_t call_id;
+
     struct hrpc_conn_writer writer;
     struct hrpc_conn_reader reader;
 };
@@ -198,11 +208,13 @@ int hrpc_conn_compare(const struct hrpc_conn *a,
  * @param conn          The connection.
  * @param addr          The address.
  * @param protocol      The protocol.
+ * @param username      The username.
  *
  * @return              1 if the connection is usable; 0 if not.
  */
 int hrpc_conn_usable(const struct hrpc_conn *conn,
-                      const struct sockaddr_in *addr, const char *protocol);
+                      const struct sockaddr_in *addr,
+                      const char *protocol, const char *username);
 
 /**
  * Destroy a connection.

hadoop-native-core/rpc/proxy.c

@@ -24,6 +24,7 @@
 #include "rpc/varint.h"
 
 #include <errno.h>
+#include <inttypes.h>
 #include <netinet/in.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -43,131 +44,22 @@
  */
 #define MAX_SEND_LEN (63 * 1024 * 1024)
 
-struct hrpc_proxy {
-    /**
-     * The messenger that this proxy is associated with.
-     */
-    struct hrpc_messenger *msgr;
-
-    /**
-     * Dynamically allocated string describing the protocol this proxy speaks. 
-     */
-    char *protocol;
-
-    /**
-     * The current call.
-     */
-    struct hrpc_call call;
-
-    /**
-     * A memory area which can be used by the current call.
-     *
-     * This will be null if userdata_len is 0.
-     */
-    uint8_t *userdata;
-
-    /**
-     * Length of userdata.
-     */
-    size_t userdata_len;
-};
-
 struct hrpc_proxy_builder {
     struct hrpc_proxy *proxy;
 };
 
 static const char OOM_ERROR[] = "OOM";
 
-struct hrpc_proxy_builder *hrpc_proxy_builder_alloc(
-            struct hrpc_messenger *msgr)
+void hrpc_proxy_init(struct hrpc_proxy *proxy,
+            struct hrpc_messenger *msgr,
+            const struct sockaddr_in *remote,
+            const char *protocol, const char *username)
 {
-    struct hrpc_proxy_builder *bld;
-
-    bld = calloc(1, sizeof(struct hrpc_proxy_builder));
-    if (!bld)
-        return NULL;
-    bld->proxy = calloc(1, sizeof(struct hrpc_proxy));
-    if (!bld->proxy) {
-        free(bld);
-        return NULL;
-    }
-    bld->proxy->msgr = msgr;
-    bld->proxy->call.remote.sin_addr.s_addr = INADDR_ANY;
-
-    return bld;
-}
-
-void hrpc_proxy_builder_free(struct hrpc_proxy_builder *bld)
-{
-    if (!bld)
-        return;
-    free(bld->proxy);
-    free(bld);
-}
-
-void hrpc_proxy_builder_set_protocol(struct hrpc_proxy_builder *bld,
-                                     const char *protocol)
-{
-    struct hrpc_proxy *proxy = bld->proxy;
-
-    if (proxy->protocol) {
-        if (proxy->protocol != OOM_ERROR) {
-            free(proxy->protocol);
-        }
-        proxy->protocol = NULL;
-    }
-    proxy->protocol = strdup(protocol);
-    if (!proxy->protocol) {
-        proxy->protocol = (char*)OOM_ERROR;
-    }
-}
-
-void hrpc_proxy_builder_set_remote(struct hrpc_proxy_builder *bld,
-                                     const struct sockaddr_in *remote)
-{
-    bld->proxy->call.remote = *remote;
-}
-
-struct hadoop_err *hrpc_proxy_create(struct hrpc_proxy_builder *bld,
-                            struct hrpc_proxy **out)
-{
-    struct hrpc_proxy *proxy;
-    
-    proxy = bld->proxy;
-    free(bld);
-    //fprintf(stderr, "proxy = %p, proxy->protocol = %s, proxy->call.cb = %p\n", proxy, proxy->protocol, proxy->call.cb);
-    if (proxy->call.remote.sin_addr.s_addr == INADDR_ANY) {
-        hrpc_proxy_free(proxy);
-        return hadoop_lerr_alloc(EINVAL, "hrpc_proxy_create: you must specify "
-                                 "a remote.");
-    }
-    if (!proxy->protocol) {
-        hrpc_proxy_free(proxy);
-        return hadoop_lerr_alloc(EINVAL, "hrpc_proxy_create: can't create "
-                                 "a proxy without a protocol argument.");
-    } else if (proxy->protocol == OOM_ERROR) {
-        // There was an OOM error during hrpc_proxy_builder_set_protocol.
-        hrpc_proxy_free(proxy);
-        return hadoop_lerr_alloc(ENOMEM, "hrpc_proxy_create: OOM error.");
-    }
-    *out = proxy;
-    return NULL;
-}
-
-void hrpc_proxy_free(struct hrpc_proxy *proxy)
-{
-    if (!proxy)
-        return;
-    if (hrpc_call_is_active(&proxy->call)) {
-        proxy_log_warn(proxy, "%s", "hrpc_proxy_free: attempt to free a proxy "
-                       "which is currently active!\n");
-        return;
-    }
-    if (proxy->protocol != OOM_ERROR) {
-        free(proxy->protocol);
-    }
-    free(proxy->userdata);
-    free(proxy);
+    memset(proxy, 0, sizeof(*proxy));
+    proxy->msgr = msgr;
+    proxy->protocol = protocol;
+    proxy->username = username;
+    proxy->call.remote = *remote;
 }
 
 struct hadoop_err *hrpc_proxy_activate(struct hrpc_proxy *proxy)
@@ -191,13 +83,8 @@ void hrpc_proxy_deactivate(struct hrpc_proxy *proxy)
 
 void *hrpc_proxy_alloc_userdata(struct hrpc_proxy *proxy, size_t size)
 {
-    if (size > proxy->userdata_len) {
-        uint8_t *new_userdata = realloc(proxy->userdata, size);
-        if (!new_userdata) {
-            return NULL;
-        }
-        proxy->userdata = new_userdata;
-        proxy->userdata_len = size;
+    if (size > RPC_PROXY_USERDATA_MAX) {
+        return NULL;
     }
     return proxy->userdata;
 }
@@ -244,15 +131,11 @@ void hrpc_proxy_start(struct hrpc_proxy *proxy,
 
     call->cb = cb;
     call->cb_data = cb_data;
-    call->protocol = strdup(proxy->protocol);
-    if (!call->protocol) {
-        hrpc_call_deliver_err(call, hadoop_lerr_alloc(ENOMEM,
-                "hrpc_proxy_start_internal: out of memory"));
-        return;
-    }
+    call->protocol = proxy->protocol;
+    call->username = proxy->username;
 
     req_header.methodname = (char*)method;
-    req_header.declaringclassprotocolname = proxy->protocol;
+    req_header.declaringclassprotocolname = (char*)proxy->protocol;
     req_header.clientprotocolversion = 1;
     req_header_len = request_header_proto__get_packed_size(&req_header);
     buf_len = varint32_size(req_header_len);
@@ -262,7 +145,7 @@ void hrpc_proxy_start(struct hrpc_proxy *proxy,
     if (buf_len >= MAX_SEND_LEN) {
         hrpc_call_deliver_err(call,
             hadoop_lerr_alloc(EINVAL, "hrpc_proxy_setup_header: the "
-                "request length is too long at %"PRId64 " bytes.  The "
+                "request length is too long at %" PRId64 " bytes.  The "
                 "maximum we will send is %d bytes.", buf_len, MAX_SEND_LEN));
         return;
     }
@@ -270,7 +153,7 @@ void hrpc_proxy_start(struct hrpc_proxy *proxy,
     if (!buf) {
         hrpc_call_deliver_err(call, 
             hadoop_lerr_alloc(ENOMEM, "hrpc_proxy_setup_header: "
-                "failed to allocate a buffer of length %"PRId64" bytes.",
+                "failed to allocate a buffer of length %" PRId64 " bytes.",
                 buf_len));
         return;
     }

hadoop-native-core/rpc/proxy.h

@@ -19,13 +19,13 @@
 #ifndef HADOOP_CORE_RPC_PROXY_H
 #define HADOOP_CORE_RPC_PROXY_H
 
+#include "rpc/call.h"
+
 #include <stdint.h> /* for uint8_t */
 #include <uv.h> /* for uv_buf_t */
 
 struct hadoop_err;
 struct hrpc_messenger;
-struct hrpc_proxy;
-struct hrpc_proxy_builder;
 
 struct hrpc_response {
     uint8_t *pb_base;
@@ -39,65 +39,58 @@ struct hrpc_sync_ctx {
     struct hrpc_response resp;
 };
 
-typedef void (*hrpc_raw_cb_t)(struct hrpc_response *,
-    struct hadoop_err *, void *);
-
 typedef size_t (*hrpc_pack_cb_t)(const void *, uint8_t *);
 
-/**
- * Allocate a Hadoop proxy builder.
- *
- * @param msgr      The Hadoop messenger that this proxy will be associated
- *                    with.
- * @return          A Hadoop proxy builder, or NULL on OOM.
- */
-struct hrpc_proxy_builder *hrpc_proxy_builder_alloc(
-                struct hrpc_messenger *msgr);
-
-/**
- * Free a Hadoop proxy builder.
- *
- * @param bld       The Hadoop proxy builder to free.
- */
-void hrpc_proxy_builder_free(struct hrpc_proxy_builder *bld);
-
-/**
- * Set the protocol used by a proxy.
- *
- * @param bld       The Hadoop proxy builder.
- * @param proto     The protocol string to use.  Will be deep-copied.
- */
-void hrpc_proxy_builder_set_protocol(struct hrpc_proxy_builder *bld,
-                                     const char *proto);
-
-/**
- * Set the remote that the proxy should connect to.
- *
- * @param bld       The Hadoop proxy builder.
- * @param remote    The remote.  Will be deep-copied.
- */
-void hrpc_proxy_builder_set_remote(struct hrpc_proxy_builder *bld,
-                                   const struct sockaddr_in *remote);
-
-/**
- * Create a Hadoop proxy
- *
- * @param bld       The Hadoop proxy builder to use.
- *                      The builder will be freed, even on failure.
- * @param out       (out param) On success, the Hadoop proxy.
- *
- * @return          On success, NULL.  On error, the error message.
- */
-struct hadoop_err *hrpc_proxy_create(struct hrpc_proxy_builder *bld,
-                            struct hrpc_proxy **out);
+#define RPC_PROXY_USERDATA_MAX 64
+
+struct hrpc_proxy {
+    /**
+     * The messenger that this proxy is associated with.
+     */
+    struct hrpc_messenger *msgr;
+
+    /**
+     * String describing the protocol this proxy speaks. 
+     */
+    const char *protocol;
+
+    /**
+     * String describing the username this proxy uses. 
+     */
+    const char *username;
+
+    /**
+     * The current call.
+     */
+    struct hrpc_call call;
+
+    /**
+     * A memory area which can be used by the current call.
+     *
+     * This will be null if userdata_len is 0.
+     */
+    uint8_t userdata[RPC_PROXY_USERDATA_MAX];
+};
 
 /**
- * Free a Hadoop proxy.
- *
- * @param proxy     The Hadoop proxy to free.  You must not attempt to free a
- *                      proxy with a call in progress.
+ * Initialize a Hadoop proxy.
+ *
+ * @param proxy     The Hadoop proxy to initialize.
+ * @param msgr      The messenger to associate the proxy with.
+ *                      This messenger must not be de-allocated while the proxy
+ *                      still exists.
+ * @param protocol  The protocol to use for this proxy.
+ *                      This string must remain valid for the lifetime of the
+ *                      proxy.
+ * @param remote    The remote to contact.  Will be copied.
+ * @param username  The username to use.
+ *                      This string must remain valid for the lifetime of the
+ *                      proxy.
  */
-void hrpc_proxy_free(struct hrpc_proxy *proxy);
+void hrpc_proxy_init(struct hrpc_proxy *proxy,
+            struct hrpc_messenger *msgr,
+            const struct sockaddr_in *remote,
+            const char *protocol, const char *username);
 
 /**
  * Mark the proxy as active.
@@ -160,6 +153,9 @@ void hrpc_proxy_sync_cb(struct hrpc_response *resp, struct hadoop_err *err,
  *
  * This method will return after queuing up the RPC to be sent.
  *
+ * Note: after the proxy has been started, you may __not__ de-allocate the
+ * proxy until the callback has happened.
+ *
  * @param proxy                 The Hadoop proxy to use.  A single proxy can
  *                                  only make one call at once.
  * @param method                The method we're calling.

hadoop-native-core/rpc/reactor.c

@@ -67,11 +67,11 @@ static struct hrpc_conn *reuse_idle_conn(struct hrpc_reactor *reactor,
 
     memset(&exemplar, 0, sizeof(exemplar));
     exemplar.remote = *remote;
-    exemplar.protocol = call->protocol;
+    exemplar.protocol = (char*)call->protocol;
     conn = RB_NFIND(hrpc_conns, &reactor->conns, &exemplar);
     if (!conn)
         return NULL;
-    if (hrpc_conn_usable(conn, remote, call->protocol)) {
+    if (hrpc_conn_usable(conn, remote, call->protocol, call->username)) {
         if (conn->writer.state == HRPC_CONN_WRITE_IDLE) {
             RB_REMOVE(hrpc_conns, &reactor->conns, conn);
             return conn;