X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=nfs3%2Frpc.c;h=76de1cd86663ddb2bc3a922ec2abc93d0db2f464;hb=fb15c75f3c8eeb3d7b277795ced44bb4928273a1;hp=a01408fd571d16a2bff1265bd383111e4d498c27;hpb=e6ca13c07e0cb5dd05d48a12cd50a2cf6179ce1a;p=bluesky.git

diff --git a/nfs3/rpc.c b/nfs3/rpc.c
index a01408f..76de1cd 100644
--- a/nfs3/rpc.c
+++ b/nfs3/rpc.c
@@ -112,6 +112,7 @@ struct rpc_reply {
 
 static void async_rpc_write(RPCConnection *rpc,
                             const char *buf, gsize len);
+static void async_rpc_flush(RPCConnection *rpc);
 
 struct rpc_fail_reply {
     uint32_t xid;
@@ -122,68 +123,6 @@ struct rpc_fail_reply {
     uint32_t accept_stat;
 };
 
-/* Routines for XDR-encoding to a growable string. */
-static bool_t xdr_string_putlong(XDR *xdrs, const long *lp)
-{
-    GString *str = (GString *)xdrs->x_private;
-    uint32_t data = htonl(*lp);
-    g_string_set_size(str, str->len + 4);
-    memcpy(str->str + str->len - 4, &data, 4);
-    return TRUE;
-}
-
-static bool_t xdr_string_putbytes(XDR *xdrs, const char *addr, u_int len)
-{
-    GString *str = (GString *)xdrs->x_private;
-    g_string_set_size(str, str->len + len);
-    memcpy(str->str + str->len - len, addr, len);
-    return TRUE;
-}
-
-static u_int xdr_string_getpos(const XDR *xdrs)
-{
-    GString *str = (GString *)xdrs->x_private;
-    return str->len;
-}
-
-static bool_t xdr_string_putint32(XDR *xdrs, const int32_t *ip)
-{
-    GString *str = (GString *)xdrs->x_private;
-    uint32_t data = htonl(*ip);
-    g_string_set_size(str, str->len + 4);
-    memcpy(str->str + str->len - 4, &data, 4);
-    return TRUE;
-}
-
-static int32_t *xdr_string_inline(XDR *xdrs, u_int len)
-{
-    GString *str = (GString *)xdrs->x_private;
-    g_string_set_size(str, str->len + len);
-    return (int32_t *)(str->str + str->len - len);
-}
-
-static void xdr_string_destroy(XDR *xdrs)
-{
-}
-
-static struct xdr_ops xdr_string_ops = {
-    .x_putlong = xdr_string_putlong,
-    .x_putbytes = xdr_string_putbytes,
-    .x_getpostn = xdr_string_getpos,
-    .x_putint32 = xdr_string_putint32,
-    .x_inline = xdr_string_inline,
-    .x_destroy = xdr_string_destroy,
-};
-
-static void xdr_string_create(XDR *xdrs, GString *string, enum xdr_op op)
-{
-    xdrs->x_op = op;
-    xdrs->x_ops = &xdr_string_ops;
-    xdrs->x_private = (char *)string;
-    xdrs->x_base = NULL;
-    xdrs->x_handy = 0;
-}
-
 static void
 async_rpc_send_failure(RPCRequest *req, enum accept_stat stat)
 {
@@ -198,9 +137,11 @@ async_rpc_send_failure(RPCRequest *req, enum accept_stat stat)
 
     g_mutex_lock(req->connection->send_lock);
     uint32_t fragment = htonl(sizeof(header) | 0x80000000);
-    async_rpc_write(req->connection, (const char *)&fragment, sizeof(fragment));
+    if (!req->connection->udp_transport)
+        async_rpc_write(req->connection, (const char *)&fragment,
+                        sizeof(fragment));
     async_rpc_write(req->connection, (const char *)&header, sizeof(header));
-    g_io_channel_flush(req->connection->channel, NULL);
+    async_rpc_flush(req->connection);
     g_mutex_unlock(req->connection->send_lock);
 
     if (req->args != NULL) {
@@ -223,6 +164,14 @@ async_rpc_send_failure(RPCRequest *req, enum accept_stat stat)
         g_free(c);
     }
 
+    if (req->connection->udp_transport) {
+        /* For UDP, a connection only exists for the duration of a single
+         * message. */
+        g_mutex_free(req->connection->send_lock);
+        g_string_free(req->connection->sendbuf, TRUE);
+        g_free(req->connection);
+    }
+
     g_free(req);
 }
 
@@ -251,10 +200,12 @@ async_rpc_send_reply(RPCRequest *req, void *result)
     g_mutex_lock(req->connection->send_lock);
     gsize msg_size = str->len;
     uint32_t fragment = htonl((msg_size + sizeof(header)) | 0x80000000);
-    async_rpc_write(req->connection, (const char *)&fragment, sizeof(fragment));
+    if (!req->connection->udp_transport)
+        async_rpc_write(req->connection, (const char *)&fragment,
+                        sizeof(fragment));
     async_rpc_write(req->connection, (const char *)&header, sizeof(header));
     async_rpc_write(req->connection, str->str, str->len);
-    g_io_channel_flush(req->connection->channel, NULL);
+    async_rpc_flush(req->connection);
     g_mutex_unlock(req->connection->send_lock);
 
     time_end = bluesky_now_hires();
@@ -285,6 +236,14 @@ async_rpc_send_reply(RPCRequest *req, void *result)
         g_free(c);
     }
 
+    if (req->connection->udp_transport) {
+        /* For UDP, a connection only exists for the duration of a single
+         * message. */
+        g_mutex_free(req->connection->send_lock);
+        g_string_free(req->connection->sendbuf, TRUE);
+        g_free(req->connection);
+    }
+
     g_free(req);
 }
 
