Stateless RPC rework

1 files changed, 64 insertions, 50 deletions

diff --git a/netcon/RPC.c b/netcon/RPC.c
index 2b396183..5433bb12 100644
--- a/netcon/RPC.c
+++ b/netcon/RPC.c
@@ -5,6 +5,9 @@
 #include <errno.h>
 #include <sys/syscall.h>
 
+#include <fcntl.h>
+#include <stdint.h>
+
 #include <sys/socket.h>
 #include <strings.h>
 #include "RPC.h"
@@ -14,18 +17,32 @@
 
 static int instance_count;
 static int rpc_count;
-static pthread_mutex_t lock;
 
+static pthread_mutex_t lock;
 void rpc_mutex_init() {
   if(pthread_mutex_init(&lock, NULL) != 0) {
     fprintf(stderr, "error while initializing service call mutex\n");
   }
 }
-
 void rpc_mutex_destroy() {
   pthread_mutex_destroy(&lock);
 }
 
+/* 
+ * Reads a new file descriptor from the service 
+ */
+int get_new_fd(int sock)
+{
+  char buf[BUF_SZ];
+  int newfd;
+  ssize_t size = sock_fd_read(sock, buf, sizeof(buf), &newfd);
+  if(size > 0){
+    return newfd;
+  }
+  fprintf(stderr, "get_new_fd(): Error, unable to read fd over (%d)\n", sock);
+  return -1;
+}
+
 /*
  * Reads a return value from the service and sets errno (if applicable)
  */
@@ -46,21 +63,6 @@ int get_retval(int rpc_sock)
   return -1;
 }
 
-/* 
- * Reads a new file descriptor from the service 
- */
-int get_new_fd(int sock)
-{
-  char buf[BUF_SZ];
-  int newfd;
-  ssize_t size = sock_fd_read(sock, buf, sizeof(buf), &newfd);
-  if(size > 0){
-    return newfd;
-  }
-  fprintf(stderr, "get_new_fd(): Error, unable to read fd over (%d)\n", sock);
-  return -1;
-}
-
 int rpc_join(const char * sockname)
 {
 	struct sockaddr_un addr;
@@ -81,9 +83,9 @@ int rpc_join(const char * sockname)
 			sleep(1);
 		}
 		else {
-			int newfd = dup2(sock, RPC_FD-instance_count);
-			close(sock);
-			return newfd;
+			//int newfd = dup2(sock, RPC_FD-instance_count);
+			//close(sock);
+			return sock;
 		}
 		attempts++;
 	}
@@ -93,14 +95,26 @@ int rpc_join(const char * sockname)
 /*
  * Send a command to the service 
  */
