From b0313ffcf8eb47f82efec86e8390ee462c74e27b Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 20 Nov 2025 03:34:05 +0100 Subject: [PATCH] Make toasty client thread-safe and implement toasty_wakeup --- inc/ToastyFS.h | 5 + misc/TODO.txt | 1 + src/chunk_server.c | 7 +- src/client.c | 270 +++++++++++++++++++++++++++++++++++------- src/metadata_server.c | 7 +- src/tcp.c | 94 ++++++++++++++- src/tcp.h | 6 +- 7 files changed, 341 insertions(+), 49 deletions(-) diff --git a/inc/ToastyFS.h b/inc/ToastyFS.h index 6cda758..c4042b1 100644 --- a/inc/ToastyFS.h +++ b/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 ////////////////////////////////////////////////////////////////////////////////// diff --git a/misc/TODO.txt b/misc/TODO.txt index e361387..c064cea 100644 --- a/misc/TODO.txt +++ b/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? diff --git a/src/chunk_server.c b/src/chunk_server.c index e4edc0e..8de1b0c 100644 --- a/src/chunk_server.c +++ b/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); diff --git a/src/client.c b/src/client.c index f4287fc..4c054a6 100644 --- a/src/client.c +++ b/src/client.c @@ -1,14 +1,18 @@ -#include "basic.h" #include #include #include #include #ifdef _WIN32 +#define WIN32_LEAN_AND_MEAN +#include #define POLL WSAPoll +typedef CRITICAL_SECTION Mutex; #else +#include #include #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) diff --git a/src/metadata_server.c b/src/metadata_server.c index e45ea31..9d2fc70 100644 --- a/src/metadata_server.c +++ b/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; diff --git a/src/tcp.c b/src/tcp.c index 8dee8eb..8bb764c 100644 --- a/src/tcp.c +++ b/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); diff --git a/src/tcp.h b/src/tcp.h index 43d5765..9417970 100644 --- a/src/tcp.h +++ b/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);