Add UDP transport support to the NFS server.

SpecSFS seems to require basic UDP functionality.

diff --git a/microbench/run.sh b/microbench/run.sh
index 8b0856d..28571ef 100755 (executable)
--- a/microbench/run.sh
+++ b/microbench/run.sh
@@ -69,6 +69,7 @@ for s in $SIZES; do
     run_cleanup
 done
 
+if false; then
 export BLUESKY_OPT_SYNC_FRONTENDS=0
 for t in 1 2 4 6 8 10 12 14 16 20 24 28 32 40 48 56 64; do
     run_thread_test $t
@@ -78,5 +79,6 @@ export BLUESKY_OPT_SYNC_FRONTENDS=1
 for t in 1 2 4 6 8 10 12 14 16 20 24 28 32 40 48 56 64; do
     run_thread_test $t
 done
+fi
 
 benchmark_cleanup

diff --git a/nfs3/nfs3_prot.h b/nfs3/nfs3_prot.h
index 77fe5ad..1b8352b 100644 (file)
--- a/nfs3/nfs3_prot.h
+++ b/nfs3/nfs3_prot.h
@@ -657,6 +657,16 @@ typedef struct {
     /* Mutex protecting send operations on the socket (to ensure that replies
      * are not accidentally interleaved). */
     GMutex *send_lock;
+
+    /* Is this a UDP connection? */
+    gboolean udp_transport;
+
+    /* For UDP connections only, the address of the sender. */
+    struct sockaddr_in peer;
+
+    /* For UDP only, a buffer for accumulating the full contents of a message
+     * before it is sent */
+    GString *sendbuf;
 } RPCConnection;
 
 /* Linked list of cleanup functions to call when a request is completed. */

diff --git a/nfs3/rpc.c b/nfs3/rpc.c
index 40cf1b0..e1cb4d1 100644 (file)
--- 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;
@@ -136,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) {
@@ -161,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);
 }
 
@@ -189,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();
@@ -223,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);
 }
 
@@ -597,6 +618,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,
@@ -616,6 +643,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)
@@ -740,6 +780,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");
@@ -808,5 +892,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);
 }