From 7591ef6ee18fd24736a0028fecd8209d81679a5a Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Wed, 25 Feb 2026 20:06:21 +0100 Subject: [PATCH] Cleanup of message.c/.h --- .gitignore | 3 + lib/basic.h | 2 + lib/byte_queue.c | 20 +-- lib/byte_queue.h | 13 +- lib/http_server.c | 29 ++-- lib/http_server.h | 6 +- lib/message.c | 371 +++++++++++++++++++++++---------------------- lib/message.h | 37 ++--- lib/tcp.c | 365 ++++++++++++++++++++++++-------------------- lib/tcp.h | 22 +-- lib/tls_schannel.c | 10 +- src/client.c | 37 ++--- src/http_proxy.c | 11 +- src/server.c | 81 +++++----- src/server.h | 2 +- 15 files changed, 512 insertions(+), 497 deletions(-) diff --git a/.gitignore b/.gitignore index 832be35..c30a0a7 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,9 @@ toastyfs toastyfs_client toastyfs_simulation +toastyfs_server +toastyfs_http_proxy +toastyfs_random_client *.gcno *.gcda *.gcov diff --git a/lib/basic.h b/lib/basic.h index dfb641f..d8fe862 100644 --- a/lib/basic.h +++ b/lib/basic.h @@ -40,6 +40,8 @@ typedef uint64_t Time; #define UNREACHABLE __builtin_trap(); +#define STATIC_ASSERT(X) _Static_assert((X), "") + bool streq(string s1, string s2); Time get_current_time(void); diff --git a/lib/byte_queue.c b/lib/byte_queue.c index aa8767f..2498c0e 100644 --- a/lib/byte_queue.c +++ b/lib/byte_queue.c @@ -68,10 +68,10 @@ int byte_queue_full(ByteQueue *queue) // // Note: // - You can't have more than one pending read. -ByteView byte_queue_read_buf(ByteQueue *queue) +string byte_queue_read_buf(ByteQueue *queue) { if (queue->flags & BYTE_QUEUE_ERROR) - return (ByteView) {NULL, 0}; + return (string) {NULL, 0}; assert((queue->flags & BYTE_QUEUE_READ) == 0); queue->flags |= BYTE_QUEUE_READ; @@ -79,9 +79,9 @@ ByteView byte_queue_read_buf(ByteQueue *queue) queue->read_target_size = queue->size; if (queue->data == NULL) - return (ByteView) {NULL, 0}; + return (string) {NULL, 0}; - return (ByteView) { queue->data + queue->head, queue->used }; + return (string) { queue->data + queue->head, queue->used }; } // Complete a previously started operation on the queue. @@ -113,15 +113,15 @@ bool byte_queue_reading(ByteQueue *queue) return (queue->flags & BYTE_QUEUE_READ) != 0; } -ByteView byte_queue_write_buf(ByteQueue *queue) +string byte_queue_write_buf(ByteQueue *queue) { if ((queue->flags & BYTE_QUEUE_ERROR) || queue->data == NULL) - return (ByteView) {NULL, 0}; + return (string) {NULL, 0}; assert((queue->flags & BYTE_QUEUE_WRITE) == 0); queue->flags |= BYTE_QUEUE_WRITE; - return (ByteView) { + return (string) { queue->data + (queue->head + queue->used), queue->size - (queue->head + queue->used), }; @@ -232,7 +232,7 @@ int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap) if (size > queue->limit) size = queue->limit; - uint8_t *data = malloc(size); + char *data = malloc(size); if (!data) { queue->flags |= BYTE_QUEUE_ERROR; return 0; @@ -263,7 +263,7 @@ int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap) void byte_queue_write(ByteQueue *queue, void *ptr, uint32_t len) { byte_queue_write_setmincap(queue, len); - ByteView dst = byte_queue_write_buf(queue); + string dst = byte_queue_write_buf(queue); if (dst.ptr) { memcpy(dst.ptr, ptr, len); byte_queue_write_ack(queue, len); @@ -290,7 +290,7 @@ void byte_queue_patch(ByteQueue *queue, ByteQueueOffset off, assert(len <= queue->used - (off - queue->curs)); // Perform the patch - uint8_t *dst = queue->data + queue->head + (off - queue->curs); + char *dst = queue->data + queue->head + (off - queue->curs); memcpy(dst, src, len); } diff --git a/lib/byte_queue.h b/lib/byte_queue.h index 0684db0..d6780a6 100644 --- a/lib/byte_queue.h +++ b/lib/byte_queue.h @@ -5,19 +5,14 @@ #include "basic.h" -typedef struct { - uint8_t *ptr; - size_t len; -} ByteView; - typedef struct { uint64_t curs; - uint8_t* data; + char* data; uint32_t head; uint32_t size; uint32_t used; uint32_t limit; - uint8_t* read_target; + char* read_target; uint32_t read_target_size; int flags; } ByteQueue; @@ -37,11 +32,11 @@ int byte_queue_error(ByteQueue *queue); int byte_queue_empty(ByteQueue *queue); int byte_queue_full(ByteQueue *queue); -ByteView byte_queue_read_buf(ByteQueue *queue); +string byte_queue_read_buf(ByteQueue *queue); void byte_queue_read_ack(ByteQueue *queue, uint32_t num); bool byte_queue_reading(ByteQueue *queue); -ByteView byte_queue_write_buf(ByteQueue *queue); +string byte_queue_write_buf(ByteQueue *queue); void byte_queue_write_ack(ByteQueue *queue, uint32_t num); int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap); void byte_queue_write(ByteQueue *queue, void *ptr, uint32_t len); diff --git a/lib/http_server.c b/lib/http_server.c index 0fc2f9b..d9c520c 100644 --- a/lib/http_server.c +++ b/lib/http_server.c @@ -99,24 +99,20 @@ void http_server_free(HTTP_Server *server) free(server->conns); } -int http_server_listen_tcp(HTTP_Server *server, string addr, uint16_t port) +int http_server_listen_tcp(HTTP_Server *server, Address addr) { - int ret = tcp_listen_tcp(&server->tcp, addr, port, true, 128); + int ret = tcp_listen_tcp(server->tcp, addr); if (ret < 0) return -1; return 0; } -int http_server_listen_tls(HTTP_Server *server, string addr, uint16_t port, +int http_server_listen_tls(HTTP_Server *server, Address addr, string cert_file, string key_file) { #ifdef TLS_ENABLED - int ret = tls_server_init(&server->tcp.tls, cert_file, key_file); - if (ret < 0) - return -1; - - ret = tcp_listen_tls(&server->tcp, addr, port, true, 128); + int ret = tcp_listen_tls(server->tcp, addr, cert_file, key_file); if (ret < 0) return -1; @@ -127,9 +123,9 @@ int http_server_listen_tls(HTTP_Server *server, string addr, uint16_t port, #endif } -int http_server_add_cert(HTTP_Server *server, string cert_file, string key_file) +int http_server_add_cert(HTTP_Server *server, string domain, string cert_file, string key_file) { - int ret = tcp_add_cert(&server->tcp, cert_file, key_file); + int ret = tcp_add_cert(server->tcp, domain, cert_file, key_file); if (ret < 0) return -1; @@ -163,10 +159,10 @@ static void http_conn_free(HTTP_Conn *conn) void http_server_process_events(HTTP_Server *server, void **ptrs, struct pollfd *arr, int num) { - tcp_process_events(&server->tcp, ptrs, arr, num); + tcp_process_events(server->tcp, ptrs, arr, num); TCP_Event event; - while (tcp_next_event(&server->tcp, &event)) { + while (tcp_next_event(server->tcp, &event)) { if (event.flags & TCP_EVENT_NEW) { // New connection. Find an HTTP_Conn struct. @@ -192,10 +188,7 @@ void http_server_process_events(HTTP_Server *server, bool defer_close = false; if (event.flags & TCP_EVENT_DATA) { - // Only idle connections can buffer bytes - assert(conn->state == HTTP_CONN_STATE_IDLE); - - ByteView src = tcp_read_buf(event.handle); + string src = tcp_read_buf(event.handle); int ret = chttp_parse_request((char*) src.ptr, src.len, &conn->request); if (ret < 0) { tcp_read_ack(event.handle, 0); @@ -234,7 +227,7 @@ void http_server_process_events(HTTP_Server *server, int http_server_register_events(HTTP_Server *server, void **ptrs, struct pollfd *arr, int cap) { - return tcp_register_events(&server->tcp, ptrs, arr, cap); + return tcp_register_events(server->tcp, ptrs, arr, cap); } bool http_server_next_request(HTTP_Server *server, @@ -373,7 +366,7 @@ void http_response_builder_submit(HTTP_ResponseBuilder builder) int ret = snprintf(buf, sizeof(buf), "%d", content_length); assert(ret > 0); assert(ret < (int) sizeof(buf)); - tcp_patch(conn->handle, conn->content_length_header_offset, buf, ret); + tcp_patch(conn->handle, conn->content_length_header_offset, (string) { buf, ret }); conn->num_served++; diff --git a/lib/http_server.h b/lib/http_server.h index 74c7804..a1f7183 100644 --- a/lib/http_server.h +++ b/lib/http_server.h @@ -38,12 +38,12 @@ typedef struct { int http_server_init(HTTP_Server *server, int max_conns); void http_server_free(HTTP_Server *server); -int http_server_listen_tcp(HTTP_Server *server, string addr, uint16_t port); +int http_server_listen_tcp(HTTP_Server *server, Address addr); -int http_server_listen_tls(HTTP_Server *server, string addr, uint16_t port, +int http_server_listen_tls(HTTP_Server *server, Address addr, string cert_file, string key_file); -int http_server_add_cert(HTTP_Server *server, string cert_file, string key_file); +int http_server_add_cert(HTTP_Server *server, string domain, string cert_file, string key_file); void http_server_process_events(HTTP_Server *server, void **ptrs, struct pollfd *arr, int cap); diff --git a/lib/message.c b/lib/message.c index e2cd1db..a255234 100644 --- a/lib/message.c +++ b/lib/message.c @@ -1,76 +1,116 @@ +#ifdef MAIN_SIMULATION +#define QUAKEY_ENABLE_MOCKS +#endif + #include #include +#include "tcp.h" #include "message.h" #define MESSAGE_SYSTEM_VERSION 1 +#define MESSAGE_SYSTEM_NODE_LIMIT 8 -int message_system_init(MessageSystem *msys, - Address *addrs, int num_addrs) +typedef struct { + bool used; + TCP_Handle handle; + int senders[MESSAGE_SYSTEM_NODE_LIMIT]; + int num_senders; + void* message; +} Channel; + +struct MessageSystem { + + TCP *tcp; + + Address addrs[MESSAGE_SYSTEM_NODE_LIMIT]; + int num_addrs; + + int max_channels; + Channel channels[]; +}; + +MessageSystem *message_system_init(Address *addrs, int num_addrs) { - if (num_addrs > MESSAGE_SYSTEM_NODE_LIMIT) - return -1; + int max_channels = 2 * num_addrs + 1; + MessageSystem *msys = malloc(sizeof(MessageSystem) + max_channels * sizeof(Channel)); + if (msys == NULL) + return NULL; + + if (num_addrs > MESSAGE_SYSTEM_NODE_LIMIT) { + free(msys); + return NULL; + } for (int i = 0; i < num_addrs; i++) msys->addrs[i] = addrs[i]; msys->num_addrs = num_addrs; - // TODO: sort addresses + addr_sort(msys->addrs, msys->num_addrs); - int max_conns = 2 * num_addrs + 1; + msys->max_channels = max_channels; + for (int i = 0; i < max_channels; i++) + msys->channels[i].used = 0; - msys->conns = malloc(max_conns * sizeof(ConnMetadata)); - if (msys->conns == NULL) - return -1; - - msys->max_conns = max_conns; - - for (int i = 0; i < max_conns; i++) { - msys->conns[i].used = false; - msys->conns[i].gen = 0; - msys->conns[i].num_senders = 0; - } - - msys->tcp = tcp_init(max_conns); + msys->tcp = tcp_init(max_channels); if (msys->tcp == NULL) { - free(msys->conns); - return -1; + free(msys); + return NULL; } - return 0; + return msys; } -int message_system_free(MessageSystem *msys) +void message_system_free(MessageSystem *msys) { tcp_free(msys->tcp); - free(msys->conns); - return 0; + free(msys); } int message_system_listen_tcp(MessageSystem *msys, Address addr) { - int ret = tcp_listen_tcp(msys->tcp, S(""), addr.port, true, 128); - if (ret < 0) - return -1; - return 0; -} - -int message_system_listen_tls(MessageSystem *msys, Address addr) -{ - int ret = tcp_listen_tls(msys->tcp, S(""), addr.port, true, 128); + int ret = tcp_listen_tcp(msys->tcp, addr); if (ret < 0) return -1; return 0; } void message_system_process_events(MessageSystem *msys, - void **ptrs, struct pollfd *arr, int num) + void **ptrs, struct pollfd *pfds, int num) { - tcp_process_events(msys->tcp, ptrs, arr, num); + tcp_process_events(msys->tcp, ptrs, pfds, num); } int message_system_register_events(MessageSystem *msys, - void **ptrs, struct pollfd *arr, int cap) + void **ptrs, struct pollfd *pfds, int cap) { - return tcp_register_events(msys->tcp, ptrs, arr, cap); + return tcp_register_events(msys->tcp, ptrs, pfds, cap); +} + +static int +find_free_channel_struct(MessageSystem *msys) +{ + int i = 0; + while (i < msys->max_channels && msys->channels[i].used) + i++; + + if (i == msys->max_channels) + return -1; // No free space + + return i; +} + +static bool has_sender(Channel *channel, int sender_idx) +{ + for (int i = 0; i < channel->num_senders; i++) + if (channel->senders[i] == sender_idx) + return true; + return false; +} + +static void add_sender(Channel *channel, int sender_idx) +{ + if (has_sender(channel, sender_idx)) + return; + channel->senders[channel->num_senders++] = sender_idx; } void *get_next_message(MessageSystem *msys) @@ -80,45 +120,35 @@ void *get_next_message(MessageSystem *msys) if (event.flags & TCP_EVENT_NEW) { - // Skip if already set up by ensure_conn (outbound connection) - if (tcp_get_user_ptr(event.handle) == NULL) { - // Use the TCP connection's slot index so that the - // ConnMetadata index always matches the TCP_Conn index. - // Searching for a "free" metadata slot could assign a - // different index when stale metadata entries remain - // from failed outbound connections. - int i = event.handle.idx; - assert(i >= 0 && i < msys->max_conns); - - ConnMetadata *meta = &msys->conns[i]; - meta->used = true; - meta->gen = event.handle.gen; - meta->num_senders = 0; - meta->message = NULL; - tcp_set_user_ptr(event.handle, meta); + int channel_idx = find_free_channel_struct(msys); + if (channel_idx < 0) { + tcp_close(event.handle); + continue; } + Channel *channel = &msys->channels[channel_idx]; + + channel->used = true; + channel->num_senders = 0; + channel->message = NULL; + channel->handle = event.handle; + + tcp_set_user_ptr(event.handle, channel); } - ConnMetadata *meta = tcp_get_user_ptr(event.handle); - assert(meta); - + Channel *channel = tcp_get_user_ptr(event.handle); if (event.flags & TCP_EVENT_HUP) { - meta->used = false; - tcp_close(event.handle); - continue; + channel->used = false; + tcp_close(channel->handle); // TODO: What if the user is still hanging on the message pointer? } if (!(event.flags & TCP_EVENT_DATA)) continue; - if (meta->message) - continue; // Already processing message - - ByteView buf = tcp_read_buf(event.handle); + string buf = tcp_read_buf(event.handle); Message message; - if (buf.len < sizeof(message)) { + if (buf.len < (int) sizeof(message)) { tcp_read_ack(event.handle, 0); continue; } @@ -129,165 +159,136 @@ void *get_next_message(MessageSystem *msys) assert(0); // TODO } - if (message.length > (uint64_t)buf.len) { + if (message.length > (uint64_t) buf.len) { tcp_read_ack(event.handle, 0); continue; // Still buffering } - // Associate this sender with the TCP connection - if (message.sender < msys->num_addrs) { - bool found = false; - for (int i = 0; i < meta->num_senders; i++) { - if (meta->senders[i] == message.sender) { - found = true; - break; - } - } - if (!found) { - meta->senders[meta->num_senders++] = message.sender; - } - } + add_sender(channel, message.sender); - meta->message = buf.ptr; + channel->message = buf.ptr; return buf.ptr; } return NULL; } -static TCP_Handle find_conn_by_message(MessageSystem *msys, void *message) +static int find_channel_by_message(MessageSystem *msys, void *raw_message) { - for (int i = 0; i < msys->max_conns; i++) { - ConnMetadata *meta = &msys->conns[i]; - if (meta->message == message) - return (TCP_Handle) { msys->tcp, meta->gen, i }; + for (int i = 0; i < msys->max_channels; i++) { + Channel *channel = &msys->channels[i]; + if (!channel->used) + continue; + if (channel->message == raw_message) + return i; } - return (TCP_Handle) {0}; + return -1; } -static TCP_Handle find_conn_by_target(MessageSystem *msys, int target) +int message_length(void *raw_message) { - for (int i = 0; i < msys->max_conns; i++) { - ConnMetadata *meta = &msys->conns[i]; - if (!meta->used) + Message message; + memcpy(&message, raw_message, sizeof(message)); + return message.length; +} + +void consume_message(MessageSystem *msys, void *raw_message) +{ + int channel_idx = find_channel_by_message(msys, raw_message); + if (channel_idx < 0) + return; + Channel *channel = &msys->channels[channel_idx]; + + channel->message = NULL; + tcp_read_ack(channel->handle, message_length(raw_message)); + tcp_mark_ready(channel->handle); +} + +static int find_channel_by_target_idx(MessageSystem *msys, int target_idx) +{ + for (int i = 0; i < msys->max_channels; i++) { + + Channel *channel = &msys->channels[i]; + if (!channel->used) continue; - for (int j = 0; j < meta->num_senders; j++) { - if (meta->senders[j] == target) { - return (TCP_Handle) { msys->tcp, meta->gen, i }; - } + + for (int j = 0; j < channel->num_senders; j++) { + if (channel->senders[j] == target_idx) + return i; } } - return (TCP_Handle) {0}; + + return -1; } -static TCP_Handle ensure_conn(MessageSystem *msys, int target) +void send_message_ex(MessageSystem *msys, int target_idx, + Message *message, void *extra, int extra_len) { - TCP_Handle handle = find_conn_by_target(msys, target); - if (handle.tcp) - return handle; + int channel_idx = find_channel_by_target_idx(msys, target_idx); + if (channel_idx < 0) { - if (target < 0 || target >= msys->num_addrs) - return (TCP_Handle) {0}; + // Find an unused channel struct + channel_idx = find_free_channel_struct(msys); + if (channel_idx < 0) + return; // No free space - int ret = tcp_connect(msys->tcp, false, &msys->addrs[target], 1); - if (ret < 0) - return (TCP_Handle) {0}; + // Establish a new connection + TCP_Handle handle; + if (tcp_connect(msys->tcp, false, &msys->addrs[target_idx], 1, &handle) < 0) + return; - // Find the newly created connection slot and pre-associate with target - for (int i = 0; i < msys->max_conns; i++) { - TCP_Conn *conn = msys->tcp.conns[i]; - if (conn->state == TCP_CONN_STATE_FREE) - continue; - if (conn->user_ptr != NULL) - continue; - if (conn->state != TCP_CONN_STATE_CONNECTING - && conn->state != TCP_CONN_STATE_ESTABLISHED) - continue; + Channel *channel = &msys->channels[channel_idx]; + channel->used = true; + channel->handle = handle; + channel->senders[0] = target_idx; + channel->num_senders = 1; + channel->message = NULL; - ConnMetadata *meta = &msys->conns[i]; - meta->used = true; - meta->gen = conn->gen; - meta->num_senders = 1; - meta->senders[0] = target; - meta->message = NULL; - - TCP_Handle h = { msys->tcp, conn->gen, i }; - tcp_set_user_ptr(h, meta); - return h; + tcp_set_user_ptr(handle, channel); } + Channel *channel = &msys->channels[channel_idx]; - return (TCP_Handle) {0}; + tcp_write(channel->handle, (string) { (void*) message, message->length - extra_len }); + if (extra_len > 0) + tcp_write(channel->handle, (string) { extra, extra_len }); } -void consume_message(MessageSystem *msys, void *ptr) +void send_message(MessageSystem *msys, int target_idx, + Message *message) { - int i = 0; - while (i < msys->max_conns && msys->conns[i].message != ptr) - i++; - - if (i == msys->max_conns) - return; // Not found - - Message message; - memcpy(&message, ptr, sizeof(message)); - - TCP_Handle handle = { msys->tcp, msys->conns[i].gen, i }; - tcp_read_ack(handle, message.length); - tcp_mark_ready(handle); - msys->conns[i].message = NULL; -} - -void send_message(MessageSystem *msys, int target, Message *message) -{ - TCP_Handle handle = ensure_conn(msys, target); - if (!handle.tcp) return; - tcp_write(handle, (string) { (char*) message, message->length }); -} - -void send_message_ex(MessageSystem *msys, int target, - Message *header, void *extra, int extra_len) -{ - TCP_Handle handle = ensure_conn(msys, target); - if (!handle.tcp) return; - int header_size = header->length - extra_len; - tcp_write(handle, (string) { (char*) header, header_size }); - tcp_write(handle, (string) { (char*) extra, extra_len }); -} - -void reply_to_message(MessageSystem *msys, void *incoming_message, - Message *outgoing_message) -{ - TCP_Handle handle = find_conn_by_message(msys, incoming_message); - if (!handle.tcp) return; - tcp_write(handle, (string) { (char*) outgoing_message, outgoing_message->length }); + send_message_ex(msys, target_idx, message, NULL, 0); } void reply_to_message_ex(MessageSystem *msys, void *incoming_message, Message *outgoing_message, void *extra, int extra_len) { - TCP_Handle handle = find_conn_by_message(msys, incoming_message); - if (!handle.tcp) return; + int channel_idx = find_channel_by_message(msys, incoming_message); + if (channel_idx < 0) + return; + Channel *channel = &msys->channels[channel_idx]; + int header_size = outgoing_message->length - extra_len; - tcp_write(handle, (string) { (char*) outgoing_message, header_size }); - tcp_write(handle, (string) { (char*) extra, extra_len }); + tcp_write(channel->handle, (string) { (char*) outgoing_message, header_size }); + if (extra_len > 0) + tcp_write(channel->handle, (string) { (char*) extra, extra_len }); +} + +void reply_to_message(MessageSystem *msys, void *incoming_message, + Message *outgoing_message) +{ + reply_to_message_ex(msys, incoming_message, outgoing_message, NULL, 0); +} + +void broadcast_message_ex(MessageSystem *msys, int self_idx, + Message *message, void *extra, int extra_len) +{ + for (int i = 0; i < msys->num_addrs; i++) { + if (i != self_idx) + send_message_ex(msys, i, message, extra, extra_len); + } } void broadcast_message(MessageSystem *msys, int self_idx, Message *message) { - for (int i = 0; i < msys->num_addrs; i++) { - if (i != self_idx) - send_message(msys, i, message); - } + broadcast_message_ex(msys, self_idx, message, NULL, 0); } - -void broadcast_message_ex(MessageSystem *msys, int self_idx, - Message *header, void *extra, int extra_len) -{ - for (int i = 0; i < msys->num_addrs; i++) { - if (i != self_idx) - send_message_ex(msys, i, header, extra, extra_len); - } -} - - - diff --git a/lib/message.h b/lib/message.h index 021f464..2e2533d 100644 --- a/lib/message.h +++ b/lib/message.h @@ -1,28 +1,11 @@ #ifndef MESSAGE_INCLUDED #define MESSAGE_INCLUDED -#include "tcp.h" +#include -#define MESSAGE_SYSTEM_NODE_LIMIT 8 +#include "basic.h" -typedef struct { - bool used; - uint16_t gen; - int senders[MESSAGE_SYSTEM_NODE_LIMIT]; - int num_senders; - void* message; -} ConnMetadata; - -typedef struct { - - TCP *tcp; - - Address addrs[MESSAGE_SYSTEM_NODE_LIMIT]; - int num_addrs; - - int max_conns; - ConnMetadata *conns; -} MessageSystem; +typedef struct MessageSystem MessageSystem; typedef struct { uint16_t version; @@ -31,22 +14,24 @@ typedef struct { uint64_t length; } Message; -int message_system_init(MessageSystem *msys, - Address *addrs, int num_addrs); +MessageSystem *message_system_init(Address *addrs, int num_addrs); -int message_system_free(MessageSystem *msys); +void message_system_free(MessageSystem *msys); int message_system_listen_tcp(MessageSystem *msys, Address addr); -int message_system_listen_tls(MessageSystem *msys, Address addr); + +struct pollfd; void message_system_process_events(MessageSystem *msys, - void **ptrs, struct pollfd *arr, int num); + void **ptrs, struct pollfd *pfds, int num); int message_system_register_events(MessageSystem *msys, - void **ptrs, struct pollfd *arr, int cap); + void **ptrs, struct pollfd *pfds, int cap); void *get_next_message(MessageSystem *msys); +int message_length(void *raw_message); + void consume_message(MessageSystem *msys, void *ptr); void send_message(MessageSystem *msys, int target, Message *message); diff --git a/lib/tcp.c b/lib/tcp.c index cab785b..229ff09 100644 --- a/lib/tcp.c +++ b/lib/tcp.c @@ -5,6 +5,7 @@ #include #include +#include "tls.h" #include "tcp.h" #ifdef _WIN32 @@ -98,6 +99,10 @@ struct TCP { SOCKET tcp_listen_fd; SOCKET tls_listen_fd; +#ifdef TLS_ENABLED + TLS_Server tls; +#endif + // Total size of the connection array and how many // structures in it are currently in use. int max_conns; @@ -123,11 +128,11 @@ static int set_socket_blocking(SOCKET fd, bool value) { #ifdef _WIN32 u_long mode = !value; - if (ioctlsocket(sock, FIONBIO, &mode) == SOCKET_ERROR) + if (ioctlsocket(fd, FIONBIO, &mode) == SOCKET_ERROR) return -1; return 0; #else - int flags = fcntl(sock, F_GETFL, 0); + int flags = fcntl(fd, F_GETFL, 0); if (flags < 0) return -1; @@ -136,19 +141,35 @@ static int set_socket_blocking(SOCKET fd, bool value) else flags |= O_NONBLOCK; - if (fcntl(sock, F_SETFL, flags) < 0) + if (fcntl(fd, F_SETFL, flags) < 0) return -1; return 0; #endif } +static int bind_2(SOCKET fd, Address addr) +{ + if (addr.is_ipv4) { + struct sockaddr_in buf; + buf.sin_family = AF_INET; + buf.sin_port = htons(addr.port); + memcpy(&buf.sin_addr, &addr.ipv4, sizeof(IPv4)); + return bind(fd, (struct sockaddr*) &buf, sizeof(buf)); + } else { + struct sockaddr_in6 buf; + buf.sin6_family = AF_INET6; + buf.sin6_port = htons(addr.port); + memcpy(&buf.sin6_addr, &addr.ipv6, sizeof(IPv6)); + return bind(fd, (struct sockaddr*) &buf, sizeof(buf)); + } +} + static SOCKET -create_listen_socket(string addr, uint16_t port, - bool reuse_addr, int backlog) +create_listen_socket(Address addr, bool reuse_addr, int backlog) { SOCKET fd = socket(AF_INET, SOCK_STREAM, 0); #ifdef _WIN32 - if (fd == INVALID_SOCKET && WSALastError() == xxx) { + if (fd == INVALID_SOCKET && WSAGetLastError() == WSANOTINITIALISED) { WSADATA wsa; WSAStartup(MAKEWORD(2, 2), &wsa); // TODO: check error fd = socket(AF_INET, SOCK_STREAM, 0); @@ -167,32 +188,11 @@ create_listen_socket(string addr, uint16_t port, int one = 1; setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (void*) &one, sizeof(one)); // TODO: mock this } +#else + (void) reuse_addr; #endif - struct in_addr addr_buf; - if (addr.len == 0) - addr_buf.s_addr = htonl(INADDR_ANY); - else { - - char copy[100]; - if (addr.len >= (int) sizeof(copy)) { - close_socket(fd); - return INVALID_SOCKET; - } - memcpy(copy, addr.ptr, addr.len); - copy[addr.len] = '\0'; - - if (inet_pton(AF_INET, copy, &addr_buf) < 0) { - close_socket(fd); - return INVALID_SOCKET; - } - } - - struct sockaddr_in bind_buf; - bind_buf.sin_family = AF_INET; - bind_buf.sin_addr = addr_buf; - bind_buf.sin_port = htons(port); - if (bind(fd, (struct sockaddr*) &bind_buf, sizeof(bind_buf)) < 0) { + if (bind_2(fd, addr) < 0) { close_socket(fd); return INVALID_SOCKET; } @@ -207,20 +207,19 @@ create_listen_socket(string addr, uint16_t port, static int connect_2(SOCKET fd, Address addr) { - int ret; - if (next_addr.is_ipv4) { + if (addr.is_ipv4) { struct sockaddr_in buf; buf.sin_family = AF_INET; - buf.sin_port = htons(next_addr.port); - STATIC_ASSERT(sizeof(buf.sin_addr) == sizeof(next_addr.ipv4)); - memcpy(&buf.sin_addr, &next_addr.ipv4, sizeof(next_addr.ipv4)); + buf.sin_port = htons(addr.port); + STATIC_ASSERT(sizeof(buf.sin_addr) == sizeof(addr.ipv4)); + memcpy(&buf.sin_addr, &addr.ipv4, sizeof(addr.ipv4)); return connect(fd, (struct sockaddr*) &buf, sizeof(buf)); } else { struct sockaddr_in6 buf; buf.sin6_family = AF_INET; - buf.sin6_port = htons(next_addr.port); - STATIC_ASSERT(sizeof(buf.sin6_addr) == sizeof(next_addr.ipv6)); - memcpy(&buf.sin6_addr, &next_addr.ipv6, sizeof(next_addr.ipv6)); + buf.sin6_port = htons(addr.port); + STATIC_ASSERT(sizeof(buf.sin6_addr) == sizeof(addr.ipv6)); + memcpy(&buf.sin6_addr, &addr.ipv6, sizeof(addr.ipv6)); return connect(fd, (struct sockaddr*) &buf, sizeof(buf)); } } @@ -276,13 +275,17 @@ void tcp_free(TCP *tcp) } // See tcp.h -int tcp_listen_tcp(TCP *tcp, string addr, uint16_t port) +int tcp_listen_tcp(TCP *tcp, Address addr) { // Ensure plaintext server mode wasn't enabled already. if (tcp->tcp_listen_fd != INVALID_SOCKET) return -1; - SOCKET fd = create_listen_socket(addr, port, reuse_addr, backlog); + // TODO: Make these configurable + bool reuse_addr = false; + int backlog = 32; + + SOCKET fd = create_listen_socket(addr, reuse_addr, backlog); if (fd == INVALID_SOCKET) return -1; @@ -291,18 +294,22 @@ int tcp_listen_tcp(TCP *tcp, string addr, uint16_t port) } // See tcp.h -int tcp_listen_tls(TCP *tcp, string addr, uint16_t port, string cert_file, string key_file) +int tcp_listen_tls(TCP *tcp, Address addr, string cert_file, string key_file) { #ifdef TLS_ENABLED // Ensure plaintext server mode wasn't enabled already. if (tcp->tls_listen_fd != INVALID_SOCKET) return -1; - SOCKET fd = create_listen_socket(addr, port, reuse_addr, backlog); + // TODO: Make these configurable + bool reuse_addr = false; + int backlog = 32; + + SOCKET fd = create_listen_socket(addr, reuse_addr, backlog); if (fd == INVALID_SOCKET) return -1; - if (tls_server_init() < 0) { + if (tls_server_init(&tcp->tls, cert_file, key_file) < 0) { close_socket(fd); return -1; } @@ -312,9 +319,8 @@ int tcp_listen_tls(TCP *tcp, string addr, uint16_t port, string cert_file, strin #else (void) tcp; (void) addr; - (void) port; - (void) reuse_addr; - (void) backlog; + (void) cert_file; + (void) key_file; return -1; #endif } @@ -323,7 +329,7 @@ int tcp_listen_tls(TCP *tcp, string addr, uint16_t port, string cert_file, strin int tcp_add_cert(TCP *tcp, string domain, string cert_file, string key_file) { #ifdef TLS_ENABLED - int ret = tcp_server_add_cert(&tcp->tls, domain, cert_file, cert_key); + int ret = tls_server_add_cert(&tcp->tls, domain, cert_file, key_file); if (ret < 0) return -1; return 0; @@ -336,7 +342,7 @@ int tcp_add_cert(TCP *tcp, string domain, string cert_file, string key_file) #endif } -static void tcp_conn_init(TCP *tcp, TCP_Conn *conn) +static void tcp_conn_init(TCP *tcp, TCP_Conn *conn, bool secure, TCP_ConnState state, SOCKET fd) { conn->state = state; conn->flags = 0; @@ -382,15 +388,15 @@ static void tcp_conn_set_addrs(TCP_Conn *conn, conn->num_addrs = num_addrs; } -static ByteView tcp_conn_write_buf(TCP_Conn *conn) +static string tcp_conn_write_buf(TCP_Conn *conn) { #ifdef TLS_ENABLED if (conn->flags & TCP_CONN_FLAG_SECURE) { int cap; char *ptr = tls_conn_net_write_buf(&conn->tls, &cap); if (ptr == NULL) - return (ByteView) {0}; - return (ByteView) { (uint8_t*) ptr, cap }; + return (string) {0}; + return (string) { ptr, cap }; } #endif @@ -406,7 +412,7 @@ static int tcp_conn_write_ack(TCP_Conn *conn, int num) tls_conn_net_write_ack(&conn->tls, num); for (bool done = false; !done; ) { byte_queue_write_setmincap(&conn->input, MIN_RECV); - ByteView buf = byte_queue_write_buf(&conn->input); + string buf = byte_queue_write_buf(&conn->input); int n = tls_conn_app_read(&conn->tls, (char*) buf.ptr, buf.len); if (n <= 0) { if (n < 0) { @@ -431,7 +437,7 @@ static int tcp_conn_write_ack(TCP_Conn *conn, int num) static void tcp_conn_tls_encrypt_output(TCP_Conn *conn) { while (!byte_queue_empty(&conn->output)) { - ByteView src = byte_queue_read_buf(&conn->output); + string src = byte_queue_read_buf(&conn->output); if (!src.ptr || src.len == 0) { byte_queue_read_ack(&conn->output, 0); break; @@ -446,7 +452,7 @@ static void tcp_conn_tls_encrypt_output(TCP_Conn *conn) } #endif -static ByteView tcp_conn_read_buf(TCP_Conn *conn) +static string tcp_conn_read_buf(TCP_Conn *conn) { #ifdef TLS_ENABLED if (conn->flags & TCP_CONN_FLAG_SECURE) { @@ -454,8 +460,8 @@ static ByteView tcp_conn_read_buf(TCP_Conn *conn) int n; char *ptr = tls_conn_net_read_buf(&conn->tls, &n); if (ptr == NULL) - return (ByteView) {0}; - return (ByteView) { (uint8_t*) ptr, n }; + return (string) {0}; + return (string) { ptr, n }; } #endif @@ -500,7 +506,7 @@ static bool tcp_conn_is_buffering(TCP_Conn *conn) static bool tcp_conn_free_maybe(TCP_Conn *conn) { - if (!conn->handled && conn->fd < 0) { + if (!conn->handled && conn->fd == INVALID_SOCKET) { tcp_conn_free(conn); return true; } else { @@ -515,6 +521,32 @@ static void tcp_conn_invalidate_handles(TCP_Conn *conn) conn->gen = 1; } +static TCP_Handle conn_to_handle(TCP *tcp, TCP_Conn *conn) +{ + TCP_Handle handle = { + .tcp=tcp, + .gen=conn->gen, + .idx=conn - tcp->conns, + }; + return handle; +} + +static TCP_Conn *handle_to_conn(TCP_Handle handle) +{ + if (handle.tcp == NULL) + return NULL; + TCP *tcp = handle.tcp; + + if (handle.idx < 0 || handle.idx >= tcp->max_conns) + return NULL; + TCP_Conn *conn = &tcp->conns[handle.idx]; + + if (conn->state == TCP_CONN_STATE_FREE || conn->gen != handle.gen) + return NULL; + + return conn; +} + static int find_free_conn_struct(TCP *tcp) { if (tcp->num_conns == tcp->max_conns) @@ -533,13 +565,13 @@ static int find_free_conn_struct(TCP *tcp) static bool connect_in_progress(void) { -#if defined(_WIN32) -#if defined(QUAKEY_ENABLE_MOCKS) +#ifdef _WIN32 +#ifdef QUAKEY_ENABLE_MOCKS assert(0); // TODO: The mock WSA function must use WSASetLastError #endif - return (WSAGetLastError() == WSAEWOULDBLOCK); + return WSAGetLastError() == WSAEWOULDBLOCK; #else - return (errno == EINPROGRESS); + return errno == EINPROGRESS; #endif } @@ -556,7 +588,7 @@ int tcp_connect(TCP *tcp, bool secure, Address *addrs, int num_addrs, TCP_Handle SOCKET fd = socket(AF_INET, SOCK_STREAM, 0); #ifdef _WIN32 - if (fd == INVALID_SOCKET && WSALastError() == xxx) { + if (fd == INVALID_SOCKET && WSAGetLastError() == WSANOTINITIALISED) { WSADATA wsa; WSAStartup(MAKEWORD(2, 2), &wsa); // TODO: check error fd = socket(AF_INET, SOCK_STREAM, 0); @@ -627,7 +659,7 @@ static int restart_connect(TCP_Conn *conn) conn->addr_idx++; if (conn->addr_idx == conn->num_addrs) return -1; // No more addresses to try - Address next_addr = &conn->addrs[conn->addr_idx]; + Address next_addr = conn->addrs[conn->addr_idx]; // Elsewhere in this file calls to socket() are // followed by the initialization of the winsock2 @@ -644,6 +676,7 @@ static int restart_connect(TCP_Conn *conn) return -1; } + TCP_ConnState state; int ret = connect_2(fd, next_addr); if (ret == 0) { if (conn->flags & TCP_CONN_FLAG_SECURE) { @@ -675,9 +708,9 @@ int tcp_register_events(TCP *tcp, void **ptrs, struct pollfd *pfds, int cap) if (tcp->tcp_listen_fd != INVALID_SOCKET) { if (tcp->num_conns < tcp->max_conns) { - arr[ret].fd = tcp->tcp_listen_fd; - arr[ret].events = POLLIN; - arr[ret].revents = 0; + pfds[ret].fd = tcp->tcp_listen_fd; + pfds[ret].events = POLLIN; + pfds[ret].revents = 0; ptrs[ret] = NULL; ret++; } @@ -685,9 +718,9 @@ int tcp_register_events(TCP *tcp, void **ptrs, struct pollfd *pfds, int cap) if (tcp->tls_listen_fd != INVALID_SOCKET) { if (tcp->num_conns < tcp->max_conns) { - arr[ret].fd = tcp->tls_listen_fd; - arr[ret].events = POLLIN; - arr[ret].revents = 0; + pfds[ret].fd = tcp->tls_listen_fd; + pfds[ret].events = POLLIN; + pfds[ret].revents = 0; ptrs[ret] = NULL; ret++; } @@ -712,9 +745,9 @@ int tcp_register_events(TCP *tcp, void **ptrs, struct pollfd *pfds, int cap) events |= POLLOUT; if (events) { - arr[ret].fd = conn->fd; - arr[ret].events = events; - arr[ret].revents = 0; + pfds[ret].fd = conn->fd; + pfds[ret].events = events; + pfds[ret].revents = 0; ptrs[ret] = conn; ret++; } @@ -755,7 +788,20 @@ accept_incoming_conns(TCP *tcp, SOCKET listen_fd) conn->events |= TCP_EVENT_NEW; tcp->num_conns++; - return 0; +} + +static bool would_block(void) +{ +#ifdef _WIN32 +#ifdef QUAKEY_ENABLE_MOCKS + assert(0); // TODO: The mock WSA function must use WSASetLastError +#endif + return WSAGetLastError() == WSAEWOULDBLOCK; +#else + return errno == EWOULDBLOCK + || errno == EAGAIN + || errno == EINTR; +#endif } // Returns true if the connection should be closed @@ -763,7 +809,7 @@ static bool read_from_net_into_conn(TCP_Conn *conn) { bool defer_close = false; - ByteView buf = tcp_conn_write_buf(conn); + string buf = tcp_conn_write_buf(conn); int n = recv(conn->fd, (char*) buf.ptr, buf.len, 0); if (n == 0) { defer_close = true; @@ -784,7 +830,7 @@ static bool write_from_conn_into_net(TCP_Conn *conn) { bool defer_close = false; - ByteView buf = tcp_conn_read_buf(conn); + string buf = tcp_conn_read_buf(conn); int n = send(conn->fd, (char*) buf.ptr, buf.len, 0); if (n < 0) { if (!would_block()) @@ -797,7 +843,7 @@ write_from_conn_into_net(TCP_Conn *conn) return defer_close; } -static void process_conn_events(TCP_Conn *conn, int revents) +static void process_conn_events(TCP *tcp, TCP_Conn *conn, int revents) { bool defer_close = false; bool defer_connect = false; @@ -822,64 +868,70 @@ static void process_conn_events(TCP_Conn *conn, int revents) conn->state = TCP_CONN_STATE_HANDSHAKE; } else { conn->state = TCP_CONN_STATE_ESTABLISHED; - conn->events |= TCP_EVENT_NEW; } } - case TCP_CONN_STATE_HANDSHAKE: - case TCP_CONN_STATE_ACCEPTING: -#ifdef TLS_ENABLED - { - if (revents & POLLIN) { - defer_close = read_from_net_into_conn(conn); - } - if (revents & POLLOUT) { - defer_close = write_from_conn_into_net(conn); - } - int ret = tls_conn_handshake(&conn->tls); - if (ret == -1) { - defer_close = true; - break; - } - - if (ret == 1) { - conn->state = TCP_CONN_STATE_ESTABLISHED; - conn->events |= TCP_EVENT_NEW; - - // Decrypt any application data already in the BIO - for (;;) { - byte_queue_write_setmincap(&conn->input, MIN_RECV); - ByteView buf = byte_queue_write_buf(&conn->input); - if (buf.ptr == NULL) - break; - int n = tls_conn_app_read(&conn->tls, (char*) buf.ptr, buf.len); - if (n <= 0) { - byte_queue_write_ack(&conn->input, 0); - break; - } - byte_queue_write_ack(&conn->input, n); - conn->events |= TCP_EVENT_DATA; - } - } - } -#endif // TLS_ENABLED - break; - case TCP_CONN_STATE_ESTABLISHED: - { - if (revents & POLLIN) { - defer_close = read_from_net_into_conn(conn); - } - if (revents & POLLOUT) { - defer_close = write_from_conn_into_net(conn); - } - } - break; - case TCP_CONN_STATE_SHUTDOWN: - { - // TODO - } - break; } break; + case TCP_CONN_STATE_HANDSHAKE: + case TCP_CONN_STATE_ACCEPTING: +#ifdef TLS_ENABLED + { + if (revents & POLLIN) { + defer_close = read_from_net_into_conn(conn); + } + if (revents & POLLOUT) { + defer_close = write_from_conn_into_net(conn); + } + int ret = tls_conn_handshake(&conn->tls); + if (ret == -1) { + defer_close = true; + break; + } + + if (ret == 1) { + conn->state = TCP_CONN_STATE_ESTABLISHED; + + // Don't set the NEW flag if the connection was + // started by us + if (conn->num_addrs > 0) { + conn->events |= TCP_EVENT_NEW; + } + + // Decrypt any application data already in the BIO + for (;;) { + byte_queue_write_setmincap(&conn->input, MIN_RECV); + string buf = byte_queue_write_buf(&conn->input); + if (buf.ptr == NULL) + break; + int n = tls_conn_app_read(&conn->tls, (char*) buf.ptr, buf.len); + if (n <= 0) { + byte_queue_write_ack(&conn->input, 0); + break; + } + byte_queue_write_ack(&conn->input, n); + conn->events |= TCP_EVENT_DATA; + } + } + } +#endif // TLS_ENABLED + break; + case TCP_CONN_STATE_ESTABLISHED: + { + if (revents & POLLIN) { + defer_close = read_from_net_into_conn(conn); + } + if (revents & POLLOUT) { + defer_close = write_from_conn_into_net(conn); + } + } + break; + case TCP_CONN_STATE_SHUTDOWN: + { + // TODO + } + break; + default: + UNREACHABLE; } if (defer_connect) { @@ -913,37 +965,11 @@ void tcp_process_events(TCP *tcp, void **ptrs, struct pollfd *pfds, int num) } } else { TCP_Conn *conn = ptrs[i]; - process_conn_events(conn, pfds[i].revents); + process_conn_events(tcp, conn, pfds[i].revents); } } } -static TCP_Handle conn_to_handle(TCP *tcp, TCP_Conn *conn) -{ - TCP_Handle handle = { - .tcp=tcp, - .gen=conn->gen, - .idx=conn - tcp->conns, - }; - return handle; -} - -static TCP_Conn *handle_to_conn(TCP_Handle handle) -{ - if (handle.tcp == NULL) - return NULL; - TCP *tcp = handle.tcp; - - if (handle.idx < 0 || handle.idx >= tcp->max_conns) - return NULL; - TCP_Conn *conn = &tcp->conns[handle.idx]; - - if (conn->state == TCP_CONN_STATE_FREE || conn->gen != handle.gen) - return NULL; - - return conn; -} - static bool conn_to_event(TCP *tcp, TCP_Conn *conn, TCP_Event *event) { @@ -1030,17 +1056,20 @@ TCP_Offset tcp_write_off(TCP_Handle handle) // See tcp.h void tcp_write(TCP_Handle handle, string data) { - while (str.len > 0) { - TCP_Conn *conn = handle_to_conn(handle); - if (conn == NULL) - return; - byte_queue_write_setmincap(&conn->output, str.len); + TCP_Conn *conn = handle_to_conn(handle); + if (conn == NULL) + return; + + while (data.len > 0) { + byte_queue_write_setmincap(&conn->output, data.len); string buf = tcp_write_buf(handle); - int num = MIN(buf.len, str.len); - memcpy(buf.ptr, str.ptr, num); + + int num = MIN(buf.len, data.len); + memcpy(buf.ptr, data.ptr, num); + tcp_write_ack(handle, num); - str.ptr += num; - str.len -= num; + data.ptr += num; + data.len -= num; } } @@ -1051,7 +1080,7 @@ void tcp_patch(TCP_Handle handle, TCP_Offset offset, string data) if (conn == NULL) return; - byte_queue_patch(&conn->output, offset, src, len); + byte_queue_patch(&conn->output, offset, data.ptr, data.len); } // See tcp.h @@ -1087,7 +1116,7 @@ void tcp_set_user_ptr(TCP_Handle handle, void *user_ptr) if (conn == NULL) return; - conn->user_ptr = ptr; + conn->user_ptr = user_ptr; } // See tcp.h diff --git a/lib/tcp.h b/lib/tcp.h index 07bd0f3..6d01a1d 100644 --- a/lib/tcp.h +++ b/lib/tcp.h @@ -32,7 +32,7 @@ void tcp_free(TCP *tcp); // Enable a listening interface for this TCP pool. Connections accepted via // this interface will be plaintext. -int tcp_listen_tcp(TCP *tcp, string addr, uint16_t port); +int tcp_listen_tcp(TCP *tcp, Address addr); // Enable a listening interface for this TCP pool. Connections accepted via // this interface will be encrypted. A single TCP pool may be configured for @@ -41,7 +41,7 @@ int tcp_listen_tcp(TCP *tcp, string addr, uint16_t port); // transparent. // The cert_file and key_file parameters refer to the certificate file and // associated private key file to use for encryption, both in PEM format. -int tcp_listen_tls(TCP *tcp, string addr, uint16_t port, string cert_file, string key_file); +int tcp_listen_tls(TCP *tcp, Address addr, string cert_file, string key_file); // If the TCP pool is configured in TLS mode (tcp_listen_tls was called), this // function can be used to add an additional certificate. Connecting sockets @@ -49,12 +49,20 @@ int tcp_listen_tls(TCP *tcp, string addr, uint16_t port, string cert_file, strin // are expecting to talk to. int tcp_add_cert(TCP *tcp, string domain, string cert_file, string key_file); +// Handle structure representing a TCP connection of the TCP pool. The contents +// should not be interpreted by users. +typedef struct { + TCP *tcp; + int idx; + int gen; +} TCP_Handle; + // Add a connection to the TCP pool by establishing one towards the specified // peer. The addrs array (of size num_addrs) contains the list of IP addresses // for the host. The TCP pool will try each address one by one until a connection // is established. If the secure argument is true, the connection will be // encrypted. -int tcp_connect(TCP *tcp, bool secure, Address *addrs, int num_addrs); +int tcp_connect(TCP *tcp, bool secure, Address *addrs, int num_addrs, TCP_Handle *handle); // Forward-declare poll item type. The user must include poll.h (Linux) or // winsock2.h (Windows) to get this definition (and the definition of poll() @@ -76,14 +84,6 @@ int tcp_register_events(TCP *tcp, void **ptrs, struct pollfd *pfds, int cap); // be passed in as it was initialized by the tcp_register_events as-is. void tcp_process_events(TCP *tcp, void **ptrs, struct pollfd *pfds, int num); -// Handle structure representing a TCP connection of the TCP pool. The contents -// should not be interpreted by users. -typedef struct { - TCP *tcp; - int idx; - int gen; -} TCP_Handle; - // Flags for the "flags" field in TCP_Event. enum { TCP_EVENT_NEW = 1<<0, diff --git a/lib/tls_schannel.c b/lib/tls_schannel.c index 47a18d3..4c5e6e2 100644 --- a/lib/tls_schannel.c +++ b/lib/tls_schannel.c @@ -387,7 +387,7 @@ int tls_conn_handshake(TLS_Conn *conn) assert(conn->handshake); // Read available ciphertext from in_buf - ByteView in = byte_queue_read_buf(&conn->in_buf); + string in = byte_queue_read_buf(&conn->in_buf); if (!in.ptr || in.len == 0) { byte_queue_read_ack(&conn->in_buf, 0); return 0; @@ -484,7 +484,7 @@ int tls_conn_handshake(TLS_Conn *conn) char *tls_conn_net_write_buf(TLS_Conn *conn, int *cap) { byte_queue_write_setmincap(&conn->in_buf, 4096); - ByteView bv = byte_queue_write_buf(&conn->in_buf); + string bv = byte_queue_write_buf(&conn->in_buf); if (!bv.ptr || bv.len == 0) { byte_queue_write_ack(&conn->in_buf, 0); return NULL; @@ -500,7 +500,7 @@ void tls_conn_net_write_ack(TLS_Conn *conn, int num) char *tls_conn_net_read_buf(TLS_Conn *conn, int *num) { - ByteView bv = byte_queue_read_buf(&conn->out_buf); + string bv = byte_queue_read_buf(&conn->out_buf); if (!bv.ptr || bv.len == 0) { byte_queue_read_ack(&conn->out_buf, 0); return NULL; @@ -534,7 +534,7 @@ int tls_conn_app_write(TLS_Conn *conn, char *src, int num) // Ensure output buffer has enough space byte_queue_write_setmincap(&conn->out_buf, total); - ByteView bv = byte_queue_write_buf(&conn->out_buf); + string bv = byte_queue_write_buf(&conn->out_buf); if (!bv.ptr || (int) bv.len < total) { // Try with less data if (!bv.ptr || (int) bv.len < header_size + trailer_size + 1) { @@ -604,7 +604,7 @@ int tls_conn_app_read(TLS_Conn *conn, char *dst, int cap) return n; } - ByteView in = byte_queue_read_buf(&conn->in_buf); + string in = byte_queue_read_buf(&conn->in_buf); if (!in.ptr || in.len == 0) { byte_queue_read_ack(&conn->in_buf, 0); return 0; diff --git a/src/client.c b/src/client.c index 50d7943..ef4f2b7 100644 --- a/src/client.c +++ b/src/client.c @@ -46,7 +46,7 @@ typedef struct { struct ToastyFS { - MessageSystem msys; + MessageSystem *msys; Address server_addrs[NODE_LIMIT]; int num_servers; @@ -148,7 +148,8 @@ ToastyFS *toastyfs_init(uint64_t client_id, char **addrs, int num_addrs) tfs->num_servers = num_addrs; addr_sort(tfs->server_addrs, tfs->num_servers); - if (message_system_init(&tfs->msys, tfs->server_addrs, num_addrs) < 0) { + tfs->msys = message_system_init(tfs->server_addrs, num_addrs); + if (tfs->msys == NULL) { free(tfs); return NULL; } @@ -162,7 +163,7 @@ ToastyFS *toastyfs_init(uint64_t client_id, char **addrs, int num_addrs) void toastyfs_free(ToastyFS *tfs) { - message_system_free(&tfs->msys); + message_system_free(tfs->msys); free(tfs->put_data); } @@ -232,7 +233,7 @@ static int begin_transfers(ToastyFS *tfs) .hash = tfs->transfers[i].hash, .size = tfs->transfers[i].size, }; - send_message_ex(&tfs->msys, tfs->transfers[i].location, + send_message_ex(tfs->msys, tfs->transfers[i].location, &msg.base, tfs->transfers[i].data, tfs->transfers[i].size); } else { FetchChunkMessage msg = { @@ -244,7 +245,7 @@ static int begin_transfers(ToastyFS *tfs) .hash = tfs->transfers[i].hash, .sender_idx = -1, // Client (not a peer server) }; - send_message(&tfs->msys, tfs->transfers[i].location, &msg.base); + send_message(tfs->msys, tfs->transfers[i].location, &msg.base); } tfs->transfers[i].state = TRANSFER_STARTED; @@ -311,7 +312,7 @@ static void replay_request(ToastyFS *tfs) for (int j = 0; j < tfs->chunk_num_holders[i]; j++) msg.oper.chunks[i].holders[j] = tfs->chunk_holders[i][j]; } - send_message(&tfs->msys, leader_idx(tfs), &msg.base); + send_message(tfs->msys, leader_idx(tfs), &msg.base); } break; case STEP_DELETE: @@ -330,7 +331,7 @@ static void replay_request(ToastyFS *tfs) }; memcpy(msg.oper.bucket, tfs->bucket, META_BUCKET_MAX); memcpy(msg.oper.key, tfs->key, META_KEY_MAX); - send_message(&tfs->msys, leader_idx(tfs), &msg.base); + send_message(tfs->msys, leader_idx(tfs), &msg.base); } break; case STEP_GET: @@ -344,7 +345,7 @@ static void replay_request(ToastyFS *tfs) }; memcpy(msg.bucket, tfs->bucket, META_BUCKET_MAX); memcpy(msg.key, tfs->key, META_KEY_MAX); - send_message(&tfs->msys, leader_idx(tfs), &msg.base); + send_message(tfs->msys, leader_idx(tfs), &msg.base); } break; default: @@ -352,7 +353,7 @@ static void replay_request(ToastyFS *tfs) } } -static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg) +static int process_message(ToastyFS *tfs, uint16_t type, string msg) { // Discard messages that arrive after the operation has completed or // before a new one has started. This handles late/stale responses @@ -370,7 +371,7 @@ static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg) if (type == MESSAGE_TYPE_FETCH_CHUNK_RESPONSE) { FetchChunkResponseMessage resp; - if (msg.len < sizeof(FetchChunkResponseMessage)) + if (msg.len < (int) sizeof(FetchChunkResponseMessage)) return -1; memcpy(&resp, msg.ptr, sizeof(resp)); char *data = (char *)msg.ptr + sizeof(resp); @@ -475,7 +476,7 @@ static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg) for (int j = 0; j < tfs->chunk_num_holders[i]; j++) msg.oper.chunks[i].holders[j] = tfs->chunk_holders[i][j]; } - send_message(&tfs->msys, leader_idx(tfs), &msg.base); + send_message(tfs->msys, leader_idx(tfs), &msg.base); tfs->step = STEP_COMMIT; client_log(tfs, "SEND COMMIT_PUT", "key=%s chunks=%d req=%lu", tfs->key, tfs->num_chunks, tfs->request_id); @@ -670,14 +671,14 @@ static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg) void toastyfs_process_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, int pnum) { - message_system_process_events(&tfs->msys, ctxs, pdata, pnum); + message_system_process_events(tfs->msys, ctxs, pdata, pnum); void *raw_message; - while ((raw_message = get_next_message(&tfs->msys)) != NULL) { + while ((raw_message = get_next_message(tfs->msys)) != NULL) { Message *header = (Message *)raw_message; - ByteView msg_view = { .ptr = raw_message, .len = header->length }; + string msg_view = { .ptr = raw_message, .len = header->length }; process_message(tfs, header->type, msg_view); - consume_message(&tfs->msys, raw_message); + consume_message(tfs->msys, raw_message); } // Check for operation timeout -- retry the current operation if the @@ -728,7 +729,7 @@ int toastyfs_register_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, i *timeout = deadline_to_timeout(deadline, now); if (pcap < POLL_CAPACITY) return -1; - return message_system_register_events(&tfs->msys, ctxs, pdata, pcap); + return message_system_register_events(tfs->msys, ctxs, pdata, pcap); } static void @@ -834,7 +835,7 @@ int toastyfs_async_get(ToastyFS *tfs, char *key, int key_len) tfs->step_time = get_current_time(); // TODO: Handle INVALID_TIME error tfs->step = STEP_GET; client_log(tfs, "ASYNC GET", "key=%s leader=%d", tfs->key, leader_idx(tfs)); - send_message(&tfs->msys, leader_idx(tfs), &msg.base); + send_message(tfs->msys, leader_idx(tfs), &msg.base); return 0; } @@ -868,7 +869,7 @@ int toastyfs_async_delete(ToastyFS *tfs, char *key, int key_len) tfs->step_time = get_current_time(); // TODO: Handle INVALID_TIME error tfs->step = STEP_DELETE; client_log(tfs, "ASYNC DELETE", "key=%s req=%lu leader=%d", tfs->key, tfs->request_id, leader_idx(tfs)); - send_message(&tfs->msys, leader_idx(tfs), &msg.base); + send_message(tfs->msys, leader_idx(tfs), &msg.base); return 0; } diff --git a/src/http_proxy.c b/src/http_proxy.c index e68535f..c864bc4 100644 --- a/src/http_proxy.c +++ b/src/http_proxy.c @@ -39,8 +39,7 @@ int http_proxy_init(void *state, int argc, char **argv, // TODO: Make these configurable int max_opers = 128; - string http_addr = S("127.0.0.1"); - uint16_t http_port = 3000; + char *http_addr = "127.0.0.1:3000"; uint64_t client_id = 999; proxy->max_opers = max_opers; @@ -56,7 +55,13 @@ int http_proxy_init(void *state, int argc, char **argv, return -1; } - if (http_server_listen_tcp(&proxy->http_server, http_addr, http_port) < 0) { + Address http_addr_2; + if (parse_addr_arg(http_addr, &http_addr_2) < 0) { + http_server_free(&proxy->http_server); + free(proxy->opers); + return -1; + } + if (http_server_listen_tcp(&proxy->http_server, http_addr_2) < 0) { http_server_free(&proxy->http_server); free(proxy->opers); return -1; diff --git a/src/server.c b/src/server.c index fab40ea..caa93af 100644 --- a/src/server.c +++ b/src/server.c @@ -97,7 +97,7 @@ static void add_vote(uint32_t *votes, int idx) *votes |= 1 << idx; } -// Helper: read a message from raw_message into ByteView-like usage +// Helper: read a message from raw_message into string-like usage // raw_message is a pointer to the message bytes in the TCP read buffer static uint64_t msg_len(void *raw_message) { @@ -123,7 +123,7 @@ static void begin_state_transfer(Server *state, int sender_idx) .op_number = state->log.count, .sender_idx = self_idx(state), }; - send_message(&state->msys,sender_idx, &message.base); + send_message(state->msys,sender_idx, &message.base); node_log(state, "SEND GET_STATE", "to=%d op=%d", sender_idx, state->log.count); } @@ -165,7 +165,7 @@ process_request(Server *state, void *raw_message) .rejected = true, .request_id = request_message.request_id, }; - reply_to_message(&state->msys, raw_message, &reply_message.base); + reply_to_message(state->msys, raw_message, &reply_message.base); node_log(state, "SEND REPLY", "client=%llu REJECTED (table full)", request_message.client_id); return HR_OK; } @@ -190,7 +190,7 @@ process_request(Server *state, void *raw_message) .result = entry->last_result, .request_id = request_message.request_id, }; - reply_to_message(&state->msys, raw_message, &reply_message.base); + reply_to_message(state->msys, raw_message, &reply_message.base); { char result_buf[64]; @@ -230,7 +230,7 @@ process_request(Server *state, void *raw_message) .client_id = request_message.client_id, .request_id = request_message.request_id, }; - broadcast_message(&state->msys, self_idx(state), &prepare_message.base); + broadcast_message(state->msys, self_idx(state), &prepare_message.base); { char oper_buf[128]; @@ -261,8 +261,8 @@ static void reply_to_client(Server *state, ClientTableEntry *table_entry, .result = result, .request_id = request_id, }; - reply_to_message(&state->msys, table_entry->pending_message, &message.base); - consume_message(&state->msys, table_entry->pending_message); + reply_to_message(state->msys, table_entry->pending_message, &message.base); + consume_message(state->msys, table_entry->pending_message); table_entry->pending_message = NULL; } } @@ -393,7 +393,7 @@ complete_view_change_and_become_primary(Server *state) .commit_index = state->commit_index, .op_number = state->log.count, }; - broadcast_message_ex(&state->msys, self_idx(state),&begin_view_message.base, state->log.entries, state->log.count * sizeof(LogEntry)); + broadcast_message_ex(state->msys, self_idx(state),&begin_view_message.base, state->log.entries, state->log.count * sizeof(LogEntry)); node_log(state, "SEND BEGIN_VIEW", "to=* view=%llu log=%d commit=%d", state->view_number, state->log.count, state->commit_index); @@ -478,10 +478,10 @@ process_recovery(Server *state, void *raw_message) }; if (is_primary(state)) { recovery_response_message.base.length += state->log.count * sizeof(LogEntry); - send_message_ex(&state->msys,recovery_message.sender_idx, &recovery_response_message.base, + send_message_ex(state->msys,recovery_message.sender_idx, &recovery_response_message.base, state->log.entries, state->log.count * sizeof(LogEntry)); } else { - send_message(&state->msys,recovery_message.sender_idx, &recovery_response_message.base); + send_message(state->msys,recovery_message.sender_idx, &recovery_response_message.base); } return HR_OK; } @@ -511,7 +511,7 @@ perform_log_transfer_for_view_change(Server *state) .commit_index = state->commit_index, .sender_idx = self_idx(state), }; - send_message_ex(&state->msys,primary_idx(state), &do_view_change_message.base, state->log.entries, state->log.count * sizeof(LogEntry)); + send_message_ex(state->msys,primary_idx(state), &do_view_change_message.base, state->log.entries, state->log.count * sizeof(LogEntry)); node_log(state, "SEND DO_VIEW_CHANGE", "to=%d view=%llu old_view=%llu log=%d commit=%d", primary_idx(state), state->view_number, state->last_normal_view, state->log.count, state->commit_index); @@ -553,7 +553,7 @@ process_begin_view_change(Server *state, void *raw_message) .view_number = message.view_number, .sender_idx = self_idx(state), }; - broadcast_message(&state->msys, self_idx(state),&message_2.base); + broadcast_message(state->msys, self_idx(state),&message_2.base); node_log(state, "SEND BEGIN_VIEW_CHG", "to=* view=%llu", message.view_number); clear_view_change_fields(state); @@ -621,7 +621,7 @@ process_begin_view(Server *state, void *raw_message) .log_index = state->log.count - 1, .view_number = state->view_number, }; - send_message(&state->msys,primary_idx(state), &ok_msg.base); + send_message(state->msys,primary_idx(state), &ok_msg.base); node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->log.count - 1, state->log.entries[state->log.count - 1].oper.bucket, state->log.entries[state->log.count - 1].oper.key); } @@ -669,7 +669,7 @@ process_get_state(Server *state, void *raw_message) .commit_index = state->commit_index, .start_index = start, }; - send_message_ex(&state->msys,get_state_message.sender_idx, &new_state_message.base, + send_message_ex(state->msys,get_state_message.sender_idx, &new_state_message.base, state->log.entries + start, num_entries * sizeof(LogEntry)); node_log(state, "SEND NEW_STATE", "to=%d entries=%d commit=%d", get_state_message.sender_idx, num_entries, state->commit_index); @@ -707,7 +707,7 @@ process_store_chunk(Server *state, void *raw_message) .hash = message.hash, .success = (ret == 0), }; - reply_to_message(&state->msys, raw_message, &ack.base); + reply_to_message(state->msys, raw_message, &ack.base); node_log(state, "RECV STORE_CHUNK", "size=%u ok=%d", message.size, ret == 0); return HR_OK; @@ -735,7 +735,7 @@ process_fetch_chunk(Server *state, void *raw_message) .hash = message.hash, .size = 0, }; - reply_to_message(&state->msys, raw_message, &response.base); + reply_to_message(state->msys, raw_message, &response.base); node_log(state, "RECV FETCH_CHUNK", "NOT_FOUND"); return HR_OK; @@ -762,7 +762,7 @@ process_fetch_chunk(Server *state, void *raw_message) .hash = message.hash, .size = 0, }; - reply_to_message(&state->msys, raw_message, &response.base); + reply_to_message(state->msys, raw_message, &response.base); node_log(state, "RECV FETCH_CHUNK", "NO_META"); return HR_OK; @@ -786,9 +786,9 @@ found_size:; .size = 0, }; if (message.sender_idx >= 0 && message.sender_idx < state->num_nodes) - send_message(&state->msys,message.sender_idx, &response.base); + send_message(state->msys,message.sender_idx, &response.base); else - reply_to_message(&state->msys, raw_message, &response.base); + reply_to_message(state->msys, raw_message, &response.base); node_log(state, "RECV FETCH_CHUNK", "READ_ERR size=%u", chunk_size); return HR_OK; } @@ -802,7 +802,7 @@ found_size:; .hash = message.hash, .size = chunk_size, }; - reply_to_message_ex(&state->msys, raw_message, &response.base, chunk_data, chunk_size); + reply_to_message_ex(state->msys, raw_message, &response.base, chunk_data, chunk_size); node_log(state, "RECV FETCH_CHUNK", "size=%u", chunk_size); @@ -842,7 +842,7 @@ process_get_blob(Server *state, void *raw_message) response.found = false; } - reply_to_message(&state->msys, raw_message, &response.base); + reply_to_message(state->msys, raw_message, &response.base); return HR_OK; } @@ -968,7 +968,7 @@ process_single_future_list_entry(Server *state) .log_index = state->log.count-1, .view_number = state->view_number, }; - send_message(&state->msys,state->future[i].sender_idx, &message.base); + send_message(state->msys,state->future[i].sender_idx, &message.base); return 1; } @@ -1052,7 +1052,7 @@ process_prepare(Server *state, void *raw_message) .log_index = state->log.count-1, .view_number = state->view_number, }; - send_message(&state->msys,message.sender_idx, &ok_message.base); + send_message(state->msys,message.sender_idx, &ok_message.base); node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", message.sender_idx, state->log.count-1, message.oper.bucket, message.oper.key); @@ -1140,7 +1140,7 @@ process_new_state(Server *state, void *raw_message) .log_index = state->log.count - 1, .view_number = state->view_number, }; - send_message(&state->msys,primary_idx(state), &prepare_ok_message.base); + send_message(state->msys,primary_idx(state), &prepare_ok_message.base); node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->log.count - 1, state->log.entries[state->log.count - 1].oper.bucket, state->log.entries[state->log.count - 1].oper.key); } @@ -1164,7 +1164,7 @@ send_redirect(Server *state, void *raw_message) }, .view_number = state->view_number, }; - reply_to_message(&state->msys, raw_message, &redirect_message.base); + reply_to_message(state->msys, raw_message, &redirect_message.base); node_log(state, "SEND REDIRECT", "view=%llu leader=%d", (unsigned long long)state->view_number, primary_idx(state)); @@ -1401,15 +1401,16 @@ int server_init(void *state_, int argc, char **argv, return -1; } - if (message_system_init(&state->msys, state->node_addrs, state->num_nodes) < 0) { + state->msys = message_system_init(state->node_addrs, state->num_nodes); + if (state->msys == NULL) { fprintf(stderr, "Node :: Couldn't setup message system\n"); return -1; } - int ret = message_system_listen_tcp(&state->msys, state->self_addr); + int ret = message_system_listen_tcp(state->msys, state->self_addr); if (ret < 0) { fprintf(stderr, "Node :: Couldn't setup TCP listener\n"); - message_system_free(&state->msys); + message_system_free(state->msys); return -1; } @@ -1431,7 +1432,7 @@ int server_init(void *state_, int argc, char **argv, .sender_idx = self_idx(state), .nonce = state->recovery_nonce, }; - broadcast_message(&state->msys, self_idx(state),&recovery_message.base); + broadcast_message(state->msys, self_idx(state),&recovery_message.base); node_log(state, "SEND RECOVERY", "to=* nonce=%llu", state->recovery_nonce); nearest_deadline(&deadline, state->recovery_time + RECOVERY_TIMEOUT_SEC * 1000000000ULL); @@ -1442,7 +1443,7 @@ int server_init(void *state_, int argc, char **argv, fprintf(stderr, "Node :: Not enough poll() capacity (got %d, needed %d)\n", pcap, POLL_CAPACITY); return -1; } - *pnum = message_system_register_events(&state->msys, ctxs, pdata, pcap); + *pnum = message_system_register_events(state->msys, ctxs, pdata, pcap); return 0; } @@ -1459,16 +1460,16 @@ int server_tick(void *state_, void **ctxs, // NETWORK EVENTS ///////////////////////////////////////////////////////////////// - message_system_process_events(&state->msys, ctxs, pdata, *pnum); + message_system_process_events(state->msys, ctxs, pdata, *pnum); - for (void *ptr; (ptr = get_next_message(&state->msys)) != NULL; ) { + for (void *ptr; (ptr = get_next_message(state->msys)) != NULL; ) { HandlerResult ret = process_message(state, ptr); if (ret == HR_OUT_OF_MEMORY) return -1; if (ret != HR_DEFERRED) - consume_message(&state->msys, ptr); + consume_message(state->msys, ptr); } ///////////////////////////////////////////////////////////////// @@ -1491,7 +1492,7 @@ int server_tick(void *state_, void **ctxs, .sender_idx = self_idx(state), .nonce = state->recovery_nonce, }; - broadcast_message(&state->msys, self_idx(state),&recovery_message.base); + broadcast_message(state->msys, self_idx(state),&recovery_message.base); node_log(state, "SEND RECOVERY", "to=* nonce=%llu", state->recovery_nonce); state->recovery_time = state->now; @@ -1523,7 +1524,7 @@ int server_tick(void *state_, void **ctxs, .sender_idx = self_idx(state), }; node_log(state, "SEND BEGIN_VIEW_CHG", "to=* view=%llu", state->view_number); - broadcast_message(&state->msys, self_idx(state),&begin_view_change_message.base); + broadcast_message(state->msys, self_idx(state),&begin_view_change_message.base); } else { nearest_deadline(&deadline, view_change_deadline); @@ -1545,7 +1546,7 @@ int server_tick(void *state_, void **ctxs, .sender_idx = self_idx(state), .commit_index = state->commit_index, }; - broadcast_message(&state->msys, self_idx(state),&commit_message.base); + broadcast_message(state->msys, self_idx(state),&commit_message.base); node_log(state, "SEND COMMIT", "to=* commit=%d", state->commit_index); state->heartbeat = state->now; @@ -1576,7 +1577,7 @@ int server_tick(void *state_, void **ctxs, .view_number = state->view_number, .sender_idx = self_idx(state), }; - broadcast_message(&state->msys, self_idx(state),&begin_view_change_message.base); + broadcast_message(state->msys, self_idx(state),&begin_view_change_message.base); node_log(state, "SEND BEGIN_VIEW_CHG", "to=* view=%llu", state->view_number); node_log(state, "STATUS CHANGE_VIEW", "view=%llu", state->view_number); @@ -1602,7 +1603,7 @@ int server_tick(void *state_, void **ctxs, .op_number = state->log.count, .sender_idx = self_idx(state), }; - send_message(&state->msys,primary_idx(state), &get_state_message.base); + send_message(state->msys,primary_idx(state), &get_state_message.base); node_log(state, "SEND GET_STATE", "to=%d op=%d", primary_idx(state), state->log.count); state->state_transfer_time = state->now; @@ -1615,7 +1616,7 @@ int server_tick(void *state_, void **ctxs, *timeout = deadline_to_timeout(deadline, state->now); if (pcap < POLL_CAPACITY) return -1; - *pnum = message_system_register_events(&state->msys, ctxs, pdata, pcap); + *pnum = message_system_register_events(state->msys, ctxs, pdata, pcap); return 0; } @@ -1628,7 +1629,7 @@ int server_free(void *state_) log_free(&state->log); log_free(&state->recovery_log); log_free(&state->view_change_log); - message_system_free(&state->msys); + message_system_free(state->msys); client_table_free(&state->client_table); meta_store_free(&state->metastore); chunk_store_free(&state->chunk_store); diff --git a/src/server.h b/src/server.h index 8c715b0..5fc7a6e 100644 --- a/src/server.h +++ b/src/server.h @@ -211,7 +211,7 @@ typedef enum { typedef struct { - MessageSystem msys; + MessageSystem *msys; Address self_addr; Address node_addrs[NODE_LIMIT];