Progress

src/client.c

@@ -27,7 +27,6 @@
 #define TAG_UPLOAD_CHUNK_MAX 2000
 
 #define PARALLEL_LIMIT 5
-#define REPLICATION_FACTOR 3
 
 typedef struct {
     SHA256   hash;
@@ -1535,23 +1534,23 @@ static void process_event_for_write(TinyDFS *tdfs,
         }
 
         SHA256 hash;
-        uint16_t expected_type;
-        if (tdfs->operations[opidx].uploads[found].chunk_index >= tdfs->operations[opidx].num_hashes)
-            expected_type = MESSAGE_TYPE_CREATE_CHUNK_SUCCESS;
-        else {
-            expected_type = MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS;
-
         if (!binary_read(&reader, &hash, sizeof(hash))) {
             // TODO
             return;
         }
-        }
+
         // Check that there is nothing else to read
         if (binary_read(&reader, NULL, 1)) {
             // TODO
             return;
         }
 
+        uint16_t expected_type;
+        if (tdfs->operations[opidx].uploads[found].chunk_index >= tdfs->operations[opidx].num_hashes)
+            expected_type = MESSAGE_TYPE_CREATE_CHUNK_SUCCESS;
+        else
+            expected_type = MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS;
+
         if (type != expected_type) {
             tdfs->operations[opidx].uploads[found].status = UPLOAD_FAILED;
         } else {
@@ -1591,7 +1590,8 @@ static void process_event_for_write(TinyDFS *tdfs,
             typedef struct {
                 SHA256  old_hash;
                 SHA256  new_hash;
-                int replication;
+                int     num_locations;
+                Address locations[REPLICATION_FACTOR];
             } ChunkUploadResult;
 
             int num_upload_results = tdfs->operations[opidx].num_chunks;
@@ -1602,21 +1602,24 @@ static void process_event_for_write(TinyDFS *tdfs,
 
             for (int i = 0; i < num_upload_results; i++) {
                 upload_results[i].old_hash = tdfs->operations[opidx].hashes[i];
-                upload_results[i].replication = 0;
+                upload_results[i].num_locations = 0;
             }
 
-            for (int i = 0; i < tdfs->operations[opidx].num_uploads; i++)
+            for (int i = 0; i < tdfs->operations[opidx].num_uploads; i++) {
-                if (tdfs->operations[opidx].uploads[i].status == UPLOAD_COMPLETED) {
+                UploadSchedule *u = &tdfs->operations[opidx].uploads[i];
-                    upload_results[tdfs->operations[opidx].uploads[i].chunk_index].new_hash = tdfs->operations[opidx].uploads[i].final_hash;
+                if (u->status == UPLOAD_COMPLETED) {
-                    upload_results[tdfs->operations[opidx].uploads[i].chunk_index].replication++;
+                    int n = upload_results[u->chunk_index].num_locations++;
+                    upload_results[u->chunk_index].locations[n] = u->address;
+                    upload_results[u->chunk_index].new_hash = u->final_hash;
+                }
             }
 
             // Now check that each chunk is replicated
             // at least N times
 
-            bool ok = false;
+            bool ok = true;
             for (int i = 0; i < num_upload_results; i++) {
-                if (upload_results[i].replication < REPLICATION_FACTOR) {
+                if (upload_results[i].num_locations < REPLICATION_FACTOR) {
                     ok = false;
                     break;
                 }
@@ -1649,9 +1652,24 @@ static void process_event_for_write(TinyDFS *tdfs,
             message_write(&writer, &num_chunks, sizeof(num_chunks));
 
             for (int i = 0; i < num_upload_results; i++) {
+
+                // TODO: newly create chunks don't have an old hash
                 message_write(&writer, &upload_results[i].old_hash, sizeof(upload_results[i].old_hash));
                 message_write(&writer, &upload_results[i].new_hash, sizeof(upload_results[i].new_hash));
-                // TODO
+
+                uint32_t tmp = upload_results[i].num_locations;
+                message_write(&writer, &tmp, sizeof(tmp));
+
+                for (int j = 0; j < upload_results[i].num_locations; j++) {
+
+                    Address addr = upload_results[i].locations[j];
+
+                    uint8_t is_ipv4 = addr.is_ipv4;
+                    message_write(&writer, &is_ipv4, sizeof(is_ipv4));
+                    if (addr.is_ipv4) message_write(&writer, &addr.ipv4, sizeof(addr.ipv4));
+                    else              message_write(&writer, &addr.ipv6, sizeof(addr.ipv6));
+                    message_write(&writer, &addr.port, sizeof(addr.port));
+                }
             }
 
             free(upload_results);

src/config.h

@@ -6,4 +6,6 @@
 #define MAX_OPERATIONS 128
 #define MAX_REQUESTS_PER_QUEUE 128
 
+#define REPLICATION_FACTOR 3
+
 #endif // CONFIG_INCLUDED

src/metadata_server.c

@@ -558,9 +558,14 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
 
     #define MAX_CHUNKS_PER_WRITE 32
 
-    Address addrs[MAX_CHUNKS_PER_WRITE];
+    typedef struct {
-    SHA256 new_hashes[MAX_CHUNKS_PER_WRITE];
+        SHA256  old_hash;
-    SHA256 old_hashes[MAX_CHUNKS_PER_WRITE];
+        SHA256  new_hash;
+        int     num_addrs;
+        Address addrs[REPLICATION_FACTOR];
+    } ChunkWriteResult;
+
+    ChunkWriteResult results[MAX_CHUNKS_PER_WRITE];
 
     for (uint32_t i = 0; i < num_chunks; i++) {
 
@@ -572,6 +577,15 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
         if (!binary_read(&reader, &new_hash, sizeof(new_hash)))
             return -1;
 
+        results[i].old_hash = old_hash;
+        results[i].new_hash = new_hash;
+
+        uint32_t num_locations;
+        if (!binary_read(&reader, &num_locations, sizeof(num_locations)))
+            return -1;
+
+        for (uint32_t j = 0; j < num_locations; j++) {
+
             uint8_t is_ipv4;
             if (!binary_read(&reader, &is_ipv4, sizeof(is_ipv4)))
                 return -1;
@@ -590,9 +604,9 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
             if (!binary_read(&reader, &addr.port, sizeof(addr.port)))
                 return -1;
 
-        addrs[i] = addr;
+            if (results[i].num_addrs < REPLICATION_FACTOR)
-        new_hashes[i] = new_hash;
+                results[i].addrs[results[i].num_addrs++] = addr;
-        old_hashes[i] = old_hash;
+        }
     }
 
     // Check that there are no more bytes to read
@@ -603,6 +617,13 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
     SHA256 removed_hashes[MAX_CHUNKS_PER_WRITE];
     int num_removed = 0;
 
+    SHA256 old_hashes[MAX_CHUNKS_PER_WRITE];
+    SHA256 new_hashes[MAX_CHUNKS_PER_WRITE];
+    for (int i = 0; i < num_chunks; i++) {
+        old_hashes[i] = results[i].old_hash;
+        new_hashes[i] = results[i].new_hash;
+    }
+
     int ret = file_tree_write(&state->file_tree, path, offset, length,
         old_hashes, new_hashes, removed_hashes, &num_removed);
 
@@ -626,13 +647,16 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
 
         // Add new chunks to add_list
         for (uint32_t i = 0; i < num_chunks; i++) {
-            int j = find_chunk_server_by_addr(state, addrs[i]);
-            if (j == -1)
-                return -1;
 
-            if (!hash_list_insert(&state->chunk_servers[j].add_list, new_hashes[i]))
+            for (int j = 0; j < results[i].num_addrs; j++) {
+
+                int k = find_chunk_server_by_addr(state, results[i].addrs[j]);
+                if (k == -1) return -1;
+
+                if (hash_list_insert(&state->chunk_servers[k].add_list, new_hashes[i]) < 0)
                     return -1;
             }
+        }
 
         // Mark removed chunks for deletion on all chunk servers that have them
         // These are chunks that were overwritten and are no longer referenced anywhere