Progress

src/chunk_server.c

@@ -34,8 +34,10 @@ pending_download_list_add(PendingDownloadList *list, Address addr, SHA256 hash)
     if (list->count == list->capacity) {
 
         int new_capacity;
-        if (list->capacity == 0) new_capacity = 8;
-        else                     new_capacity = 2 * list->capacity;
+        if (list->capacity == 0)
+            new_capacity = 8;
+        else
+            new_capacity = 2 * list->capacity;
 
         PendingDownload *new_items = sys_malloc(new_capacity * sizeof(PendingDownload));
         if (new_items == NULL)
@@ -276,6 +278,7 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView
 
     // Process add_list: ensure chunks exist and move from orphaned if needed
     for (uint32_t i = 0; i < add_count; i++) {
+
         char main_path[PATH_MAX];
         char orphaned_path[PATH_MAX];
 
@@ -679,6 +682,7 @@ process_client_upload_chunk(ChunkServer *state, int conn_idx, ByteView msg)
         return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_UPLOAD_CHUNK_ERROR, S("Invalid message"));
 
     string data = { (char*) reader.src + reader.cur, data_len };
+    // TODO: Shoudn't we skip the data here? binary_read(&reader, NULL, data_len)
 
     // Check that there are no more bytes to read
     if (binary_read(&reader, NULL, 1))
@@ -757,10 +761,17 @@ static int
 process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView msg)
 {
     switch (type) {
-        case MESSAGE_TYPE_CREATE_CHUNK: return process_client_create_chunk(state, conn_idx, msg);
-        case MESSAGE_TYPE_UPLOAD_CHUNK: return process_client_upload_chunk(state, conn_idx, msg);
-        case MESSAGE_TYPE_DOWNLOAD_CHUNK: return process_client_download_chunk(state, conn_idx, msg);
-        default:break;
+        case MESSAGE_TYPE_CREATE_CHUNK:
+        return process_client_create_chunk(state, conn_idx, msg);
+
+        case MESSAGE_TYPE_UPLOAD_CHUNK:
+        return process_client_upload_chunk(state, conn_idx, msg);
+
+        case MESSAGE_TYPE_DOWNLOAD_CHUNK:
+        return process_client_download_chunk(state, conn_idx, msg);
+
+        default:
+        break;
     }
     return -1;
 }
