From e124952fe2c7122caf1c48a4010c0c2e52d5355f Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Wed, 5 Nov 2025 14:26:57 +0100 Subject: [PATCH] Progress --- TODO.txt | 2 + src/basic.c | 40 ++++ src/basic.h | 4 + src/chunk_server.c | 40 +++- src/client.c | 403 +++++++++++++++++++++++++++++++++++++++- src/main_server.c | 10 +- src/main_test.c | 43 ++--- src/metadata_server.c | 17 +- src/simulation_client.c | 5 +- src/system.c | 20 +- src/tcp.c | 12 +- src/tcp.h | 2 +- 12 files changed, 511 insertions(+), 87 deletions(-) diff --git a/TODO.txt b/TODO.txt index 41395cc..a06dc61 100644 --- a/TODO.txt +++ b/TODO.txt @@ -6,3 +6,5 @@ - When a client received metadata and starts reading or writing to a chunk server, it should try connecting to all addresses of a chunk servers, not just the first one. - Return the number of bytes read or written in the TinyDFS_Result struct - Make parallel uploads/downloads configurable +- Recalculate next write locations whenever a write occurs, not at each read +- Make sure there are no mixups with the default tag value for connections, which I think is -1? diff --git a/src/basic.c b/src/basic.c index 1d4f014..e284536 100644 --- a/src/basic.c +++ b/src/basic.c @@ -1,3 +1,5 @@ +#include + #include "basic.h" #include "system.h" @@ -54,3 +56,41 @@ Time get_current_time(void) } #endif } + +bool getargb(int argc, char **argv, char *name) +{ + for (int i = 0; i < argc; i++) + if (!strcmp(argv[i], name)) + return true; + return false; +} + +string getargs(int argc, char **argv, char *name, char *fallback) +{ + for (int i = 0; i < argc; i++) + if (!strcmp(argv[i], name)) { + i++; + if (i == argc) + break; + return (string) { argv[i], strlen(argv[i]) }; + } + return (string) { fallback, strlen(fallback) }; +} + +int getargi(int argc, char **argv, char *name, int fallback) +{ + for (int i = 0; i < argc; i++) + if (!strcmp(argv[i], name)) { + + i++; + if (i == argc) + break; + + int tmp = atoi(argv[i]); + if (tmp == 0 && argv[i][0] != '0') // best effort + break; + + return tmp; + } + return fallback; +} diff --git a/src/basic.h b/src/basic.h index 9afc0c0..232df36 100644 --- a/src/basic.h +++ b/src/basic.h @@ -25,4 +25,8 @@ typedef uint64_t Time; bool streq(string s1, string s2); Time get_current_time(void); +bool getargb(int argc, char **argv, char *name); +string getargs(int argc, char **argv, char *name, char *fallback); +int getargi(int argc, char **argv, char *name, int fallback); + #endif // BASIC_INCLUDED diff --git a/src/chunk_server.c b/src/chunk_server.c index 920b587..0e767d9 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -766,15 +766,18 @@ process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout) { - (void) argc; - (void) argv; + string addr = getargs(argc, argv, "--addr", "127.0.0.1"); + int port = getargi(argc, argv, "--port", 8081); + string path = getargs(argc, argv, "--path", "chunk_server_data/"); - char addr[] = "127.0.0.1"; - uint16_t port = 8080; - string path = S("chunk_server_data_0/"); + string remote_addr = getargs(argc, argv, "--remote-addr", "127.0.0.1"); + int remote_port = getargi(argc, argv, "--remote-port", 8080); - char metadata_server_addr[] = "127.0.0.1"; - uint16_t metadata_server_port = 8081; + if (port <= 0 || port >= 1<<16) + return -1; + + if (remote_port <= 0 || remote_port >= 1<<16) + return -1; tcp_context_init(&state->tcp); @@ -793,18 +796,37 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts state->downloading = false; pending_download_list_init(&state->pending_download_list); + char tmp[1<<10]; + if (remote_addr.len >= (int) sizeof(tmp)) { + tcp_context_free(&state->tcp); + return -1; + } + memcpy(tmp, remote_addr.ptr, remote_addr.len); + tmp[remote_addr.len] = '\0'; + // Initialize metadata server address // // TODO: This should also support IPv6 state->metadata_server_addr.is_ipv4 = true; - if (inet_pton(AF_INET, metadata_server_addr, &state->metadata_server_addr.ipv4) != 1) { + if (inet_pton(AF_INET, tmp, &state->metadata_server_addr.ipv4) != 1) { tcp_context_free(&state->tcp); chunk_store_free(&state->store); return -1; } - state->metadata_server_addr.port = metadata_server_port; + state->metadata_server_addr.port = remote_port; state->metadata_server_disconnect_time = 0; + printf("Chunk server set up (local=%.*s:%d, remote=%.*s:%d, path=%.*s)\n", + addr.len, + addr.ptr, + port, + remote_addr.len, + remote_addr.ptr, + remote_port, + path.len, + path.ptr + ); + *timeout = -1; // No timeout needed for chunk server initially return tcp_register_events(&state->tcp, contexts, polled); } diff --git a/src/client.c b/src/client.c index 98c5ef8..98601f2 100644 --- a/src/client.c +++ b/src/client.c @@ -1,3 +1,4 @@ +#include "basic.h" #include #include #include @@ -38,6 +39,57 @@ typedef enum { OPERATION_TYPE_WRITE, } OperationType; +typedef struct { + SHA256 hash; + char * src; + int len; + int num_holders; + Address holders[MAX_CHUNK_HOLDERS][MAX_CHUNK_SERVER_ADDR]; +} WriteChunk; + +typedef enum { + + // This upload wasn't started yet + UPLOAD_WAITING, + + // This upload started + UPLOAD_PENDING, + + // This upload was WAITING but then + // was marked as IGNORED + UPLOAD_IGNORED, + + // Upload was PENDING and FAILED + UPLOAD_FAILED, + + // Upload was PENDING, then COMPLETED + // successfully + UPLOAD_COMPLETED, + +} UploadScheduleStatus; + +typedef struct { + UploadScheduleStatus status; + + // Location of the chunk to be patched. + // The server local ID is used to indicate + // that different addresses refer to the + // same server. + // The no_hash flag indicates that this is + // a new chunk and doesn't need to patch + // an old one. + int server_lid; + bool no_hash; + Address address; + SHA256 hash; + + // The patch offset and data + char *src; + int off; + int len; + +} UploadSchedule; + typedef struct { OperationType type; @@ -51,6 +103,9 @@ typedef struct { int ranges_count; int num_pending; + UploadSchedule *uploads; + int num_uploads; + TinyDFS_Result result; } Operation; @@ -129,7 +184,6 @@ TinyDFS *tinydfs_init(char *addr, uint16_t port) tdfs->num_chunk_servers = 0; for (int i = 0; i < MAX_CHUNK_SERVERS; i++) { tdfs->chunk_servers[i].used = false; - // Note: RequestQueue initialized in get_chunk_server_connection() } return tdfs; @@ -885,6 +939,90 @@ static void process_event_for_read(TinyDFS *tdfs, } } +static int start_upload(Operation *o) +{ + int found = -1; + + // Find a PENDING operation that can be started + for (int i = 0; i < o->num_uploads; i++) { + + if (o->status != UPLOAD_PENDING) + continue; + + // Can't start uploads of a chunk to the + // same server twice. + bool invalid = false; + for (int j = 0; j < o->num_uploads; j++) { + + if (j == i) + continue; + + if (o->uploads[i].server_lid == o->uploads[j].server_lid || + addr_eql(o->uploads[i].address, o->uploads[j].address)) { + invalid = true; + break; + } + } + + if (invalid) + continue; + + found = i; + } + + if (found < 0) + return -1; // No upload can be started at this time + + int ret = get_chunk_server_connection(tdfs, o->uploads[found].address); + + int tag = xxx; + + ByteQueue *output; + int ret = tcp_connect(tcp, o->uploads[found].address, tag, &output); + if (ret < 0) + return -1; + + if (o->uploads[found].no_hash) { + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_CREATE_CHUNK); + + uint32_t chunk_size = xxx; + uint32_t target_off = 0; + uint32_t target_len = 0; + + message_write(&writer, &chunk_size, sizeof(chunk_size)); + message_write(&writer, &target_off, sizeof(target_off)); + message_write(&writer, &target_len, sizeof(target_len)); + message_write(&writer, data.ptr, data.len); + + if (message_writer_free(&writer) < 0) { + // TODO + } + + } else { + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_UPLOAD_CHUNK); + + SHA256 target_hash = xxx; + uint32_t target_off = 0; + uint32_t target_len = 0; + + message_write(&writer, &target_hash, sizeof(target_hash)); + message_write(&writer, &target_off, sizeof(target_off)); + message_write(&writer, &target_len, sizeof(target_len)); + message_write(&writer, data.ptr, data.len); + + if (message_writer_free(&writer) < 0) { + // TODO + } + } + + o->uploads[found].status = UPLOAD_PENDING; + return 0; +} + static void process_event_for_write(TinyDFS *tdfs, int opidx, int request_tag, ByteView msg) { @@ -896,14 +1034,261 @@ static void process_event_for_write(TinyDFS *tdfs, switch (request_tag) { case TAG_RETRIEVE_METADATA_FOR_WRITE: - // Process metadata response and initiate chunk uploads - // This would involve: - // 1. Parsing the metadata response (chunk locations, hashes) - // 2. Computing new chunk data by patching existing chunks - // 3. Uploading new chunks to chunk servers - // 4. Committing the write to the metadata server with new hashes - // For now, this operation is not fully implemented - tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + { + // We are expecting one of: + // MESSAGE_TYPE_READ_ERROR + // MESSAGE_TYPE_READ_SUCCESS + + BinaryReader reader = { msg.ptr, msg.len, 0 }; + + if (!binary_read(&reader, NULL, sizeof(uint16_t))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + uint16_t type; + if (!binary_read(&reader, &type, sizeof(type))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + if (type != MESSAGE_TYPE_READ_SUCCESS) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + if (!binary_read(&reader, NULL, sizeof(uint32_t))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + uint32_t chunk_size; + if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + uint32_t num_hashes; + if (!binary_read(&reader, &num_hashes, sizeof(num_hashes))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + // TODO: !!! IMPORTANT !!! This should also account for new chunks, not patched. It does not do so at the moment + // TODO: This may overestimate by a lot the actual memory required by the array + client->operations[opidx].uploads = sys_malloc(num_hashes * MAX_CHUNK_HOLDERS * MAX_CHUNK_SERVER_ADDR * sizeof(UploadSchedule)); + if (client->operations[opidx].uploads == NULL) { + // TODO + } + + int next_server_lid = 0; + client->operations[opidx].num_uploads = 0; + for (uint32_t i = 0; i < num_hashes; i++) { + + void *src = xxx; + int off = xxx; + int len = xxx; + + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + uint32_t num_holders; + if (!binary_read(&reader, &num_holders, sizeof(num_holders))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + for (uint32_t j = 0; j < num_holders; j++) { + + int server_lid = next_server_lid; + next_server_lid++; + + uint32_t num_ipv4; + if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + for (uint32_t k = 0; k < num_ipv4; k++) { + + IPv4 ipv4; + if (!binary_read(&reader, &ipv4, sizeof(ipv4))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + uint16_t port; + if (!binary_read(&reader, &port, sizeof(port))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + UploadSchedule upload; + upload.status = UPLOAD_WAITING; + upload.server_lid = server_lid; + upload.no_hash = false; + upload.address.is_ipv4 = true; + upload.address.ipv4 = ipv4; + upload.address.port = port; + upload.hash = hash; + upload.src = src; + upload.off = off; + upload.len = len; + int n = client->operations[opidx].num_uploads++; + client->operations[opidx].uploads[n] = upload; + } + + uint32_t num_ipv6; + if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + for (uint32_t k = 0; k < num_ipv6; k++) { + + IPv6 ipv6; + if (!binary_read(&reader, &ipv6, sizeof(ipv6))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + uint16_t port; + if (!binary_read(&reader, &port, sizeof(port))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + UploadSchedule upload; + upload.status = UPLOAD_WAITING; + upload.server_lid = server_lid; + upload.no_hash = false; + upload.address.is_ipv4 = false; + upload.address.ipv6 = ipv6; + upload.address.port = port; + upload.hash = hash; + upload.src = src; + upload.off = off; + upload.len = len; + int n = client->operations[opidx].num_uploads++; + client->operations[opidx].uploads[n] = upload; + } + } + } + + uint32_t num_locations; + if (!binary_read(&reader, &num_locations, sizeof(num_locations))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + for (uint32_t i = 0; i < num_locations; i++) { + + uint32_t num_ipv4; + if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + for (uint32_t k = 0; k < num_ipv4; k++) { + + IPv4 ipv4; + if (!binary_read(&reader, &ipv4, sizeof(ipv4))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + uint16_t port; + if (!binary_read(&reader, &port, sizeof(port))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + // TODO + } + + uint32_t num_ipv6; + if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + for (uint32_t k = 0; k < num_ipv6; k++) { + + IPv6 ipv6; + if (!binary_read(&reader, &ipv6, sizeof(ipv6))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + uint16_t port; + if (!binary_read(&reader, &port, sizeof(port))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + // TODO + } + } + + // Now start the first batch of uploads + int started = 0; + for (int i = 0; i < xxx; i++) { + if (start_upload(&tdfs->operations[opidx]) == 0) + started++; + } + + if (started == 0) { + // We already failed + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + return; + } + + // TODO: Now we need to upload the patches to N of the + // chunk servers that are holding each old chunk + // All new chunks need to be written to the specified + // locations at least N times. If any upload fails, + // the write fails. If all writes succede, the client + // sends the metadata server a WRITE operation + // swapping the old hashes with the new ones. + // + // The algorithm should go like this: + // - Iterate over each chunk + // - Pick the first N holders of the chunk. If less than N + // are available, pick M. + // - For each pick, take the first address and start the + // chunk upload + // + // If an upload fails, + // + // + // + // example upload schedule: + // chunk_A server_A addr_0 + // chunk_A server_A addr_1 + // chunk_A server_B addr_0 + // chunk_A server_B addr_1 + // chunk_A server_B addr_2 + // chunk_A server_C addr_0 + // chunk_B server_D addr_0 + // chunk_B server_E addr_0 + // chunk_B server_E addr_1 + // chunk_B server_F addr_0 + // + // If an upload succedes, all uploads of the chunk to the same server + // are removed and if this was the N-th successful upload of a chunk, + // all uploads of the same chunk are removed. + // + // Uploads to the same chunk server with different addresses can't + // be parallelized, so + + // The client should not try any random N chunk servers + // for upload. It must try all chunk servers until N respond + + tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; + } return; default: diff --git a/src/main_server.c b/src/main_server.c index 506a6d2..f2b4792 100644 --- a/src/main_server.c +++ b/src/main_server.c @@ -53,17 +53,9 @@ int chunk_server_main(int argc, char **argv) return 0; } -bool is_leader(int argc, char **argv) -{ - for (int i = 1; i < argc; i++) - if (!strcmp(argv[i], "--leader") || !strcmp(argv[i], "-l")) - return true; - return false; -} - int main(int argc, char **argv) { - if (is_leader(argc, argv)) + if (getargb(argc, argv, "--leader")) return metadata_server_main(argc, argv); else return chunk_server_main(argc, argv); diff --git a/src/main_test.c b/src/main_test.c index a42287b..c0a8bda 100644 --- a/src/main_test.c +++ b/src/main_test.c @@ -18,45 +18,28 @@ int main(int argc, char **argv) startup_simulation(); // Spawn metadata server (leader) - spawn_simulated_process("--addr 127.0.0.1 8080 --leader"); + spawn_simulated_process("--addr 127.0.0.1 --port 8080 --leader"); // Spawn chunk servers - spawn_simulated_process("--addr 127.0.0.1 8081"); - spawn_simulated_process("--addr 127.0.0.1 8082"); - spawn_simulated_process("--addr 127.0.0.1 8083"); - spawn_simulated_process("--addr 127.0.0.1 8084"); - spawn_simulated_process("--addr 127.0.0.1 8085"); - spawn_simulated_process("--addr 127.0.0.1 8086"); - spawn_simulated_process("--addr 127.0.0.1 8087"); - spawn_simulated_process("--addr 127.0.0.1 8088"); - spawn_simulated_process("--addr 127.0.0.1 8089"); - spawn_simulated_process("--addr 127.0.0.1 8090"); + spawn_simulated_process("--addr 127.0.0.1 --port 8081 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_0/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8082 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_1/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8083 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_2/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8084 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_3/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8085 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_4/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8086 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_5/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8087 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_6/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8088 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_7/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8089 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_8/"); + spawn_simulated_process("--addr 127.0.0.1 --port 8090 --remote-addr 127.0.0.1 --remote-port 8080 --path chunk_server_data_9/"); // Spawn simulation client - spawn_simulated_process("--client --server 127.0.0.1 8080"); + spawn_simulated_process("--client --remote-addr 127.0.0.1 --remote-port 8080"); printf("Running simulation (press Ctrl+C to stop)...\n"); - - // Run for a limited number of iterations for testing - int iteration = 0; - int max_iterations = 100000; // Increased to allow client operations to complete - while (!simulation_should_stop && iteration < max_iterations) { + while (!simulation_should_stop) update_simulation(); - iteration++; - - // Print progress every 10000 iterations - if (iteration % 10000 == 0) { - fprintf(stderr, "Iteration %d...\n", iteration); - fflush(stderr); - } - } - - if (iteration >= max_iterations) { - printf("\nSimulation stopped after %d iterations\n", max_iterations); - } cleanup_simulation(); - printf("Simulation complete!\n"); return 0; } diff --git a/src/metadata_server.c b/src/metadata_server.c index 7eadbf8..e54c5b3 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -480,6 +480,9 @@ process_client_read(MetadataServer *state, int conn_idx, ByteView msg) int locations[MAX_CHUNK_SERVERS]; int num_locations = choose_servers_for_write(state, locations, state->replication_factor); + uint32_t tmp = num_locations; + message_write(&writer, &tmp, sizeof(tmp)); + for (int j = 0; j < num_locations; j++) message_write_server_addr(&writer, &state->chunk_servers[locations[j]]); @@ -752,11 +755,11 @@ static bool is_chunk_server_message_type(uint16_t type) int metadata_server_init(MetadataServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout) { - (void) argc; - (void) argv; + string addr = getargs(argc, argv, "--addr", "127.0.0.1"); + int port = getargi(argc, argv, "--port", 8080); - char addr[] = "127.0.0.1"; - uint16_t port = 8080; + if (port <= 0 || port >= 1<<16) + return -1; state->replication_factor = 3; if (state->replication_factor > MAX_CHUNK_SERVERS) @@ -778,6 +781,12 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co return -1; } + printf("Metadata server set up (local=%.*s:%d)\n", + addr.len, + addr.ptr, + port + ); + *timeout = -1; // No timeout needed for metadata server return tcp_register_events(&state->tcp, contexts, polled); } diff --git a/src/simulation_client.c b/src/simulation_client.c index 0b1e2d2..dfa4a21 100644 --- a/src/simulation_client.c +++ b/src/simulation_client.c @@ -34,11 +34,8 @@ int simulation_client_init(SimulationClient *client, int argc, char **argv, uint16_t port; parse_server_addr(argc, argv, &addr, &port); - printf("[Client] Initializing TinyDFS client, connecting to %s:%u\n", addr, port); - client->tdfs = tinydfs_init(addr, port); if (client->tdfs == NULL) { - fprintf(stderr, "[Client] Failed to initialize TinyDFS client\n"); return -1; } @@ -51,7 +48,7 @@ int simulation_client_init(SimulationClient *client, int argc, char **argv, client->list_op = -1; client->delete_op = -1; - printf("[Client] Initialized successfully\n"); + printf("Client set up (remote=%s:%d)\n", addr, port); *timeout = 0; // Wake up immediately to start processing return tinydfs_process_events(client->tdfs, contexts, polled, 0); diff --git a/src/system.c b/src/system.c index 3664a9c..8556115 100644 --- a/src/system.c +++ b/src/system.c @@ -241,22 +241,6 @@ static void process_poll_array(Process *process, } } -static bool is_leader(int argc, char **argv) -{ - for (int i = 0; i < argc; i++) - if (!strcmp("--leader", argv[i]) || !strcmp("-l", argv[i])) - return true; - return false; -} - -static bool is_client(int argc, char **argv) -{ - for (int i = 0; i < argc; i++) - if (!strcmp("--client", argv[i]) || !strcmp("-c", argv[i])) - return true; - return false; -} - #define MAX_ARGS 128 static bool is_space(char c) @@ -303,8 +287,8 @@ int spawn_simulated_process(char *args) cur++; } - bool leader = is_leader(argc, argv); - bool client = is_client(argc, argv); + bool leader = getargb(argc, argv, "--leader"); + bool client = getargb(argc, argv, "--client"); Process *process = malloc(sizeof(Process)); if (process == NULL) diff --git a/src/tcp.c b/src/tcp.c index 8a80577..a1ae01a 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -24,16 +24,22 @@ bool addr_eql(Address a, Address b) return true; } -static SOCKET create_listen_socket(char *addr, uint16_t port) +static SOCKET create_listen_socket(string addr, uint16_t port) { SOCKET fd = sys_socket(AF_INET, SOCK_STREAM, 0); if (fd == INVALID_SOCKET) return INVALID_SOCKET; + char tmp[1<<10]; + if (addr.len >= (int) sizeof(tmp)) + return INVALID_SOCKET; + memcpy(tmp, addr.ptr, addr.len); + tmp[addr.len] = '\0'; + struct sockaddr_in bind_buf; bind_buf.sin_family = AF_INET; bind_buf.sin_port = htons(port); - if (inet_pton(AF_INET, addr, &bind_buf.sin_addr) != 1) + if (inet_pton(AF_INET, tmp, &bind_buf.sin_addr) != 1) return INVALID_SOCKET; if (sys_bind(fd, (struct sockaddr*) &bind_buf, sizeof(bind_buf))) @@ -110,7 +116,7 @@ int tcp_index_from_tag(TCP *tcp, int tag) return -1; } -int tcp_listen(TCP *tcp, char *addr, uint16_t port) +int tcp_listen(TCP *tcp, string addr, uint16_t port) { SOCKET listen_fd = create_listen_socket(addr, port); if (listen_fd == INVALID_SOCKET) diff --git a/src/tcp.h b/src/tcp.h index 3c3398d..de4b7e0 100644 --- a/src/tcp.h +++ b/src/tcp.h @@ -62,7 +62,7 @@ bool addr_eql(Address a, Address b); void tcp_context_init(TCP *tcp); void tcp_context_free(TCP *tcp); int tcp_index_from_tag(TCP *tcp, int tag); -int tcp_listen(TCP *tcp, char *addr, uint16_t port); +int tcp_listen(TCP *tcp, string addr, uint16_t port); int tcp_next_message(TCP *tcp, int conn_idx, ByteView *msg, uint16_t *type); void tcp_consume_message(TCP *tcp, int conn_idx); int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd *polled, int num_polled);