From 3e65ecb93d62d628b99d68fec8b1ec94993f4f09 Mon Sep 17 00:00:00 2001 From: Joseph Henry Date: Mon, 11 Jan 2016 10:12:59 -0800 Subject: Stateless RPC rework --- netcon/RPC.c | 114 +++++++++++++++++++++++++++++++++-------------------------- 1 file changed, 64 insertions(+), 50 deletions(-) (limited to 'netcon/RPC.c') 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 #include +#include +#include + #include #include #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(×tamp)); + + 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], ×tring, 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 { -- cgit v1.2.3