@@ -768,8 +779,6 @@ process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView
 static void
 start_connecting_to_metadata_server(ChunkServer *state)
 {
-    printf("Chunk server is connecting to the metadata server\n");
-
     Time current_time = get_current_time();
 
     ByteQueue *output;
@@ -900,13 +909,11 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled
         switch (events[i].type) {
 
             case EVENT_CONNECT:
-            printf("New connection to chunk server\n");
             if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER)
-                state->disconnect_time = 0;
+                state->disconnect_time = INVALID_TIME;
             break;
 
             case EVENT_DISCONNECT:
-            printf("Dropped connection to chunk server\n");
             switch (tcp_get_tag(&state->tcp, conn_idx)) {
 
                 case TAG_METADATA_SERVER:
@@ -939,8 +946,6 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled
                         break;
                     }
 
-                    printf("Processing message to chunk server\n");
-
                     switch (tcp_get_tag(&state->tcp, conn_idx)) {
                         case TAG_METADATA_SERVER:
                         ret = process_metadata_server_message(state, conn_idx, msg_type, msg);

src/client.c

@@ -116,6 +116,7 @@ typedef struct {
     uint32_t chunk_size;
     UploadSchedule *uploads;
     int num_uploads;
+    int cap_uploads;
 
     TinyDFS_Result result;
 } Operation;
@@ -1107,6 +1108,37 @@ static int count_pending_uploads(TinyDFS *tdfs, int opidx)
     return n;
 }
 
+static int schedule_upload(TinyDFS *tdfs, int opidx, UploadSchedule upload)
+{
+    if (tdfs->operations[opidx].num_uploads == tdfs->operations[opidx].cap_uploads) {
+
+        int new_cap_uploads;
+        if (tdfs->operations[opidx].uploads == NULL)
+            new_cap_uploads = 0;
+        else
+            new_cap_uploads = 2 * tdfs->operations[opidx].cap_uploads;
+
+        UploadSchedule *uploads = sys_malloc(new_cap_uploads * sizeof(UploadSchedule));
+        if (uploads == NULL)
+            return -1;
+
+        if (tdfs->operations[opidx].num_uploads > 0) {
+            memcpy(
+                uploads,
+                tdfs->operations[opidx].uploads,
+                tdfs->operations[opidx].num_uploads * sizeof(UploadSchedule)
+            );
+            free(tdfs->operations[opidx].uploads);
+        }
+
+        tdfs->operations[opidx].uploads = uploads;
+        tdfs->operations[opidx].cap_uploads = new_cap_uploads;
+    }
+
+    tdfs->operations[opidx].uploads[tdfs->operations[opidx].num_uploads++] = upload;
+    return 0;
+}
+
 static void process_event_for_write(TinyDFS *tdfs,
     int opidx, int request_tag, ByteView msg)
 {
@@ -1153,18 +1185,18 @@ static void process_event_for_write(TinyDFS *tdfs,
             return;
         }
 
+        uint32_t num_all_hasehs = (tdfs->operations[opidx].len + tdfs->operations[opidx].chunk_size - 1) / tdfs->operations[opidx].chunk_size;
+        uint32_t num_new_hashes = num_all_hasehs - num_hashes;
+
         tdfs->operations[opidx].num_hashes = num_hashes; // TODO: overflow
         tdfs->operations[opidx].hashes = sys_malloc(num_hashes * sizeof(SHA256));
         if (tdfs->operations[opidx].hashes == NULL) {
             // TODO
         }
 
-        // 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
-        tdfs->operations[opidx].uploads = sys_malloc(num_hashes * MAX_CHUNK_SERVERS * MAX_SERVER_ADDRS * sizeof(UploadSchedule));
-        if (tdfs->operations[opidx].uploads == NULL) {
-            // TODO
-        }
+        tdfs->operations[opidx].uploads = NULL;
+        tdfs->operations[opidx].num_uploads = 0;
+        tdfs->operations[opidx].cap_uploads = 0;
 
         char *full_ptr = tdfs->operations[opidx].ptr;
         int   full_off = tdfs->operations[opidx].off;
@@ -1235,7 +1267,9 @@ static void process_event_for_write(TinyDFS *tdfs,
                     upload.src = src;
                     upload.off = off;
                     upload.len = len;
-                    tdfs->operations[opidx].uploads[tdfs->operations[opidx].num_uploads++] = upload;
+                    if (schedule_upload(tdfs, opidx, upload) < 0) {
+                        // TODO
+                    }
                 }
 
                 uint32_t num_ipv6;
@@ -1268,8 +1302,9 @@ static void process_event_for_write(TinyDFS *tdfs,
                     upload.src = src;
                     upload.off = off;
                     upload.len = len;
-                    tdfs->operations[opidx].uploads[tdfs->operations[opidx].num_uploads++] = upload;
-                }
+                    if (schedule_upload(tdfs, opidx, upload) < 0) {
+                        // TODO
+                    }                }
             }
         }
 
@@ -1281,6 +1316,9 @@ static void process_event_for_write(TinyDFS *tdfs,
 
         for (uint32_t i = 0; i < num_locations; i++) {
 
+            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 };
@@ -1301,7 +1339,37 @@ static void process_event_for_write(TinyDFS *tdfs,
                     return;
                 }
 
+                int old_relative_off = relative_off;
+
+                for (int w = 0; w < num_new_hashes; w++) {
+
+                    char *src = full_ptr + relative_off;
+
+                    int off = 0;
+                    if (num_hashes == 0 && w == 0)
+                       off = full_off % chunk_size;
+
+                    int len = full_len - relative_off;
+                    if (len > chunk_size)
+                        len = chunk_size;
+
+                    relative_off += len;
+
+                    UploadSchedule upload;
+                    upload.status = UPLOAD_WAITING;
+                    upload.server_lid = server_lid;
+                    upload.address.is_ipv4 = true;
+                    upload.address.ipv4 = ipv4;
+                    upload.address.port = port;
+                    upload.hash_lid = -1;
+                    upload.src = src;
+                    upload.off = off;
+                    upload.len = len;
+                    if (schedule_upload(tdfs, opidx, upload) < 0) {
                         // TODO
+                    }                }
+
+                relative_off = old_relative_off;
             }
 
             uint32_t num_ipv6;