-int rpc_send_command(int cmd, int rpc_sock, void *data, int len)
+int rpc_send_command(int cmd, int forfd, void *data, int len)
 {
-  char cmdbuf[BUF_SZ];
-  cmdbuf[0] = cmd;
-  memcpy(&cmdbuf[1], data, len);
-
   pthread_mutex_lock(&lock);
-  char metabuf[BUF_SZ]; // portion of buffer which contains RPC metadata for debugging
+  char padding[] = {0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89};
+  char cmdbuf[BUF_SZ], magic[TOKEN_SIZE], metabuf[BUF_SZ];
+  memcpy(magic+MAGIC_SIZE, padding, TOKEN_SIZE);
+  uint64_t magic_num;
+
+  // ephemeral RPC socket used only for this command
+  int rpc_sock = rpc_join("/root/dev/ztest/nc_e5cd7a9e1c3511dd");
+  // Generate token
+  int fdrand = open("/dev/urandom", O_RDONLY);
+  read(fdrand, &magic, MAGIC_SIZE);
+  memcpy(&magic_num, magic, MAGIC_SIZE);  
+  cmdbuf[CMD_ID_IDX] = cmd;
+  memcpy(&cmdbuf[MAGIC_IDX], &magic_num, MAGIC_SIZE);
+  memcpy(&cmdbuf[STRUCT_IDX], data, len);
+
+  // Format: [sig_byte] + [cmd_id] + [magic] + [meta] + [payload]
+
 #ifdef VERBOSE
   /*
   #define IDX_PID       0
@@ -119,43 +133,52 @@ int rpc_send_command(int cmd, int rpc_sock, void *data, int len)
   time_t timestamp;
   timestamp = time(NULL);
   strftime(timestring, sizeof(timestring), "%H:%M:%S", localtime(&timestamp));
+
+  metabuf[IDX_SIGNAL_BYTE] = 'R';
+  
   memcpy(&metabuf[IDX_PID],     &pid,         sizeof(pid_t)      ); /* pid       */
   memcpy(&metabuf[IDX_TID],     &tid,         sizeof(pid_t)      ); /* tid       */
   memcpy(&metabuf[IDX_COUNT],   &rpc_count,   sizeof(rpc_count)  ); /* rpc_count */
   memcpy(&metabuf[IDX_TIME],    &timestring,   20                ); /* timestamp */
 #endif
   /* Combine command flag+payload with RPC metadata */
-  memcpy(&metabuf[IDX_PAYLOAD], cmdbuf, len);
+  memcpy(&metabuf[IDX_PAYLOAD], cmdbuf, len + 1 + MAGIC_SIZE);
+  
+  // Write RPC
   int n_write = write(rpc_sock, &metabuf, BUF_SZ);
   if(n_write < 0) {
-    fprintf(stderr, "Error writing command to service (CMD = %d)\n", cmdbuf[0]);
+    fprintf(stderr, "Error writing command to service (CMD = %d)\n", cmdbuf[CMD_ID_IDX]);
     errno = 0;
   }
-  
+  // Write token to corresponding data stream
+  if(n_write > 0 && forfd > -1){
+    usleep(5000);
+    int w = send(forfd, &magic, TOKEN_SIZE, 0);
+  }
+  // Process response from service
   int ret = ERR_OK;
   if(n_write > 0) {
-    if(cmdbuf[0]==RPC_SOCKET) {
-    	ret = get_new_fd(rpc_sock);
+    if(cmdbuf[CMD_ID_IDX]==RPC_SOCKET) {
+      pthread_mutex_unlock(&lock);
+      return rpc_sock; // Used as new socket
     }
-    if(cmdbuf[0]==RPC_MAP_REQ
-      || cmdbuf[0]==RPC_CONNECT
-      || cmdbuf[0]==RPC_BIND
-      || cmdbuf[0]==RPC_LISTEN
-      || cmdbuf[0]==RPC_MAP) {
+    if(cmdbuf[CMD_ID_IDX]==RPC_CONNECT
+      || cmdbuf[CMD_ID_IDX]==RPC_BIND
+      || cmdbuf[CMD_ID_IDX]==RPC_LISTEN) {
     	ret = get_retval(rpc_sock);
     }
-    if(cmdbuf[0]==RPC_GETSOCKNAME) {
-      ret = n_write;
+    if(cmdbuf[CMD_ID_IDX]==RPC_GETSOCKNAME) {
+      pthread_mutex_unlock(&lock);
+      return rpc_sock; // Don't close rpc here, we'll use it to read getsockopt_st
     }
   }
-  else {
+  else
     ret = -1;
-  }
+  close(rpc_sock); // We're done with this RPC socket, close it (if type-R)
   pthread_mutex_unlock(&lock);
   return ret;
 }
 
-
 /* 
  * Send file descriptor 
  */
@@ -166,21 +189,17 @@ ssize_t sock_fd_write(int sock, int fd)
   struct iovec iov;
   char buf = '\0';
   int buflen = 1;
-
   union {
         struct cmsghdr  cmsghdr;
     char control[CMSG_SPACE(sizeof (int))];
   } cmsgu;
   struct cmsghdr *cmsg;
-
   iov.iov_base = &buf;
   iov.iov_len = buflen;
-
   msg.msg_name = NULL;
   msg.msg_namelen = 0;
   msg.msg_iov = &iov;
   msg.msg_iovlen = 1;
-
   if (fd != -1) {
       msg.msg_control = cmsgu.control;
       msg.msg_controllen = sizeof(cmsgu.control);
@@ -193,13 +212,11 @@ ssize_t sock_fd_write(int sock, int fd)
       msg.msg_control = NULL;
       msg.msg_controllen = 0;
   }
-
   size = sendmsg(sock, &msg, 0);
   if (size < 0)
       perror ("sendmsg");
   return size;
 }
-
 /* 
  * Read a file descriptor 
  */
@@ -214,10 +231,8 @@ ssize_t sock_fd_read(int sock, void *buf, ssize_t bufsize, int *fd)
       char control[CMSG_SPACE(sizeof (int))];
     } cmsgu;
     struct cmsghdr *cmsg;
-
     iov.iov_base = buf;
     iov.iov_len = bufsize;
-
     msg.msg_name = NULL;
     msg.msg_namelen = 0;
     msg.msg_iov = &iov;
@@ -239,7 +254,6 @@ ssize_t sock_fd_read(int sock, void *buf, ssize_t bufsize, int *fd)
           fprintf (stderr, "invalid cmsg_type %d\n",cmsg->cmsg_type);
           return -1;
       }
-
       *fd = *((int *) CMSG_DATA(cmsg));
     } else *fd = -1;
   } else {