Implement pipe() mock

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_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_chunk_server    -DMAIN_CHUNK_SERVER    -Wall -Wextra -ggdb -O0 -Iinclude -Iquakey/include

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

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 (!socket_queue_full(desc->output) && is_connected_and_accepted(desc))
-                    if (is_connected_and_accepted(desc))
+                    revents |= POLLOUT;
-                        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;

src/client.c

@@ -36,8 +36,8 @@ typedef pthread_mutex_t Mutex;
 
 #define PARALLEL_LIMIT 5
 
-//#define CLIENT_TRACE(fmt, ...) {}
+#define CLIENT_TRACE(fmt, ...) {}
-#define CLIENT_TRACE(fmt, ...) fprintf(stderr, "CLIENT: " fmt "\n", ##__VA_ARGS__);
+//#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;

src/metadata_server.c

@@ -9,8 +9,8 @@
 #include "message.h"
 #include "metadata_server.h"
 
-//#define MS_TRACE(fmt, ...) {}
+#define MS_TRACE(fmt, ...) {}
-#define MS_TRACE(fmt, ...) fprintf(stderr, "MS: " fmt "\n", ##__VA_ARGS__);
+//#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);
                     }

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

src/test_client.h

@@ -14,6 +14,7 @@ typedef struct {
     ToastyFS*       toasty;
     ToastyHandle    handle;
     TestClientState state;
+    int  tick;
     char buf[1<<10];
 } TestClient;