From cb08ad76d3a1c1887173c8e2521f60ce120a3088 Mon Sep 17 00:00:00 2001 From: David Sheets Date: Tue, 12 Apr 2016 16:04:20 +0100 Subject: [PATCH] transfused: use vsock transport instead of 9p Signed-off-by: David Sheets --- alpine/packages/diagnostics/diagnostics | 7 +- alpine/packages/transfused/Makefile | 4 +- .../packages/transfused/etc/init.d/transfused | 4 - .../include/uapi/linux/vm_sockets.h | 161 ++++++ alpine/packages/transfused/transfused.c | 517 ++++++++++-------- alpine/packages/transfused/transfused.h | 10 +- alpine/packages/transfused/transfused_log.c | 4 +- alpine/packages/transfused/transfused_log.h | 2 + alpine/packages/transfused/transfused_vsock.c | 99 ++++ alpine/packages/transfused/transfused_vsock.h | 2 + 10 files changed, 555 insertions(+), 255 deletions(-) create mode 100644 alpine/packages/transfused/include/uapi/linux/vm_sockets.h create mode 100644 alpine/packages/transfused/transfused_vsock.c create mode 100644 alpine/packages/transfused/transfused_vsock.h diff --git a/alpine/packages/diagnostics/diagnostics b/alpine/packages/diagnostics/diagnostics index cee2ee736..6e1bb4f66 100755 --- a/alpine/packages/diagnostics/diagnostics +++ b/alpine/packages/diagnostics/diagnostics @@ -7,11 +7,8 @@ DEV=$(mount | grep '/dev/[sxv]da. on /var type') [ $? -eq 0 ] && printf "✓ Drive mounted: $DEV\n" || printf "✗ No drive mounted\n" INET=$(ifconfig eth0 2> /dev/null | grep 'inet addr') [ $? -eq 0 ] && printf "✓ Network connected: $INET\n" || printf "✗ No network connection\n" -if [ -d /Transfuse ] -then - FUSE=$(ps -eo args | grep '^/sbin/transfused') - [ $? -eq 0 ] && printf "✓ Process transfused running\n" || printf "✗ No transfused process\n" -fi +FUSE=$(ps -eo args | grep '^/sbin/transfused') +[ $? -eq 0 ] && printf "✓ Process transfused running\n" || printf "✗ No transfused process\n" MDNS=$(ps -eo args | grep '^/sbin/mdnstool') [ $? -eq 0 ] && printf "✓ Process mdnstool running: $MDNS\n" || printf "✗ No mdnstool process\n" HUPPER=$(ps -eo args | grep '^/bin/hupper') diff --git a/alpine/packages/transfused/Makefile b/alpine/packages/transfused/Makefile index 978566e9c..f2e1fbf35 100644 --- a/alpine/packages/transfused/Makefile +++ b/alpine/packages/transfused/Makefile @@ -1,7 +1,7 @@ .PHONY: all -HDR=transfused.h transfused_log.h -SRC=transfused.c transfused_log.c +HDR=transfused.h transfused_log.h transfused_vsock.h +SRC=transfused.c transfused_log.c transfused_vsock.c DEPS=$(HDR) $(SRC) diff --git a/alpine/packages/transfused/etc/init.d/transfused b/alpine/packages/transfused/etc/init.d/transfused index 3b5ae5017..7e915b051 100755 --- a/alpine/packages/transfused/etc/init.d/transfused +++ b/alpine/packages/transfused/etc/init.d/transfused @@ -6,9 +6,7 @@ start() { ebegin "Starting FUSE socket passthrough" - mkdir -p /Transfuse mkdir -p /Mac - mount -t 9p -o trans=virtio,dfltuid=1001,dfltgid=50,version=9p2000,msize=32768 fuse /Transfuse if cat /proc/cmdline | grep -q 'com.docker.driverDir' then @@ -38,8 +36,6 @@ start() stop() { - [ -d /Transfuse ] || exit 0 - ebegin "Stopping FUSE socket passthrough" PIDFILE=/var/run/transfused.pid diff --git a/alpine/packages/transfused/include/uapi/linux/vm_sockets.h b/alpine/packages/transfused/include/uapi/linux/vm_sockets.h new file mode 100644 index 000000000..41934a185 --- /dev/null +++ b/alpine/packages/transfused/include/uapi/linux/vm_sockets.h @@ -0,0 +1,161 @@ +/* + * VMware vSockets Driver + * + * Copyright (C) 2007-2013 VMware, Inc. All rights reserved. + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the Free + * Software Foundation version 2 and no later version. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for + * more details. + */ + +#ifndef _UAPI_VM_SOCKETS_H +#define _UAPI_VM_SOCKETS_H + +#ifdef __KERNEL__ +#include +#else +#define __kernel_sa_family_t sa_family_t +#include +#endif + +/* Option name for STREAM socket buffer size. Use as the option name in + * setsockopt(3) or getsockopt(3) to set or get an unsigned long long that + * specifies the size of the buffer underlying a vSockets STREAM socket. + * Value is clamped to the MIN and MAX. + */ + +#define SO_VM_SOCKETS_BUFFER_SIZE 0 + +/* Option name for STREAM socket minimum buffer size. Use as the option name + * in setsockopt(3) or getsockopt(3) to set or get an unsigned long long that + * specifies the minimum size allowed for the buffer underlying a vSockets + * STREAM socket. + */ + +#define SO_VM_SOCKETS_BUFFER_MIN_SIZE 1 + +/* Option name for STREAM socket maximum buffer size. Use as the option name + * in setsockopt(3) or getsockopt(3) to set or get an unsigned long long + * that specifies the maximum size allowed for the buffer underlying a + * vSockets STREAM socket. + */ + +#define SO_VM_SOCKETS_BUFFER_MAX_SIZE 2 + +/* Option name for socket peer's host-specific VM ID. Use as the option name + * in getsockopt(3) to get a host-specific identifier for the peer endpoint's + * VM. The identifier is a signed integer. + * Only available for hypervisor endpoints. + */ + +#define SO_VM_SOCKETS_PEER_HOST_VM_ID 3 + +/* Option name for determining if a socket is trusted. Use as the option name + * in getsockopt(3) to determine if a socket is trusted. The value is a + * signed integer. + */ + +#define SO_VM_SOCKETS_TRUSTED 5 + +/* Option name for STREAM socket connection timeout. Use as the option name + * in setsockopt(3) or getsockopt(3) to set or get the connection + * timeout for a STREAM socket. + */ + +#define SO_VM_SOCKETS_CONNECT_TIMEOUT 6 + +/* Option name for using non-blocking send/receive. Use as the option name + * for setsockopt(3) or getsockopt(3) to set or get the non-blocking + * transmit/receive flag for a STREAM socket. This flag determines whether + * send() and recv() can be called in non-blocking contexts for the given + * socket. The value is a signed integer. + * + * This option is only relevant to kernel endpoints, where descheduling the + * thread of execution is not allowed, for example, while holding a spinlock. + * It is not to be confused with conventional non-blocking socket operations. + * + * Only available for hypervisor endpoints. + */ + +#define SO_VM_SOCKETS_NONBLOCK_TXRX 7 + +/* The vSocket equivalent of INADDR_ANY. This works for the svm_cid field of + * sockaddr_vm and indicates the context ID of the current endpoint. + */ + +#define VMADDR_CID_ANY -1U + +/* Bind to any available port. Works for the svm_port field of + * sockaddr_vm. + */ + +#define VMADDR_PORT_ANY -1U + +/* Use this as the destination CID in an address when referring to the + * hypervisor. VMCI relies on it being 0, but this would be useful for other + * transports too. + */ + +#define VMADDR_CID_HYPERVISOR 0 + +/* This CID is specific to VMCI and can be considered reserved (even VMCI + * doesn't use it anymore, it's a legacy value from an older release). + */ + +#define VMADDR_CID_RESERVED 1 + +/* Use this as the destination CID in an address when referring to the host + * (any process other than the hypervisor). VMCI relies on it being 2, but + * this would be useful for other transports too. + */ + +#define VMADDR_CID_HOST 2 + +/* Invalid vSockets version. */ + +#define VM_SOCKETS_INVALID_VERSION -1U + +/* The epoch (first) component of the vSockets version. A single byte + * representing the epoch component of the vSockets version. + */ + +#define VM_SOCKETS_VERSION_EPOCH(_v) (((_v) & 0xFF000000) >> 24) + +/* The major (second) component of the vSockets version. A single byte + * representing the major component of the vSockets version. Typically + * changes for every major release of a product. + */ + +#define VM_SOCKETS_VERSION_MAJOR(_v) (((_v) & 0x00FF0000) >> 16) + +/* The minor (third) component of the vSockets version. Two bytes representing + * the minor component of the vSockets version. + */ + +#define VM_SOCKETS_VERSION_MINOR(_v) (((_v) & 0x0000FFFF)) + +/* Address structure for vSockets. The address family should be set to + * AF_VSOCK. The structure members should all align on their natural + * boundaries without resorting to compiler packing directives. The total size + * of this structure should be exactly the same as that of struct sockaddr. + */ + +struct sockaddr_vm { + __kernel_sa_family_t svm_family; + unsigned short svm_reserved1; + unsigned int svm_port; + unsigned int svm_cid; + unsigned char svm_zero[sizeof(struct sockaddr) - + sizeof(sa_family_t) - + sizeof(unsigned short) - + sizeof(unsigned int) - sizeof(unsigned int)]; +}; + +#define IOCTL_VM_SOCKETS_GET_LOCAL_CID _IO(7, 0xb9) + +#endif /* _UAPI_VM_SOCKETS_H */ diff --git a/alpine/packages/transfused/transfused.c b/alpine/packages/transfused/transfused.c index f31abf669..cc6b55c7b 100644 --- a/alpine/packages/transfused/transfused.c +++ b/alpine/packages/transfused/transfused.c @@ -21,16 +21,15 @@ #include #include "transfused_log.h" +#include "transfused_vsock.h" -#define COPY_BUFSZ 65536 -// The Linux 9p driver/xhyve virtio-9p device have bugs in the -// zero-copy code path which is triggered by I/O of more than 1024 -// bytes. For an unknown reason, these defects are unusually prominent -// in the event channel use pattern. -#define EVENT_BUFSZ 1024 +#define IN_BUFSZ ((1 << 20) + 16) +#define OUT_BUFSZ ((1 << 20) + 64) +#define EVENT_BUFSZ 4096 #define DEFAULT_FUSERMOUNT "/bin/fusermount" -#define DEFAULT_SOCKET9P_ROOT "/Transfuse" +#define DEFAULT_SOCKET "v:_:1525" +#define DEFAULT_SERVER "v:2:1524" #define RMDIR_SYSCALL 0 #define UNLINK_SYSCALL 1 @@ -41,12 +40,14 @@ // these could be turned into an enum probably but... C standard nausea char * default_fusermount = DEFAULT_FUSERMOUNT; -char * default_socket9p_root = DEFAULT_SOCKET9P_ROOT; +char * default_socket = DEFAULT_SOCKET; +char * default_server = DEFAULT_SERVER; char * usage = - "usage: transfused [-p pidfile] [-9 socket9p_root] [-f fusermount]\n" + "usage: transfused [-p pidfile] [-d server] [-s socket] [-f fusermount]\n" " [-l logfile] [-m mount_trigger] [-t triggerlog]\n" " -p pidfile\tthe path at which to write the pid of the process\n" - " -9 " DEFAULT_SOCKET9P_ROOT "\tthe root of the 9p socket file system\n" + " -d " DEFAULT_SERVER "\tthe server address to use ('v:addr:port')\n" + " -s " DEFAULT_SOCKET "\tthe socket address to use ('v:addr:port')\n" " -f " DEFAULT_FUSERMOUNT "\tthe fusermount executable to use\n" " -l logfile\tthe log file to use before the mount trigger\n" " -m mount_trigger\tthe mountpoint to use to trigger log switchover\n" @@ -58,7 +59,6 @@ pthread_attr_t detached; typedef struct { char * descr; - long connection; char * tag; int from; int to; @@ -101,24 +101,55 @@ void unlock(char *const descr, pthread_mutex_t * mutex) { die(1, "", "unlock %s: ", descr); } +int bind_socket(const char * socket) { + int sock; + + if (socket[0] == 0) + die(2, NULL, "Socket family required"); + if (socket[1] != ':') + die(2, NULL, "Socket address required"); + + switch (socket[0]) { + case 'v': + sock = bind_vsock(socket + 2); + break; + default: + die(2, NULL, "Unknown socket family '%c'", socket[0]); + } + + return sock; +} + +int connect_socket(const char * socket) { + int sock; + + if (socket[0] == 0) + die(2, NULL, "Socket family required"); + if (socket[1] != ':') + die(2, NULL, "Scoket address required"); + + switch (socket[0]) { + case 'v': + sock = connect_vsock(socket + 2); + break; + default: + die(2, NULL, "Unknown socket family '%c'", socket[0]); + } + + return sock; +} + char ** read_opts(connection_t * connection, char * buf) { - int read_fd; - char * read_path; int read_count; int optc = 1; char ** optv; + size_t mount_len; - if (asprintf(&read_path, "%s/connections/%ld/read", - connection->params->socket9p_root, connection->id) == -1) - die(1, "Couldn't allocate read path", ""); - - read_fd = open(read_path, O_RDONLY); - if (read_fd == -1) - die(1, "couldn't open read path", "For connection %ld, ", connection->id); - - read_count = read(read_fd, buf, COPY_BUFSZ - 1); - if (read_count == -1) die(1, "read_opts error reading", ""); + // TODO: deal with socket read conditions e.g. EAGAIN + read_count = read(connection->sock, buf, EVENT_BUFSZ - 1); + if (read_count < 0) die(1, "read_opts error reading", ""); + // TODO: protocol should deal with short read buf[read_count] = 0x0; for (int i = 0; i < read_count; i++) { @@ -137,7 +168,10 @@ char ** read_opts(connection_t * connection, char * buf) { } } - free(read_path); + mount_len = strnlen(optv[optc - 1], 4096) + 1; + connection->mount_point = must_malloc("mount point string", mount_len); + strncpy(connection->mount_point, optv[optc - 1], mount_len - 1); + connection->mount_point[mount_len - 1] = '\0'; return optv; } @@ -146,48 +180,54 @@ uint64_t message_id(uint64_t * message) { return message[1]; } -void copy(copy_thread_state * copy_state) { +int read_message(char * descr, int fd, char * buf) { + int read_count; + size_t nbyte; + uint32_t len; + + // TODO: socket read conditions e.g. EAGAIN + read_count = read(fd, buf, 4); + if (read_count != 4) { + if (read_count < 0) die(1, "", "copy %s: error reading: ", descr); + if (read_count == 0) die(1, NULL, "copy %s: EOF reading length", descr); + die(1, NULL, "copy %s: short read length %d", descr, read_count); + } + len = *((uint32_t *) buf); + if (len > IN_BUFSZ) + die(1, NULL, "copy %s: message size %d exceeds buffer capacity %d", + len, IN_BUFSZ); + + nbyte = (size_t) (len - 4); + buf += 4; + + do { + // TODO: socket read conditions e.g. EAGAIN + read_count = read(fd, buf, nbyte); + if (read_count < 0) die(1, "", "copy %s: error reading: ", descr); + if (read_count == 0) die(1, NULL, "copy %s: EOF reading", descr); + nbyte -= read_count; + buf += read_count; + } while (nbyte != 0); + + return (int) len; +} + +void copy_into_fuse(copy_thread_state * copy_state) { int from = copy_state->from; int to = copy_state->to; char * descr = copy_state->descr; int read_count, write_count; - long connection = copy_state->connection; - char * tag = copy_state->tag; void * buf; - buf = must_malloc(descr, COPY_BUFSZ); + buf = must_malloc(descr, IN_BUFSZ); while(1) { - read_count = read(from, buf, COPY_BUFSZ); - if (read_count == -1) die(1, "", "copy %s: error reading: ", descr); - - if (debug) { - int trace_fd; - char * trace_path; - - if (asprintf(&trace_path, "/tmp/transfused.%ld.%s.%llu", - connection, tag, message_id(buf)) == -1) - die(1, "Couldn't allocate trace packet path", ""); - - trace_fd = open(trace_path, O_WRONLY | O_CREAT, 0600); - if (trace_fd == -1) - die(1, "couldn't open trace packet path", "For %s, ", descr); - - write_count = write(trace_fd, buf, read_count); - if (write_count == -1) - die(1, "", "copy %s trace: error writing %s: ", descr, trace_path); - - if (write_count != read_count) - die(1, NULL, "copy %s trace: read %d but only wrote %d", - descr, read_count, write_count); - - close(trace_fd); - free(trace_path); - } + read_count = read_message(descr, from, (char *) buf); write_count = write(to, buf, read_count); - if (write_count == -1) die(1, "", "copy %s: error writing: ", descr); + if (write_count < 0) die(1, "", "copy %s: error writing: ", descr); + // /dev/fuse accepts only complete writes if (write_count != read_count) die(1, NULL, "copy %s: read %d but only wrote %d", descr, read_count, write_count); @@ -196,8 +236,42 @@ void copy(copy_thread_state * copy_state) { free(buf); } -void * copy_clean_from(copy_thread_state * copy_state) { - copy(copy_state); +void write_exactly(char * descr, int fd, char * buf, size_t nbyte) { + int write_count; + + do { + // TODO: socket write conditions e.g. EAGAIN + write_count = write(fd, buf, nbyte); + if (write_count < 0) die(1, "", "copy %s: error writing: ", descr); + if (write_count == 0) die(1, "", "copy %s: 0 write: ", descr); + + nbyte -= write_count; + buf += write_count; + } while (nbyte != 0); +} + +void copy_outof_fuse(copy_thread_state * copy_state) { + int from = copy_state->from; + int to = copy_state->to; + char * descr = copy_state->descr; + int read_count; + void * buf; + + buf = must_malloc(descr, OUT_BUFSZ); + + while(1) { + // /dev/fuse only returns complete reads + read_count = read(from, buf, OUT_BUFSZ); + if (read_count < 0) die(1, "", "copy %s: error reading: ", descr); + + write_exactly(descr, to, (char *) buf, read_count); + } + + free(buf); +} + +void * copy_clean_into_fuse(copy_thread_state * copy_state) { + copy_into_fuse(copy_state); close(copy_state->from); @@ -207,12 +281,12 @@ void * copy_clean_from(copy_thread_state * copy_state) { return NULL; } -void * copy_clean_from_thread(void * copy_state) { - return (copy_clean_from((copy_thread_state *) copy_state)); +void * copy_clean_into_fuse_thread(void * copy_state) { + return (copy_clean_into_fuse((copy_thread_state *) copy_state)); } -void * copy_clean_to(copy_thread_state * copy_state) { - copy(copy_state); +void * copy_clean_outof_fuse(copy_thread_state * copy_state) { + copy_outof_fuse(copy_state); close(copy_state->to); @@ -222,8 +296,8 @@ void * copy_clean_to(copy_thread_state * copy_state) { return NULL; } -void * copy_clean_to_thread(void * copy_state) { - return (copy_clean_to((copy_thread_state *) copy_state)); +void * copy_clean_outof_fuse_thread(void * copy_state) { + return (copy_clean_outof_fuse((copy_thread_state *) copy_state)); } int recv_fd(int sock) { @@ -326,57 +400,35 @@ int get_fuse_sock(connection_t * conn, int optc, char *const optv[]) { } void start_reader(connection_t * connection, int fuse) { - int read_fd; - char * read_path; pthread_t child; copy_thread_state * copy_state; - if (asprintf(&read_path, "%s/connections/%ld/read", - connection->params->socket9p_root, connection->id) == -1) - die(1, "Couldn't allocate read path", ""); - - read_fd = open(read_path, O_RDONLY); - if (read_fd == -1) - die(1, "couldn't open read path", "For connection %ld, ", connection->id); - copy_state = (copy_thread_state *) must_malloc("start_reader copy_state", sizeof(copy_thread_state)); - copy_state->descr = read_path; - copy_state->connection = connection->id; + copy_state->descr = connection->mount_point; copy_state->tag = "read"; - copy_state->from = read_fd; + copy_state->from = connection->sock; copy_state->to = fuse; if ((errno = pthread_create(&child, &detached, - copy_clean_from_thread, copy_state))) - die(1, "", "couldn't create read copy thread for connection %ld: ", - connection->id); + copy_clean_into_fuse_thread, copy_state))) + die(1, "", "couldn't create read copy thread for mount %s: ", + connection->mount_point); } void start_writer(connection_t * connection, int fuse) { - int write_fd; - char * write_path; pthread_t child; copy_thread_state * copy_state; - if (asprintf(&write_path, "%s/connections/%ld/write", - connection->params->socket9p_root, connection->id) == -1) - die(1, "Couldn't allocate write path", ""); - - write_fd = open(write_path, O_WRONLY); - if (write_fd == -1) - die(1, "couldn't open write path", "For connection %ld, ", connection->id); - copy_state = (copy_thread_state *) must_malloc("do_write copy_state", sizeof(copy_thread_state)); - copy_state->descr = write_path; - copy_state->connection = connection->id; + copy_state->descr = connection->mount_point; copy_state->tag = "write"; copy_state->from = fuse; - copy_state->to = write_fd; + copy_state->to = connection->sock; if ((errno = pthread_create(&child, &detached, - copy_clean_to_thread, copy_state))) - die(1, "", "Couldn't create write copy thread for connection %ld: ", - connection->id); + copy_clean_outof_fuse_thread, copy_state))) + die(1, "", "Couldn't create write copy thread for mount %s: ", + connection->mount_point); } void * mount_connection(connection_t * conn) { @@ -388,7 +440,7 @@ void * mount_connection(connection_t * conn) { pthread_cond_t copy_halt; int should_halt = 0; - buf = (char *) must_malloc("read_opts packet malloc", COPY_BUFSZ); + buf = (char *) must_malloc("read_opts packet malloc", EVENT_BUFSZ); optv = read_opts(conn, buf); @@ -425,8 +477,8 @@ void * mount_connection(connection_t * conn) { lock("copy lock", ©_lock); while (!should_halt) if ((errno = pthread_cond_wait(©_halt, ©_lock))) - die(1, "", "Couldn't wait for copy halt for connection %ld: ", - conn->id); + die(1, "", "Couldn't wait for copy halt for mount %s: ", + conn->mount_point); unlock("copy lock", ©_lock); free(conn); @@ -439,36 +491,26 @@ void * mount_thread(void * connection) { } void write_pid(connection_t * connection) { - int write_fd; - char * write_path; pid_t pid = gettid(); char * pid_s; int pid_s_len, write_count; - if (asprintf(&write_path, "%s/connections/%ld/write", - connection->params->socket9p_root, connection->id) == -1) - die(1, "Couldn't allocate write path", ""); - - write_fd = open(write_path, O_WRONLY); - if (write_fd == -1) - die(1, "couldn't open write path", "For connection %ld, ", connection->id); - if (asprintf(&pid_s, "%lld", (long long) pid) == -1) die(1, "Couldn't allocate pid string", ""); pid_s_len = strlen(pid_s); - write_count = write(write_fd, pid_s, pid_s_len); - if (write_count == -1) + // TODO: check for socket write conditions e.g. EAGAIN + write_count = write(connection->sock, pid_s, pid_s_len); + if (write_count < 0) die(1, "Error writing pid", ""); + // TODO: handle short writes if (write_count != pid_s_len) die(1, NULL, "Error writing pid %s to socket: only wrote %d bytes", pid_s, write_count); - close(write_fd); free(pid_s); - free(write_path); } void perform_syscall(connection_t * conn, uint8_t syscall, char path[]) { @@ -517,8 +559,6 @@ void perform_syscall(connection_t * conn, uint8_t syscall, char path[]) { } void * event_thread(void * connection_ptr) { - char * read_path; - int read_fd; int read_count, event_len, path_len; void * buf; connection_t * connection = connection_ptr; @@ -527,32 +567,25 @@ void * event_thread(void * connection_ptr) { uint8_t syscall; void * msg; - // This thread registers with the mounted file system as being an + // This thread registers with the file system server as being an // fsnotify event actuator. Other mounted file system interactions // (such as self-logging) SHOULD NOT occur on this thread. write_pid(connection); - if (asprintf(&read_path, "%s/connections/%ld/read", - connection->params->socket9p_root, connection->id) == -1) - die(1, "Couldn't allocate read path", ""); - - read_fd = open(read_path, O_RDONLY); - if (read_fd == -1) - die(1, "couldn't open read path", "For connection %ld, ", connection->id); - buf = must_malloc("incoming event buffer", EVENT_BUFSZ); while(1) { - read_count = read(read_fd, buf, EVENT_BUFSZ); - if (read_count == -1) die(1, "event thread: error reading", ""); + read_count = read(connection->sock, buf, EVENT_BUFSZ); + if (read_count < 0) die(1, "event thread: error reading", ""); event_len = (int) ntohs(*((uint16_t *) buf)); if (debug) - thread_log_time(connection, "read %d bytes from connection %ld\n", - read_count, connection->id); + thread_log_time(connection, "read %d bytes from event connection\n", + read_count); + // TODO: handle short read if (read_count != event_len) { thread_log_time(connection, "event thread: only read %d of %d\n", read_count, event_len); @@ -578,9 +611,67 @@ void * event_thread(void * connection_ptr) { perform_syscall(connection, syscall, path); } - close(read_fd); free(buf); - free(read_path); + // TODO: close connection + return NULL; +} + +void write_pidfile(char * pidfile) { + int fd; + pid_t pid = getpid(); + char * pid_s; + int pid_s_len, write_count; + + if (asprintf(&pid_s, "%lld", (long long) pid) == -1) + die(1, "Couldn't allocate pidfile string", ""); + + pid_s_len = strlen(pid_s); + + fd = open(pidfile, O_WRONLY | O_CREAT | O_TRUNC, 0644); + if (fd == -1) + die(1, "", "Couldn't open pidfile path %s: ", pidfile); + + write_count = write(fd, pid_s, pid_s_len); + if (write_count == -1) + die(1, "", "Error writing pidfile %s: ", pidfile); + + if (write_count != pid_s_len) + die(1, NULL, "Error writing %s to pidfile %s: only wrote %d bytes", + pid_s, pidfile, write_count); + + close(fd); + free(pid_s); +} + +void * init_thread(void * params_ptr) { + parameters * params = params_ptr; + int init_sock = connect_socket(params->server); + int write_count, read_count; + char init_msg[6] = { '\6', '\0', '\0', '\0', '\0', '\0' }; + void * buf; + + // TODO: handle short write/socket conditions + write_count = write(init_sock, init_msg, sizeof(init_msg)); + if (write_count < 0) die(1, "init thread: couldn't write init", ""); + if (write_count != sizeof(init_msg)) + die(1, "init thread: incomplete write", ""); + + buf = must_malloc("incoming init buffer", EVENT_BUFSZ); + + // TODO: handle short read/socket conditions + read_count = read(init_sock, buf, EVENT_BUFSZ); + if (read_count < 0) die(1, "init thread: error reading", ""); + // TODO: handle other messages + if (read_count != 6) die(1, "init thread: response not 6", ""); + for (int i = 0; i < sizeof(init_msg); i++) + if (((char *)buf)[i] != init_msg[i]) + die(1, "init thread: unexpected message", ""); + + // we've gotten Continue so write the pidfile + if (params->pidfile != NULL) + write_pidfile(params->pidfile); + + // TODO: handle more messages return NULL; } @@ -601,7 +692,7 @@ void parse_parameters(int argc, char * argv[], parameters * params) { int errflg = 0; params->pidfile = NULL; - params->socket9p_root = NULL; + params->socket = NULL; params->fusermount = NULL; params->logfile = NULL; params->logfile_fd = 0; @@ -610,15 +701,19 @@ void parse_parameters(int argc, char * argv[], parameters * params) { params->trigger_fd = 0; lock_init("fd_lock", ¶ms->fd_lock, NULL); - while ((c = getopt(argc, argv, ":p:9:f:l:m:t:")) != -1) { + while ((c = getopt(argc, argv, ":p:d:s:f:l:m:t:")) != -1) { switch(c) { case 'p': params->pidfile = optarg; break; - case '9': - params->socket9p_root = optarg; + case 'd': + params->server = optarg; + break; + + case 's': + params->socket = optarg; break; case 'f': @@ -675,15 +770,11 @@ void parse_parameters(int argc, char * argv[], parameters * params) { exit(2); } - if (params->socket9p_root == NULL) - params->socket9p_root = default_socket9p_root; - if (access(params->socket9p_root, X_OK)) { - fprintf(stderr, - "-9 %s path to socket 9p root directory must be executable: ", - params->socket9p_root); - perror(""); - exit(2); - } + if (params->socket == NULL) + params->socket = default_socket; + + if (params->server == NULL) + params->server = default_server; if (params->logfile != NULL && access(params->logfile, W_OK)) if (errno != ENOENT) { @@ -702,101 +793,58 @@ void parse_parameters(int argc, char * argv[], parameters * params) { } } -void write_pidfile(char * pidfile) { - int fd; - pid_t pid = getpid(); - char * pid_s; - int pid_s_len, write_count; +void serve(parameters * params) { + ssize_t read_count; + char subproto_selector; + pthread_t child; + connection_t * conn; + void * (*connection_handler_thread)(void *); - if (asprintf(&pid_s, "%lld", (long long) pid) == -1) - die(1, "Couldn't allocate pidfile string", ""); + if (listen(params->sock, 16)) + die(1, "listen", ""); - pid_s_len = strlen(pid_s); + if ((errno = pthread_create(&child, &detached, init_thread, params))) + die(1, "", "Couldn't create initialization thread: "); - fd = open(pidfile, O_WRONLY | O_CREAT | O_TRUNC, 0644); - if (fd == -1) - die(1, "", "Couldn't open pidfile path %s: ", pidfile); + while (1) { + conn = (connection_t *) must_malloc("connection state", + sizeof(connection_t)); + conn->params = params; + conn->mount_point = ""; - write_count = write(fd, pid_s, pid_s_len); - if (write_count == -1) - die(1, "", "Error writing pidfile %s: ", pidfile); + conn->sock = accept(params->sock, &conn->sa_client, &conn->socklen_client); + if (conn->sock < 0) + die(1, "accept", ""); - if (write_count != pid_s_len) - die(1, NULL, "Error writing %s to pidfile %s: only wrote %d bytes", - pid_s, pidfile, write_count); + // TODO: check for socket read conditions e.g. EAGAIN + read_count = read(conn->sock, &subproto_selector, 1); + if (read_count <= 0) + die(1, "read subprotocol selector", ""); - close(fd); - free(pid_s); -} - -#define ID_LEN 512 - -void process_events(char * events_path, int events, parameters * params) { - char buf[ID_LEN]; - int read_count; - long conn_id; - pthread_t child; - connection_t * conn; - void * (*connection_handler_thread)(void *); - - while (1) { - conn = (connection_t *) must_malloc("connection state", - sizeof(connection_t)); - conn->params = params; - conn->id = 0; - - read_count = read(events, buf, ID_LEN - 1); - if (read_count == -1) { - die(1, "", "Error reading events path %s: ", events_path); - } else if (read_count == 0) { - // TODO: this is probably the 9p server's fault due to - // not dropping the read 0 to force short read if - // the real read is flushed - log_time(conn, "read 0 from event stream %s\n", events_path); - free(conn); - continue; - } - - buf[read_count] = 0x0; - - if (read_count < 2) { - die(1, NULL, "Event connection id isn't long enough"); - } - - errno = 0; - conn_id = strtol(buf + 1, NULL, 10); - if (errno) die(1, "failed", "Connection id of string '%s'", buf); - - conn->id = conn_id; - - if (debug) log_time(conn, "handle connection %ld\n", conn_id); - - switch (buf[0]) { - case 'm': - conn->type_descr = "mount"; - connection_handler_thread = mount_thread; - break; - case 'e': - conn->type_descr = "event"; - connection_handler_thread = event_thread; - break; - default: - die(1, NULL, "Unknown connection type '%c'", buf[0]); - } - - if ((errno = pthread_create(&child, &detached, - connection_handler_thread, conn))) - die(1, "", "Couldn't create thread for %s connection '%ld': ", - conn->type_descr, conn_id); - - if (debug) log_time(conn, "thread spawned\n"); + switch (subproto_selector) { + case 'm': + conn->type_descr = "mount"; + connection_handler_thread = mount_thread; + break; + case 'e': + conn->type_descr = "event"; + connection_handler_thread = event_thread; + break; + default: + die(1, NULL, "Unknown subprotocol type '%c'", subproto_selector); } + + if ((errno = pthread_create(&child, &detached, + connection_handler_thread, conn))) + die(1, "", "Couldn't create thread for %s connection: ", + conn->type_descr); + + if (debug) log_time(conn, "thread spawned\n"); + } } int main(int argc, char * argv[]) { - int events; parameters params; - char * events_path; struct rlimit core_limit; core_limit.rlim_cur = RLIM_INFINITY; @@ -813,25 +861,14 @@ int main(int argc, char * argv[]) { PTHREAD_CREATE_DETACHED))) die(1, "Couldn't set pthread detach state", ""); - if (params.pidfile != NULL) write_pidfile(params.pidfile); - if (params.logfile != NULL) { params.logfile_fd = open(params.logfile, O_WRONLY | O_APPEND | O_CREAT); if (params.logfile_fd == -1) die(1, "", "Couldn't open log file %s: ", params.logfile); } - if (asprintf(&events_path, "%s/events", params.socket9p_root) == -1) - die(1, "Couldn't allocate events path", ""); + params.sock = bind_socket(params.socket); + serve(¶ms); - events = open(events_path, O_RDONLY | O_CLOEXEC); - if (events != -1) process_events(events_path, events, ¶ms); - - connection_t top; - top.params = ¶ms; - top.id = 0; - log_time(&top, "Failed to open events path %s: %s\n", - events_path, strerror(errno)); - free(events_path); - return 1; + return 0; } diff --git a/alpine/packages/transfused/transfused.h b/alpine/packages/transfused/transfused.h index 7e4ce149a..048ae68d7 100644 --- a/alpine/packages/transfused/transfused.h +++ b/alpine/packages/transfused/transfused.h @@ -1,12 +1,15 @@ #include +#include typedef struct { - char * socket9p_root; + char * server; + char * socket; char * fusermount; char * pidfile; char * logfile; char * mount_trigger; char * trigger_log; + int sock; pthread_mutex_t fd_lock; int logfile_fd; int trigger_fd; @@ -14,8 +17,11 @@ typedef struct { typedef struct { parameters * params; - long id; char * type_descr; + char * mount_point; + struct sockaddr sa_client; + socklen_t socklen_client; + int sock; } connection_t; pthread_attr_t detached; diff --git a/alpine/packages/transfused/transfused_log.c b/alpine/packages/transfused/transfused_log.c index b4fd272ae..1a5cf22fe 100644 --- a/alpine/packages/transfused/transfused_log.c +++ b/alpine/packages/transfused/transfused_log.c @@ -129,8 +129,8 @@ void thread_log_time(connection_t * conn, const char * fmt, ...) { // log demand to be low. if ((errno = pthread_create(&logger, &detached, log_time_thread, log_state))) - die(1, "", "Couldn't create log thread for %s connection '%ld': ", - conn->type_descr, conn->id); + die(1, "", "Couldn't create log thread for %s connection %s: ", + conn->type_descr, conn->mount_point); } void log_continue_locked(connection_t * connection, const char * fmt, ...) { diff --git a/alpine/packages/transfused/transfused_log.h b/alpine/packages/transfused/transfused_log.h index 95af5b869..07fdd0d6e 100644 --- a/alpine/packages/transfused/transfused_log.h +++ b/alpine/packages/transfused/transfused_log.h @@ -1,3 +1,5 @@ +#include + #include "transfused.h" void die(int exit_code, const char * perror_arg, const char * fmt, ...); diff --git a/alpine/packages/transfused/transfused_vsock.c b/alpine/packages/transfused/transfused_vsock.c new file mode 100644 index 000000000..25003a6eb --- /dev/null +++ b/alpine/packages/transfused/transfused_vsock.c @@ -0,0 +1,99 @@ +#include +#include + +#include + +#include "include/uapi/linux/vm_sockets.h" + +#include "transfused_log.h" + +long parse_cid(const char * address) { + char * end = NULL; + long cid = strtol(address, &end, 10); + if (address == end || *end != ':') { + *end = 0; + die(2, NULL, "Invalid vsock cid: %s", address); + } + return cid; +} + +long parse_port(const char * port_str) { + char * end = NULL; + long port = strtol(port_str, &end, 10); + if (port_str == end || *end != '\0') { + *end = 0; + die(2, NULL, "Invalid vsock port: %s", port_str); + } + return port; +} + +int find_colon(const char * address) { + int colon = 0; + + while (address[colon] != '\0') + if (address[colon] == ':') break; + else colon++; + + if (address[colon] == '\0') + die(2, NULL, "Missing port in vsock address %s", address); + + return colon; +} + +int bind_vsock(const char * address) { + long cid, port; + int colon; + + struct sockaddr_vm sa_listen = { + .svm_family = AF_VSOCK, + }; + + int sock_fd; + + colon = find_colon(address); + + if (address[0] == '_' && colon == 1) cid = VMADDR_CID_ANY; + else cid = parse_cid(address); + + port = parse_port(address + colon + 1); + + sa_listen.svm_cid = cid; + sa_listen.svm_port = port; + + sock_fd = socket(AF_VSOCK, SOCK_STREAM, 0); + if (sock_fd < 0) + die(1, "socket(AF_VSOCK)", ""); + + if (bind(sock_fd, (struct sockaddr *) &sa_listen, sizeof(sa_listen))) + die(1, "bind(AF_VSOCK)", ""); + + return sock_fd; +} + +int connect_vsock(const char * address) { + long cid, port; + int colon; + + struct sockaddr_vm sa_connect = { + .svm_family = AF_VSOCK, + }; + + int sock_fd; + + colon = find_colon(address); + + cid = parse_cid(address); + port = parse_port(address + colon + 1); + + sa_connect.svm_cid = cid; + sa_connect.svm_port = port; + + sock_fd = socket(AF_VSOCK, SOCK_STREAM, 0); + if (sock_fd < 0) + die(1, "socket(AF_VSOCK)", ""); + + if (connect(sock_fd, (struct sockaddr *) &sa_connect, sizeof(sa_connect))) + die(1, "connect(AF_VSOCK)", ""); + + return sock_fd; +} diff --git a/alpine/packages/transfused/transfused_vsock.h b/alpine/packages/transfused/transfused_vsock.h new file mode 100644 index 000000000..a6bef6f08 --- /dev/null +++ b/alpine/packages/transfused/transfused_vsock.h @@ -0,0 +1,2 @@ +int bind_vsock(const char * address); +int connect_vsock(const char * address);