@@ -1312,6 +1380,8 @@ static void process_event_for_write(TinyDFS *tdfs,
 
             for (uint32_t k = 0; k < num_ipv6; k++) {
 
+                char *src = full_ptr + relative_off;
+
                 IPv6 ipv6;
                 if (!binary_read(&reader, &ipv6, sizeof(ipv6))) {
                     tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
@@ -1324,7 +1394,35 @@ static void process_event_for_write(TinyDFS *tdfs,
                     return;
                 }
 
+                int old_relative_off = relative_off;
+
+                for (int w = 0; w < num_new_hashes; w++) {
+
+                    int off = 0;
+                    if (num_hashes == 0 && w == 0)
+                       off = full_off % chunk_size;
+
+                    int len = full_len - relative_off;
+                    if (len > chunk_size)
+                        len = chunk_size;
+
+                    relative_off += len;
+
+                    UploadSchedule upload;
+                    upload.status = UPLOAD_WAITING;
+                    upload.server_lid = server_lid;
+                    upload.address.is_ipv4 = false;
+                    upload.address.ipv6 = ipv6;
+                    upload.address.port = port;
+                    upload.hash_lid = -1;
+                    upload.src = src;
+                    upload.off = off;
+                    upload.len = len;
+                    if (schedule_upload(tdfs, opidx, upload) < 0) {
                         // TODO
+                    }                }
+
+                relative_off = old_relative_off;
             }
         }
 

src/file_tree.c

@@ -423,14 +423,19 @@ int file_tree_read(FileTree *ft, string path,
     uint64_t first_chunk_index = off / f->chunk_size;
     uint64_t  last_chunk_index = (off + len - 1) / f->chunk_size;
 
+    if (first_chunk_index >= f->num_chunks)
+        return 0;
+
+    if (last_chunk_index >= f->num_chunks) {
+        if (f->num_chunks == 0)
+            return 0;
+        last_chunk_index = f->num_chunks-1;
+    }
+
     int num_hashes = 0;
     for (uint32_t i = first_chunk_index; i <= last_chunk_index; i++) {
 
-        SHA256 hash;
-        if (i >= f->num_chunks)
-            hash = ZERO_HASH;
-        else
-            hash = f->chunks[i];
+        SHA256 hash = f->chunks[i];
 
         if (num_hashes < max_hashes)
             hashes[num_hashes] = hash;

src/main_test.c

@@ -35,7 +35,6 @@ int main(int argc, char **argv)
     // Spawn simulation client
     spawn_simulated_process("--client --remote-addr 127.0.0.1 --remote-port 8080");
 
-    printf("Running simulation (press Ctrl+C to stop)...\n");
     while (!simulation_should_stop)
         update_simulation();
 

src/metadata_server.c

@@ -28,7 +28,11 @@ static int hash_list_insert(HashList *hash_list, SHA256 hash)
 
     if (hash_list->count == hash_list->capacity) {
 
-        int new_capacity = hash_list->capacity ? hash_list->capacity * 2 : 16;
+        int new_capacity;
+        if (hash_list->items == NULL)
+            new_capacity = 16;
+        else
+            new_capacity = 2 * hash_list->capacity;
 
         SHA256 *new_items = sys_realloc(hash_list->items, new_capacity * sizeof(SHA256));
         if (new_items == NULL)
@@ -448,9 +452,9 @@ process_client_read(MetadataServer *state, int conn_idx, ByteView msg)
 
     } else {
 
-        MessageWriter writer;
-
         ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
+
+        MessageWriter writer;
         message_writer_init(&writer, output, MESSAGE_TYPE_READ_SUCCESS);
 
         if (chunk_size > UINT32_MAX) {
@@ -659,8 +663,6 @@ chunk_server_from_conn(MetadataServer *state, int conn_idx)
 static int process_chunk_server_auth(MetadataServer *state,
     int conn_idx, ByteView msg)
 {
-    printf("Received auth message from chunk server\n"); // TODO
-
     ChunkServerPeer *chunk_server = chunk_server_from_conn(state, conn_idx);
     chunk_server->num_addrs = 0;
 
@@ -810,13 +812,11 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
         switch (events[i].type) {
 
             case EVENT_CONNECT:
-            printf("New connection to metadata server\n");
             tcp_set_tag(&state->tcp, conn_idx, CONNECTION_TAG_UNKNOWN);
             break;
 
             case EVENT_DISCONNECT:
             {
-                printf("Dropped connection to metadata server\n");
                 int tag = tcp_get_tag(&state->tcp, conn_idx);
                 if (tag >= 0) {
                     chunk_server_peer_free(&state->chunk_servers[tag]);
@@ -839,11 +839,8 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
                         break;
                     }
 
-                    printf("Processing message to metadata server\n");
-
                     if (tcp_get_tag(&state->tcp, conn_idx) == CONNECTION_TAG_UNKNOWN) {
                         if (is_chunk_server_message_type(msg_type)) {
-                            printf("Metadata server determined peer is a chunk server\n"); // TODO
                             int chunk_server_idx = state->num_chunk_servers++;
                             chunk_server_peer_init(&state->chunk_servers[chunk_server_idx]);
                             tcp_set_tag(&state->tcp, conn_idx, chunk_server_idx);

src/system.c

@@ -636,8 +636,6 @@ void update_simulation(void)
 
             assert(current_time <= wakeup_time);
             current_time = wakeup_time;
-
-            //printf("T=%2.2f ms\n", (float) current_time / 1000000);
         }
 
         int timeout = -1;