@@ -549,7 +508,8 @@ static async_rpc_init()
     main_context = g_main_context_new();
     main_loop = g_main_loop_new(main_context, FALSE);
 
-    rpc_thread_pool = g_thread_pool_new(async_rpc_task, NULL, -1, FALSE, NULL);
+    rpc_thread_pool = g_thread_pool_new(async_rpc_task, NULL,
+                                        bluesky_max_threads, FALSE, NULL);
 
     /* Arrange to have the cache writeback code run every five seconds. */
     GSource *source = g_timeout_source_new_seconds(5);
@@ -659,6 +619,12 @@ static gboolean async_rpc_dispatch(RPCConnection *rpc)
 static void async_rpc_write(RPCConnection *rpc,
                             const char *buf, gsize len)
 {
+    if (rpc->udp_transport) {
+        g_string_append_len(rpc->sendbuf, buf, len);
+        return;
+    }
+
+    /* Normal TCP path */
     while (len > 0) {
         gsize written = 0;
         switch (g_io_channel_write_chars(rpc->channel, buf, len,
@@ -678,6 +644,19 @@ static void async_rpc_write(RPCConnection *rpc,
     // g_io_channel_flush(rpc->channel, NULL);
 }
 
+/* Flush a completed message out to the RPC socket */
+static void async_rpc_flush(RPCConnection *rpc)
+{
+    if (rpc->udp_transport) {
+        sendto(g_io_channel_unix_get_fd(rpc->channel),
+               rpc->sendbuf->str, rpc->sendbuf->len, 0,
+               (struct sockaddr *)&rpc->peer, sizeof(struct sockaddr_in));
+        return;
+    } else {
+        g_io_channel_flush(rpc->channel, NULL);
+    }
+}
+
 static gboolean async_rpc_do_read(GIOChannel *channel,
                                   GIOCondition condition,
                                   gpointer data)
@@ -746,7 +725,7 @@ static gboolean async_rpc_do_read(GIOChannel *channel,
         /* We were reading in the fragment body. */
         rpc->frag_len -= bytes_read;
 
-        if (rpc->frag_len = 0x80000000) {
+        if (rpc->frag_len == 0x80000000) {
             /* We have a complete message since this was the last fragment and
              * there are no more bytes in it.  Dispatch the message. */
             if (!async_rpc_dispatch(rpc)) {
@@ -802,6 +781,50 @@ static async_rpc_register_listening(int fd)
     g_source_unref(source);
 }
 
+static gboolean async_rpc_do_udp(GIOChannel *channel,
+                                 GIOCondition condition,
+                                 gpointer data)
+{
+    char buf[65536];
+
+    struct sockaddr_in src;
+    socklen_t addrlen = sizeof(struct sockaddr_in);
+    ssize_t len = recvfrom(g_io_channel_unix_get_fd(channel),
+                           buf, sizeof(buf), 0,
+                           (struct sockaddr *)&src, &addrlen);
+    if (len < 0) {
+        fprintf(stderr, "UDP read error: %m, shutting down UDP\n");
+        return FALSE;
+    }
+
+    g_assert(len < sizeof(buf));
+
+    RPCConnection *rpc = g_new0(RPCConnection, 1);
+    rpc->channel = channel;
+    rpc->msgbuf = g_string_new_len(buf, len);
+    rpc->send_lock = g_mutex_new();
+    rpc->udp_transport = TRUE;
+    memcpy(&rpc->peer, &src, sizeof(struct sockaddr_in));
+    rpc->sendbuf = g_string_new("");
+
+    /* We have a complete message since this was the last fragment and
+     * there are no more bytes in it.  Dispatch the message. */
+    async_rpc_dispatch(rpc);
+
+    return TRUE;
+}
+
+static async_rpc_register_listening_udp(int fd)
+{
+    GIOChannel *channel = g_io_channel_unix_new(fd);
+    g_io_channel_set_encoding(channel, NULL, NULL);
+    GSource *source = g_io_create_watch(channel, G_IO_IN);
+    g_source_set_callback(source, (GSourceFunc)async_rpc_do_udp,
+                          NULL, NULL);
+    g_source_attach(source, main_context);
+    g_source_unref(source);
+}
+
 static gpointer async_rpc_run(gpointer data)
 {
     g_print("Starting NFS main loop...\n");
@@ -870,5 +893,27 @@ void register_rpc()
 
     async_rpc_register_listening(fd);
 
+    /* Minimal UDP NFSv3 support */
+    fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    if (fd < 0) {
+        fprintf(stderr, "Unable to create NFS UDP socket: %m\n");
+        exit(1);
+    }
+
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(NFS_SERVICE_PORT);
+    addr.sin_addr.s_addr = INADDR_ANY;
+    if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
+        fprintf(stderr, "Unable to bind to NFS UDP address: %m\n");
+        exit(1);
+    }
+
+    if (!pmap_set(NFS_PROGRAM, NFS_V3, IPPROTO_UDP, NFS_SERVICE_PORT)) {
+        fprintf(stderr, "Could not register NFS UDP RPC service!\n");
+        exit(1);
+    }
+
+    async_rpc_register_listening_udp(fd);
+
     g_thread_create(async_rpc_run, NULL, TRUE, NULL);
 }