From 3bb30dcd09fcf79c07f099c78f5ba52583bbfa92 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Tue, 27 Jan 2026 22:49:12 +0100 Subject: [PATCH] Implement pipe() mock --- build.sh | 6 ++-- quakey/include/quakey.h | 2 ++ quakey/src/quakey.c | 73 ++++++++++++++++++++++++++++++++++++----- src/client.c | 6 ++-- src/metadata_server.c | 5 +-- src/test_client.c | 20 +++++++---- src/test_client.h | 1 + 7 files changed, 90 insertions(+), 23 deletions(-) diff --git a/build.sh b/build.sh index 872c0b9..c9abd40 100644 --- a/build.sh +++ b/build.sh @@ -1,4 +1,4 @@ -gcc src/sha256.c src/basic.c src/file_system.c src/byte_queue.c src/file_tree.c src/message.c src/tcp.c src/wal.c src/hash_set.c src/client.c src/metadata_server.c src/chunk_server.c src/random_client.c src/test_client.c src/main.c quakey/src/mockfs.c quakey/src/quakey.c -o toasty_simulation -Iquakey/include -Iinclude -Wall -Wextra -ggdb -O0 -DMAIN_SIMULATION -DFAULT_INJECTION +#gcc src/sha256.c src/basic.c src/file_system.c src/byte_queue.c src/file_tree.c src/message.c src/tcp.c src/wal.c src/hash_set.c src/client.c src/metadata_server.c src/chunk_server.c src/random_client.c src/test_client.c src/main.c quakey/src/mockfs.c quakey/src/quakey.c -o toasty_simulation -Iquakey/include -Iinclude -Wall -Wextra -ggdb -O0 -DMAIN_SIMULATION -DFAULT_INJECTION gcc src/sha256.c src/basic.c src/file_system.c src/byte_queue.c src/file_tree.c src/message.c src/tcp.c src/wal.c src/hash_set.c src/client.c src/metadata_server.c src/chunk_server.c src/random_client.c src/test_client.c src/main.c quakey/src/mockfs.c quakey/src/quakey.c -o toasty_test -Iquakey/include -Iinclude -Wall -Wextra -ggdb -O0 -DMAIN_TEST -gcc src/sha256.c src/basic.c src/file_system.c src/byte_queue.c src/file_tree.c src/message.c src/tcp.c src/wal.c src/hash_set.c src/client.c src/metadata_server.c src/chunk_server.c src/random_client.c src/test_client.c src/main.c -o toasty_metadata_server -DMAIN_METADATA_SERVER -Wall -Wextra -ggdb -O0 -Iinclude -Iquakey/include -gcc src/sha256.c src/basic.c src/file_system.c src/byte_queue.c src/file_tree.c src/message.c src/tcp.c src/wal.c src/hash_set.c src/client.c src/metadata_server.c src/chunk_server.c src/random_client.c src/test_client.c src/main.c -o toasty_chunk_server -DMAIN_CHUNK_SERVER -Wall -Wextra -ggdb -O0 -Iinclude -Iquakey/include +#gcc src/sha256.c src/basic.c src/file_system.c src/byte_queue.c src/file_tree.c src/message.c src/tcp.c src/wal.c src/hash_set.c src/client.c src/metadata_server.c src/chunk_server.c src/random_client.c src/test_client.c src/main.c -o toasty_metadata_server -DMAIN_METADATA_SERVER -Wall -Wextra -ggdb -O0 -Iinclude -Iquakey/include +#gcc src/sha256.c src/basic.c src/file_system.c src/byte_queue.c src/file_tree.c src/message.c src/tcp.c src/wal.c src/hash_set.c src/client.c src/metadata_server.c src/chunk_server.c src/random_client.c src/test_client.c src/main.c -o toasty_chunk_server -DMAIN_CHUNK_SERVER -Wall -Wextra -ggdb -O0 -Iinclude -Iquakey/include diff --git a/quakey/include/quakey.h b/quakey/include/quakey.h index 3bdfa70..ad070cd 100644 --- a/quakey/include/quakey.h +++ b/quakey/include/quakey.h @@ -110,6 +110,7 @@ int mock_connect(int fd, void *addr, unsigned long addr_len); int mock_getsockopt(int fd, int level, int optname, void *optval, socklen_t *optlen); int mock_listen(int fd, int backlog); int mock_accept(int fd, void *addr, socklen_t *addr_len); +int mock_pipe(int *fds); int mock_recv(int fd, char *dst, int len, int flags); int mock_send(int fd, char *src, int len, int flags); int mock_clock_gettime(clockid_t clockid, struct timespec *tp); @@ -152,6 +153,7 @@ void mock_free(void *ptr); #define getsockopt mock_getsockopt #define listen mock_listen #define accept mock_accept +#define pipe mock_pipe #define recv mock_recv #define send mock_send #define clock_gettime mock_clock_gettime diff --git a/quakey/src/quakey.c b/quakey/src/quakey.c index c57682e..44305ea 100644 --- a/quakey/src/quakey.c +++ b/quakey/src/quakey.c @@ -101,6 +101,8 @@ typedef enum { // The Desc represent an open directory DESC_DIRECTORY, + + DESC_PIPE, } DescType; typedef enum { @@ -148,12 +150,13 @@ struct Desc { AcceptQueue accept_queue; ///////////////////////////////////////// - // Connection socket fields + // Pipe or Connection socket fields // Status code of the connecting process ConnectStatus connect_status; // These are used when the socket is still connecting + // (only used by sockets, not pipes) Addr connect_addr; u16 connect_port; @@ -793,6 +796,7 @@ static void desc_free(Sim *sim, Desc *desc, bool rst) } accept_queue_free(&desc->accept_queue); break; + case DESC_PIPE: case DESC_SOCKET_C: // Remove any pending events targeting this descriptor before freeing it if (sim) @@ -805,7 +809,7 @@ static void desc_free(Sim *sim, Desc *desc, bool rst) // Connection was waiting to be accepted accept_queue_remove(&peer->accept_queue, desc); } else { - ASSERT(peer->type == DESC_SOCKET_C); + ASSERT(peer->type == DESC_SOCKET_C || peer->type == DESC_PIPE); peer->peer = NULL; peer->connect_status = rst ? CONNECT_STATUS_RESET : CONNECT_STATUS_CLOSE; } @@ -1056,6 +1060,7 @@ static bool host_ready(Host *host) return true; } break; + case DESC_PIPE: case DESC_SOCKET_C: if (desc->connect_status == CONNECT_STATUS_NOHOST || desc->connect_status == CONNECT_STATUS_RESET) @@ -1108,20 +1113,22 @@ static void set_revents_in_poll_array(Host *host) revents = POLLIN; } break; + case DESC_PIPE: case DESC_SOCKET_C: if (desc->connect_status == CONNECT_STATUS_NOHOST || desc->connect_status == CONNECT_STATUS_RESET) revents |= POLLERR; if (events & POLLIN) { // TODO: should report prover events when hup and rst are set - if (!socket_queue_empty(desc->input) || desc->connect_status == CONNECT_STATUS_CLOSE) + if (!socket_queue_empty(desc->input)) + revents |= POLLIN; + if (desc->connect_status == CONNECT_STATUS_RESET || + desc->connect_status == CONNECT_STATUS_CLOSE) revents |= POLLIN; } if (events & POLLOUT) { - if (!socket_queue_full(desc->output)) { - if (is_connected_and_accepted(desc)) - revents |= POLLOUT; - } + if (!socket_queue_full(desc->output) && is_connected_and_accepted(desc)) + revents |= POLLOUT; } break; case DESC_FILE: @@ -1258,6 +1265,39 @@ static int host_create_socket(Host *host, AddrFamily family) return desc_idx; } +static int host_create_pipe(Host *host, int *desc_idxs) +{ + int desc_idx_1 = find_empty_desc_struct(host); + if (desc_idx_1 < 0) + return HOST_ERROR_FULL; + Desc *desc_1 = &host->desc[desc_idx_1]; + + int desc_idx_2 = find_empty_desc_struct(host); + if (desc_idx_2 < 0) + return HOST_ERROR_FULL; + Desc *desc_2 = &host->desc[desc_idx_2]; + + desc_1->type = DESC_PIPE; + desc_1->non_blocking = false; + desc_1->connect_status = CONNECT_STATUS_DONE; + desc_1->peer = desc_2; + desc_1->input = socket_queue_init(1<<12); + desc_1->output = socket_queue_init(1<<12); + + desc_2->type = DESC_PIPE; + desc_2->non_blocking = false; + desc_2->connect_status = CONNECT_STATUS_DONE; + desc_2->peer = desc_1; + desc_2->input = socket_queue_init(1<<12); + desc_2->output = socket_queue_init(1<<12); + + desc_idxs[0] = desc_idx_1; + desc_idxs[1] = desc_idx_2; + + host->num_desc += 2; + return 0; +} + static bool is_desc_idx_valid(Host *host, int desc_idx) { // Out of bounds @@ -1699,7 +1739,8 @@ static int host_read(Host *host, int desc_idx, char *dst, int len) #endif int num = 0; - if (desc->type == DESC_SOCKET_C) { + if (desc->type == DESC_SOCKET_C || + desc->type == DESC_PIPE) { num = recv_inner(desc, dst, len); } else if (desc->type == DESC_FILE) { #ifdef FAULT_INJECTION @@ -1742,7 +1783,8 @@ static int host_write(Host *host, int desc_idx, char *src, int len) #endif int num = 0; - if (desc->type == DESC_SOCKET_C) { + if (desc->type == DESC_SOCKET_C || + desc->type == DESC_PIPE) { num = send_inner(desc, src, len); } else if (desc->type == DESC_FILE) { #ifdef FAULT_INJECTION @@ -2602,6 +2644,19 @@ int mock_connect(int fd, void *addr, unsigned long addr_len) return -1; } +int mock_pipe(int *fds) +{ + Host *host = host___; + if (host == NULL) + abort_("Call to mock_pipe() with no node scheduled\n"); + + int ret = host_create_pipe(host, fds); + if (ret < 0) + return EIO; + + return 0; +} + static int convert_linux_open_flags_to_mockfs(int flags) { int lfs_flags = 0; diff --git a/src/client.c b/src/client.c index 0f84db0..c6ad55e 100644 --- a/src/client.c +++ b/src/client.c @@ -36,8 +36,8 @@ typedef pthread_mutex_t Mutex; #define PARALLEL_LIMIT 5 -//#define CLIENT_TRACE(fmt, ...) {} -#define CLIENT_TRACE(fmt, ...) fprintf(stderr, "CLIENT: " fmt "\n", ##__VA_ARGS__); +#define CLIENT_TRACE(fmt, ...) {} +//#define CLIENT_TRACE(fmt, ...) fprintf(stderr, "CLIENT: " fmt "\n", ##__VA_ARGS__); typedef struct { SHA256 hash; @@ -1555,6 +1555,8 @@ static void process_event_for_write(ToastyFS *toasty, int next_server_lid = 0; toasty->operations[opidx].num_uploads = 0; + + // This loop is for chunks that already exist in the file for (uint32_t i = 0; i < num_hashes; i++) { char *src = full_ptr + relative_off; diff --git a/src/metadata_server.c b/src/metadata_server.c index dbe3242..156a112 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -9,8 +9,8 @@ #include "message.h" #include "metadata_server.h" -//#define MS_TRACE(fmt, ...) {} -#define MS_TRACE(fmt, ...) fprintf(stderr, "MS: " fmt "\n", ##__VA_ARGS__); +#define MS_TRACE(fmt, ...) {} +//#define MS_TRACE(fmt, ...) fprintf(stderr, "MS: " fmt "\n", ##__VA_ARGS__); static void chunk_server_peer_init(ChunkServerPeer *chunk_server, Time current_time) { @@ -1226,6 +1226,7 @@ int metadata_server_tick(void *state_, void **ctxs, MS_TRACE("Message from client"); ret = process_client_message(state, conn_idx, msg_type, msg); } else { + MS_TRACE("Message from CS (conn_idx=%d)", conn_idx); state->chunk_servers[tag].last_response_time = current_time; ret = process_chunk_server_message(state, conn_idx, msg_type, msg); } diff --git a/src/test_client.c b/src/test_client.c index 71629dc..6bfabcb 100644 --- a/src/test_client.c +++ b/src/test_client.c @@ -52,10 +52,11 @@ int test_client_init(void *state_, int argc, char **argv, return -1; client->state = TEST_CLIENT_STATE_0; + client->tick = 0; printf("Client set up (remote=%s:%d)\n", addr, port); - *timeout = 0; + *timeout = -1; if (pcap < TCP_POLL_CAPACITY) return -1; *pnum = toasty_process_events(client->toasty, ctxs, pdata, *pnum); @@ -94,11 +95,16 @@ int test_client_tick(void *state_, void **ctxs, int ret; ToastyResult result; case TEST_CLIENT_STATE_0: + if (client->tick < 10) { + *timeout = 0; + client->tick++; + return 0; + } client->handle = toasty_begin_create_file(client->toasty, file_path, 128); if (client->handle == TOASTY_INVALID) { assert(0); // TODO } - printf(">>> Create started <<<<<<<<<<<<\n"); + printf("Create started\n"); client->state = TEST_CLIENT_STATE_1; break; case TEST_CLIENT_STATE_1: @@ -109,7 +115,7 @@ int test_client_tick(void *state_, void **ctxs, if (ret == 1) { break; } - printf(">>> Create completed <<<<<<<<<<<<\n"); + printf("Create completed\n"); assert(ret == 0); if (result.type != TOASTY_RESULT_CREATE_SUCCESS) { assert(0); // TODO @@ -118,7 +124,7 @@ int test_client_tick(void *state_, void **ctxs, if (client->handle == TOASTY_INVALID) { assert(0); // TODO } - printf(">>> Write started <<<<<<<<<<<<\n"); + printf("Write started\n"); client->state = TEST_CLIENT_STATE_2; break; case TEST_CLIENT_STATE_2: @@ -129,7 +135,7 @@ int test_client_tick(void *state_, void **ctxs, if (ret == 1) { break; } - printf(">>> Write completed <<<<<<<<<<<<\n"); + printf("Write completed\n"); assert(ret == 0); if (result.type != TOASTY_RESULT_WRITE_SUCCESS) { assert(0); // TODO @@ -138,7 +144,7 @@ int test_client_tick(void *state_, void **ctxs, if (client->handle == TOASTY_INVALID) { assert(0); // TODO } - printf(">>> Read started <<<<<<<<<<<<\n"); + printf("Read started\n"); client->state = TEST_CLIENT_STATE_3; break; case TEST_CLIENT_STATE_3: @@ -149,7 +155,7 @@ int test_client_tick(void *state_, void **ctxs, if (ret == 1) { break; } - printf(">>> Read completed <<<<<<<<<<<<\n"); + printf("Read completed\n"); assert(ret == 0); if (result.type != TOASTY_RESULT_READ_SUCCESS) { assert(0); // TODO diff --git a/src/test_client.h b/src/test_client.h index 8779878..af5e210 100644 --- a/src/test_client.h +++ b/src/test_client.h @@ -14,6 +14,7 @@ typedef struct { ToastyFS* toasty; ToastyHandle handle; TestClientState state; + int tick; char buf[1<<10]; } TestClient;