Add UDP transport support to the NFS server.

[bluesky.git] / nfs3 / rpc.c

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);
 }