Make toasty client thread-safe and implement toasty_wakeup

inc/ToastyFS.h

@@ -41,6 +41,11 @@ ToastyFS *toasty_connect(ToastyString addr, uint16_t port);
 // Release all resources associated to this client
 void toasty_disconnect(ToastyFS *toasty);
 
+// Threads can call this function to wake up a thread blocked
+// inside "toasty_wait_result".
+// Return 0 on success, -1 on error.
+int toasty_wakeup(ToastyFS *toasty);
+
 //////////////////////////////////////////////////////////////////////////////////
 // BLOCKING API
 //////////////////////////////////////////////////////////////////////////////////

misc/TODO.txt

@@ -22,3 +22,4 @@
 [ ] Check write() errors when appending WAL entries
 [ ] When WAL entries are partially written, remove partial data and fail
 [ ] Find a way to ensure listing operations of large directories are handled
+[ ] What happens if a client adds a chunk such that the hash already exists in the system? The client doesn't know it exists already. Will this lead to over-replication?

src/chunk_server.c

@@ -834,7 +834,8 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts
     state->trace = trace;
     state->reconnect_delay = 1; // 1 second
 
-    tcp_context_init(&state->tcp);
+    if (tcp_context_init(&state->tcp) < 0)
+        return -1;
 
     int ret = tcp_listen(&state->tcp, addr, port);
     if (ret < 0) {
@@ -929,6 +930,10 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled
         int conn_idx = events[i].conn_idx;
         switch (events[i].type) {
 
+            case EVENT_WAKEUP:
+            // Do nothing
+            break;
+
             case EVENT_CONNECT:
             if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) {
                 assert(state->disconnect_time == INVALID_TIME);

src/client.c

@@ -1,14 +1,18 @@
-#include "basic.h"
 #include <assert.h>
 #include <stdint.h>
 #include <string.h>
 #include <stdlib.h>
 
 #ifdef _WIN32
+#define WIN32_LEAN_AND_MEAN
+#include <windows.h>
 #define POLL WSAPoll
+typedef CRITICAL_SECTION Mutex;
 #else
+#include <pthread.h>
 #include <arpa/inet.h>
 #define POLL poll
+typedef pthread_mutex_t Mutex;
 #endif
 
 #include "tcp.h"
@@ -160,6 +164,7 @@ typedef struct {
 struct ToastyFS {
 
     TCP tcp;
+    Mutex mutex;
 
     MetadataServer metadata_server;
 
@@ -170,6 +175,54 @@ struct ToastyFS {
     Operation operations[MAX_OPERATIONS];
 };
 
+static int mutex_init(Mutex *mutex)
+{
+#ifdef _WIN32
+    InitializeCriticalSection(mutex); // TODO: mock?
+    return 0;
+#else
+    if (pthread_mutex_init(mutex, NULL)) // TODO: mock
+        return -1;
+    return 0;
+#endif
+}
+
+static int mutex_free(Mutex *mutex)
+{
+#ifdef _WIN32
+    DeleteCriticalSection(mutex); // TODO: mock?
+    return 0;
+#else
+    if (pthread_mutex_destroy(mutex)) // TODO: mock
+        return -1;
+    return 0;
+#endif
+}
+
+static int mutex_lock(Mutex *mutex)
+{
+#ifdef _WIN32
+    EnterCriticalSection(mutex); // TODO: mock?
+    return 0;
+#else
+    if (pthread_mutex_lock(mutex)) // TODO: mock
+        return -1;
+    return 0;
+#endif
+}
+
+static int mutex_unlock(Mutex *mutex)
+{
+#ifdef _WIN32
+    LeaveCriticalSection(mutex); // TODO: mock?
+    return 0;
+#else
+    if (pthread_mutex_unlock(mutex)) // TODO: mock
+        return -1;
+    return 0;
+#endif
+}
+
 static void request_queue_init(RequestQueue *reqs)
 {
     reqs->head = 0;
@@ -281,10 +334,20 @@ ToastyFS *toasty_connect(ToastyString addr, uint16_t port)
         }
     }
 
-    tcp_context_init(&toasty->tcp);
+    if (mutex_init(&toasty->mutex) < 0) {
+        sys_free(toasty);
+        return NULL;
+    }
+
+    if (tcp_context_init(&toasty->tcp) < 0) {
+        mutex_free(&toasty->mutex);
+        sys_free(toasty);
+        return NULL;
+    }
 
     if (tcp_connect(&toasty->tcp, addr2, TAG_METADATA_SERVER, NULL) < 0) {
         tcp_context_free(&toasty->tcp);
+        mutex_free(&toasty->mutex);
         sys_free(toasty);
         return NULL;
     }
@@ -310,6 +373,21 @@ ToastyFS *toasty_connect(ToastyString addr, uint16_t port)
     return toasty;
 }
 
+int toasty_wakeup(ToastyFS *toasty)
+{
+    if (mutex_lock(&toasty->mutex) < 0)
+        return -1;
+
+    int ret;
+    if (tcp_wakeup(&toasty->tcp) < 0)
+        ret = -1;
+
+    if (mutex_unlock(&toasty->mutex) < 0)
+        return -1;
+
+    return ret;
+}
+
 static void
 metadata_server_request_start(ToastyFS *toasty, MessageWriter *writer, uint16_t type)
 {
@@ -346,6 +424,7 @@ metadata_server_request_end(ToastyFS *toasty, MessageWriter *writer, int opidx,
 void toasty_disconnect(ToastyFS *toasty)
 {
     tcp_context_free(&toasty->tcp);
+    mutex_free(&toasty->mutex);
     sys_free(toasty);
 }
 
@@ -449,31 +528,38 @@ static int send_download_chunk(ToastyFS *toasty, int chunk_server_idx,
 static ToastyHandle begin_create(ToastyFS *toasty,
     ToastyString path, bool is_dir, uint32_t chunk_size)
 {
+    int opidx = -1;
+    ToastyHandle handle = TOASTY_INVALID;
+
+    if (mutex_lock(&toasty->mutex) < 0)
+        return TOASTY_INVALID;
+
     OperationType type = OPERATION_TYPE_CREATE;
 
-    int opidx = alloc_operation(toasty, type, 0, NULL, 0);
+    opidx = alloc_operation(toasty, type, 0, NULL, 0);
     if (opidx < 0)
-        return TOASTY_INVALID;
+        goto unlock_and_exit;
+
+    if (path.len > UINT16_MAX) {
+        free_operation(toasty, opidx);
+        opidx = -1;
+        goto unlock_and_exit;
+    }
+    uint16_t tmp = path.len;
+    uint8_t tmp_u8 = is_dir;
 
     MessageWriter writer;
     metadata_server_request_start(toasty, &writer, MESSAGE_TYPE_CREATE);
 
-    if (path.len > UINT16_MAX) {
-        free_operation(toasty, opidx);
-        return TOASTY_INVALID;
-    }
-    uint16_t tmp = path.len;
     message_write(&writer, &tmp, sizeof(tmp));
-
     message_write(&writer, path.ptr, path.len);
-
-    uint8_t tmp_u8 = is_dir;
     message_write(&writer, &tmp_u8, sizeof(tmp_u8));
 
     if (!is_dir) {
         if (chunk_size == 0 || chunk_size > UINT32_MAX) {
             free_operation(toasty, opidx);
-            return TOASTY_INVALID;
+            opidx = -1;
+            goto unlock_and_exit;
         }
         uint32_t tmp_u32 = chunk_size;
         message_write(&writer, &tmp_u32, sizeof(tmp_u32));
@@ -481,10 +567,19 @@ static ToastyHandle begin_create(ToastyFS *toasty,
 
     if (metadata_server_request_end(toasty, &writer, opidx, 0) < 0) {
         free_operation(toasty, opidx);
-        return TOASTY_INVALID;
+        opidx = -1;
+        goto unlock_and_exit;
     }
 
-    return operation_to_handle(toasty, opidx);
+    handle = operation_to_handle(toasty, opidx);
+
+unlock_and_exit:
+    if (mutex_unlock(&toasty->mutex) < 0) {
+        if (opidx > -1)
+            free_operation(toasty, opidx);
+        return TOASTY_INVALID;
+    }
+    return handle;
 }
 
 ToastyHandle toasty_begin_create_dir(ToastyFS *toasty, ToastyString path)
@@ -500,15 +595,22 @@ ToastyHandle toasty_begin_create_file(ToastyFS *toasty, ToastyString path,
 
 ToastyHandle toasty_begin_delete(ToastyFS *toasty, ToastyString path)
 {
+    int opidx = -1;
+    ToastyHandle handle = TOASTY_INVALID;
+
+    if (mutex_lock(&toasty->mutex) < 0)
+        return TOASTY_INVALID;
+
     OperationType type = OPERATION_TYPE_DELETE;
 
-    int opidx = alloc_operation(toasty, type, 0, NULL, 0);
+    opidx = alloc_operation(toasty, type, 0, NULL, 0);
     if (opidx < 0)
-        return TOASTY_INVALID;
+        goto unlock_and_exit;
 
     if (path.len > UINT16_MAX) {
         free_operation(toasty, opidx);
-        return TOASTY_INVALID;
+        opidx = -1;
+        goto unlock_and_exit;
     }
     uint16_t tmp = path.len;
 
@@ -518,38 +620,61 @@ ToastyHandle toasty_begin_delete(ToastyFS *toasty, ToastyString path)
     message_write(&writer, path.ptr, path.len);
     if (metadata_server_request_end(toasty, &writer, opidx, 0) < 0) {
         free_operation(toasty, opidx);
-        return TOASTY_INVALID;
+        opidx = -1;
+        goto unlock_and_exit;
     }
 
-    return operation_to_handle(toasty, opidx);
+    handle = operation_to_handle(toasty, opidx);
+
+unlock_and_exit:
+    if (mutex_unlock(&toasty->mutex) < 0) {
+        if (opidx > -1)
+            free_operation(toasty, opidx);
+        return TOASTY_INVALID;
+    }
+    return handle;
 }
 
 ToastyHandle toasty_begin_list(ToastyFS *toasty, ToastyString path)
 {
+    int opidx = -1;
+    ToastyHandle handle = TOASTY_INVALID;
+
+    if (mutex_lock(&toasty->mutex) < 0)
+        return TOASTY_INVALID;
+
     OperationType type = OPERATION_TYPE_LIST;
 
-    int opidx = alloc_operation(toasty, type, 0, NULL, 0);
+    opidx = alloc_operation(toasty, type, 0, NULL, 0);
     if (opidx < 0)
-        return TOASTY_INVALID;
+        goto unlock_and_exit;
 
     if (path.len > UINT16_MAX) {
         free_operation(toasty, opidx);
-        return TOASTY_INVALID;
+        opidx = -1;
+        goto unlock_and_exit;
     }
     uint16_t tmp = path.len;
 
     MessageWriter writer;
     metadata_server_request_start(toasty, &writer, MESSAGE_TYPE_LIST);
-
     message_write(&writer, &tmp, sizeof(tmp));
     message_write(&writer, path.ptr, path.len);
-
     if (metadata_server_request_end(toasty, &writer, opidx, 0) < 0) {
         free_operation(toasty, opidx);
-        return TOASTY_INVALID;
+        opidx = -1;
+        goto unlock_and_exit;
     }
 
-    return operation_to_handle(toasty, opidx);
+    handle = operation_to_handle(toasty, opidx);
+
+unlock_and_exit:
+    if (mutex_unlock(&toasty->mutex) < 0) {
+        if (opidx > -1)
+            free_operation(toasty, opidx);
+        return TOASTY_INVALID;
+    }
+    return handle;
 }
 
 static int send_read_message(ToastyFS *toasty, int opidx, int tag, ToastyString path, uint32_t offset, uint32_t length)
@@ -571,36 +696,66 @@ static int send_read_message(ToastyFS *toasty, int opidx, int tag, ToastyString
 
 ToastyHandle toasty_begin_read(ToastyFS *toasty, ToastyString path, int off, void *dst, int len)
 {
+    int opidx = -1;
+    ToastyHandle handle = TOASTY_INVALID;
+
+    if (mutex_lock(&toasty->mutex) < 0)
+        goto unlock_and_exit;
+
     OperationType type = OPERATION_TYPE_READ;
 
-    int opidx = alloc_operation(toasty, type, off, dst, len);
+    opidx = alloc_operation(toasty, type, off, dst, len);
     if (opidx < 0)
-        return TOASTY_INVALID;
+        goto unlock_and_exit;
 
     if (send_read_message(toasty, opidx, TAG_RETRIEVE_METADATA_FOR_READ, path, off, len) < 0) {
         free_operation(toasty, opidx);
-        return TOASTY_INVALID;
+        opidx = -1;
+        goto unlock_and_exit;
     }
 
-    return operation_to_handle(toasty, opidx);
+    handle = operation_to_handle(toasty, opidx);
+
+unlock_and_exit:
+    if (mutex_unlock(&toasty->mutex) < 0) {
+        if (opidx > -1)
+            free_operation(toasty, opidx);
+        return TOASTY_INVALID;
+    }
+    return handle;
 }
 
 ToastyHandle toasty_begin_write(ToastyFS *toasty, ToastyString path, int off, void *src, int len)
 {
+    int opidx = -1;
+    ToastyHandle handle = TOASTY_INVALID;
+
+    if (mutex_lock(&toasty->mutex) < 0)
+        goto unlock_and_exit;
+
     OperationType type = OPERATION_TYPE_WRITE;
 
-    int opidx = alloc_operation(toasty, type, off, src, len);
+    opidx = alloc_operation(toasty, type, off, src, len);
     if (opidx < 0)
-        return TOASTY_INVALID;
+        goto unlock_and_exit;
 
     toasty->operations[opidx].path = path; // TODO: must be a copy
 
     if (send_read_message(toasty, opidx, TAG_RETRIEVE_METADATA_FOR_WRITE, path, off, len) < 0) {
         free_operation(toasty, opidx);
-        return TOASTY_INVALID;
+        opidx = -1;
+        goto unlock_and_exit;
     }
 
-    return operation_to_handle(toasty, opidx);
+    handle = operation_to_handle(toasty, opidx);
+
+unlock_and_exit:
+    if (mutex_unlock(&toasty->mutex) < 0) {
+        if (opidx > -1)
+            free_operation(toasty, opidx);
+        return TOASTY_INVALID;
+    }
+    return handle;
 }
 
 void toasty_free_result(ToastyResult *result)
@@ -1862,13 +2017,22 @@ translate_operation_into_result(ToastyFS *toasty, int opidx, ToastyResult *resul
 
 int toasty_get_result(ToastyFS *toasty, ToastyHandle handle, ToastyResult *result)
 {
+    int ret;
+
+    if (mutex_lock(&toasty->mutex) < 0) {
+        ret = -1;
+        goto unlock_and_exit;
+    }
+
     int opidx;
     if (handle == TOASTY_INVALID)
         opidx = -1;
     else {
         opidx = handle_to_operation(toasty, handle);
-        if (opidx < 0)
-            return -1;
+        if (opidx < 0) {
+            ret = -1;
+            goto unlock_and_exit;
+        }
     }
 
     if (opidx < 0) {
@@ -1878,19 +2042,30 @@ int toasty_get_result(ToastyFS *toasty, ToastyHandle handle, ToastyResult *resul
                 continue;
             j++;
 
-            if (translate_operation_into_result(toasty, i, result))
-                return 0;
+            if (translate_operation_into_result(toasty, i, result)) {
+                ret = 0;
+                goto unlock_and_exit;
+            }
         }
     } else {
-        if (translate_operation_into_result(toasty, opidx, result))
-            return 0;
+        if (translate_operation_into_result(toasty, opidx, result)) {
+            ret = 0;
+            goto unlock_and_exit;
+        }
     }
 
-    return 1;
+    ret = 1;
+unlock_and_exit:
+    if (mutex_unlock(&toasty->mutex) < 0)
+        return -1;
+    return ret;
 }
 
 int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *polled, int num_polled)
 {
+    if (mutex_lock(&toasty->mutex) < 0)
+        return -1;
+
     int num_events;
     Event events[MAX_CONNS+1];
 
@@ -1899,6 +2074,10 @@ int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *poll
         int conn_idx = events[i].conn_idx;
         switch (events[i].type) {
 
+            case EVENT_WAKEUP:
+            // Do nothing
+            break;
+
             case EVENT_CONNECT:
             {
                 int tag = tcp_get_tag(&toasty->tcp, conn_idx);
@@ -2003,7 +2182,12 @@ int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *poll
         }
     }
 
-    return tcp_register_events(&toasty->tcp, contexts, polled);
+    int ret = tcp_register_events(&toasty->tcp, contexts, polled);
+
+    if (mutex_unlock(&toasty->mutex) < 0)
+        return -1;
+
+    return ret;
 }
 
 int toasty_wait_result(ToastyFS *toasty, ToastyHandle handle, ToastyResult *result, int timeout)

src/metadata_server.c

@@ -1001,7 +1001,8 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co
     for (int i = 0; i < MAX_CHUNK_SERVERS; i++)
         state->chunk_servers[i].used = false;
 
-    tcp_context_init(&state->tcp);
+    if (tcp_context_init(&state->tcp) < 0)
+        return -1;
 
     int ret = tcp_listen(&state->tcp, addr, port);
     if (ret < 0) {
@@ -1050,6 +1051,10 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
         int conn_idx = events[i].conn_idx;
         switch (events[i].type) {
 
+            case EVENT_WAKEUP:
+            // Do nothing
+            break;
+
             case EVENT_CONNECT:
             tcp_set_tag(&state->tcp, conn_idx, CONNECTION_TAG_UNKNOWN, false);
             break;

src/tcp.c

@@ -86,6 +86,65 @@ static SOCKET create_listen_socket(string addr, uint16_t port)
     return fd;
 }
 
+static int create_socket_pair(SOCKET *a, SOCKET *b)
+{
+#ifdef _WIN32
+    SOCKET sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    if (sock == INVALID_SOCKET)
+        return -1;
+
+    // Bind to loopback address with port 0 (dynamic port assignment)
+    struct sockaddr_in addr;
+    int addr_len = sizeof(addr);
+    memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); // 127.0.0.1
+    addr.sin_port = 0; // Let system choose port
+
+    if (bind(sock, (struct sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR) {
+        closesocket(sock);
+        return -1;
+    }
+
+    if (getsockname(sock, (struct sockaddr*)&addr, &addr_len) == SOCKET_ERROR) {
+        closesocket(sock);
+        return -1;
+    }
+
+    if (connect(sock, (struct sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR) {
+        closesocket(sock);
+        return -1;
+    }
+
+    *a = sock;
+    *b = sock;
+
+    // Optional: Set socket to non-blocking mode
+    // This prevents send() from blocking if the receive buffer is full
+    u_long mode = 1;
+    ioctlsocket(sock, FIONBIO, &mode); // TODO: does this fail?
+    return 0;
+#else
+    int fds[2];
+    if (pipe(fds) < 0)
+        return -1;
+    *a = fds[0];
+    *b = fds[1];
+    return 0;
+#endif
+}
+
+static void close_socket_pair(SOCKET a, SOCKET b)
+{
+#ifdef _WIN32
+    closesocket(a);
+    (void) b;
+#else
+    close(a);
+    close(b);
+#endif
+}
+
 static void conn_init(Connection *conn, SOCKET fd, bool connecting)
 {
     conn->fd = fd;
@@ -122,10 +181,15 @@ static int conn_events(Connection *conn)
     return events;
 }
 
-void tcp_context_init(TCP *tcp)
+int tcp_context_init(TCP *tcp)
 {
     tcp->listen_fd = INVALID_SOCKET;
     tcp->num_conns = 0;
+
+    if (create_socket_pair(&tcp->wait_fd, &tcp->signal_fd) < 0)
+        return -1;
+
+    return 0;
 }
 
 void tcp_context_free(TCP *tcp)
@@ -140,6 +204,14 @@ void tcp_context_free(TCP *tcp)
 
     if (tcp->listen_fd != INVALID_SOCKET)
         CLOSE_SOCKET(tcp->listen_fd);
+
+    close_socket_pair(tcp->wait_fd, tcp->signal_fd);
+}
+
+int tcp_wakeup(TCP *tcp)
+{
+    send(tcp->signal_fd, "0", 1, 0); // TODO: Handle error
+    return 0;
 }
 
 int tcp_index_from_tag(TCP *tcp, int tag)
@@ -199,6 +271,15 @@ int tcp_register_events(TCP *tcp, void **contexts, struct pollfd *polled)
 {
     int num_polled = 0;
 
+    // TODO: Check that there is enough capacity in the polled array.
+    //       At the moment only MAX_CONNS plus 1 for the listener are
+    //       allowed.
+    polled[num_polled].fd = tcp->wait_fd;
+    polled[num_polled].events = POLLIN;
+    polled[num_polled].revents = 0;
+    contexts[num_polled] = NULL;
+    num_polled++;
+
     if (tcp->listen_fd != INVALID_SOCKET && tcp->num_conns < MAX_CONNS) {
         polled[num_polled].fd = tcp->listen_fd;
         polled[num_polled].events = POLLIN;
@@ -222,14 +303,21 @@ int tcp_register_events(TCP *tcp, void **contexts, struct pollfd *polled)
 }
 
 // The "events" array must be an array of capacity MAX_CONNS+1
+// TODO: Actually, it should have capacity MAX_CONNS+2
 int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd *polled, int num_polled)
 {
     bool removed[MAX_CONNS+1];
 
     int num_events = 0;
-    for (int i = 0; i < num_polled; i++) {
+    for (int i = 1; i < num_polled; i++) {
 
-        if (polled[i].fd == tcp->listen_fd) {
+        if (polled[i].fd == tcp->wait_fd) {
+
+            char buf[100];
+            recv(tcp->wait_fd, buf, sizeof(buf), 0); // TODO: Make sure all bytes are consumed
+            events[num_events++] = (Event) { EVENT_WAKEUP, -1, -1 };
+
+        } else if (polled[i].fd == tcp->listen_fd) {
 
             assert(contexts[i] == NULL);
 

src/tcp.h

@@ -15,6 +15,7 @@
 #define MAX_CONNS 512
 
 typedef enum {
+    EVENT_WAKEUP,
     EVENT_MESSAGE,
     EVENT_CONNECT,
     EVENT_DISCONNECT,
@@ -38,13 +39,16 @@ typedef struct {
 
 typedef struct {
     SOCKET listen_fd;
+    SOCKET wait_fd;
+    SOCKET signal_fd;
     int    num_conns;
     Connection conns[MAX_CONNS];
 } TCP;
 
 bool addr_eql(Address a, Address b);
-void tcp_context_init(TCP *tcp);
+int  tcp_context_init(TCP *tcp);
 void tcp_context_free(TCP *tcp);
+int  tcp_wakeup(TCP *tcp);
 int  tcp_index_from_tag(TCP *tcp, int tag);
 int  tcp_listen(TCP *tcp, string addr, uint16_t port);
 int  tcp_next_message(TCP *tcp, int conn_idx, ByteView *msg, uint16_t *type);