Progress

src/client.c

@@ -20,6 +20,10 @@
 
 #define TAG_RETRIEVE_METADATA_FOR_READ  1
 #define TAG_RETRIEVE_METADATA_FOR_WRITE 2
+#define TAG_COMMIT_WRITE 3
+
+#define TAG_UPLOAD_CHUNK_MIN 1000
+#define TAG_UPLOAD_CHUNK_MAX 2000
 
 typedef struct {
     SHA256   hash;
@@ -127,9 +131,21 @@ typedef struct {
 } MetadataServer;
 
 typedef struct {
+
     bool used;
-    Address      addr;
+
+    // List of addresses associated to this chunk server
+    int num_addrs;
+    Address addrs[MAX_CHUNK_SERVER_ADDR];
+
+    // Index of the address currently in use
+    int current_addr_idx;
+
+    // If the connection was established
+    bool connected;
+
     RequestQueue reqs;
+
 } ChunkServer;
 
 struct TinyDFS {
@@ -248,39 +264,71 @@ request_queue_pop(RequestQueue *reqs, Request *req)
     return 0;
 }
 
+static bool
+have_insertection(Address *a, int a_num, Address *b, int b_num)
+{
+    for (int i = 0; i < a_num; i++)
+        for (int j = 0; j < b_num; j++)
+            if (addr_eql(a[i], b[j]))
+                return true;
+    return false;
+}
+
 // Get or create connection to a chunk server
-static int get_chunk_server_connection(TinyDFS *tdfs, Address addr)
+static ByteQueue *get_chunk_server_output_buffer(TinyDFS *tdfs, Address *addrs, int num_addrs)
 {
     // Check if already connected
-    for (int i = 0; i < tdfs->num_chunk_servers; i++) {
-        if (tdfs->chunk_servers[i].used && addr_eql(tdfs->chunk_servers[i].addr, addr)) {
-            int conn_idx = tcp_index_from_tag(&tdfs->tcp, i);
-            if (conn_idx >= 0)
-                return i;
-        }
-    }
 
-    // Find free slot
+    int found = -1;
-    int idx = -1;
+    for (int i = 0; i < tdfs->num_chunk_servers; i++) {
-    for (int i = 0; i < MAX_CHUNK_SERVERS; i++) {
+
-        if (!tdfs->chunk_servers[i].used) {
+        if (!tdfs->chunk_servers[i].used)
-            idx = i;
+           continue;
+
+        if (have_insertection(addrs, num_addrs, tdfs->chunk_servers[i].addrs, tdfs->chunk_servers[i].num_addrs)) {
+            found = i;
             break;
         }
     }
-    if (idx < 0) return -1;
 
-    // Connect
+    ByteQueue *output;
-    if (tcp_connect(&tdfs->tcp, addr, idx, NULL) < 0)
+    if (found == -1) {
-        return -1;
+
+        if (tdfs->num_chunk_servers == MAX_CHUNK_SERVERS)
+            return NULL;
+
+        // Find free slot
+        found = 0;
+        while (tdfs->chunk_servers[found].used)
+            found++;
+
+        if (num_addrs > MAX_CHUNK_SERVER_ADDR)
+            num_addrs = MAX_CHUNK_SERVER_ADDR;
+        tdfs->chunk_servers[found].num_addrs = num_addrs;
+        memcpy(tdfs->chunk_servers[found].addrs, addrs, num_addrs * sizeof(Address));
+
+        tdfs->chunk_servers[found].used = true;
+        tdfs->chunk_servers[found].current_addr_idx = 0;
+        tdfs->chunk_servers[found].connected = false;
+
+        request_queue_init(&tdfs->chunk_servers[found].reqs);
+
+        if (tcp_connect(&tdfs->tcp, addr, found, &output) < 0)
+            return NULL;
 
         // Initialize
         tdfs->chunk_servers[idx].used = true;
         tdfs->chunk_servers[idx].addr = addr;
         request_queue_init(&tdfs->chunk_servers[idx].reqs);
         tdfs->num_chunk_servers++;
+    } else {
+        int conn_idx = tcp_index_from_tag(&tdfs->tcp, found);
+        assert(conn_idx > -1);
 
-    return idx;
+        output = tcp_output_buffer(&tdfs->tcp, conn_idx);
+    }
+
+    return output;
 }
 
 // Send download request for a chunk
@@ -973,14 +1021,10 @@ static int start_upload(Operation *o)
     if (found < 0)
         return -1; // No upload can be started at this time
 
-    int ret = get_chunk_server_connection(tdfs, o->uploads[found].address);
+    ByteQueue *output = get_chunk_server_output_buffer(tdfs, &o->uploads[found].address, 1);
-
+    if (output == NULL) {
-    int tag = xxx;
+        // TODO
-
+    }
-    ByteQueue *output;
-    int ret = tcp_connect(tcp, o->uploads[found].address, tag, &output);
-    if (ret < 0)
-        return -1;
 
     if (o->uploads[found].no_hash) {
 
@@ -1019,6 +1063,8 @@ static int start_upload(Operation *o)
         }
     }
 
+    // TODO: push request tag to chunk server request queue
+
     o->uploads[found].status = UPLOAD_PENDING;
     return 0;
 }
@@ -1031,10 +1077,8 @@ static void process_event_for_write(TinyDFS *tdfs,
         return;
     }
 
-    switch (request_tag) {
+    if (request_tag == TAG_RETRIEVE_METADATA_FOR_WRITE) {
 
-        case TAG_RETRIEVE_METADATA_FOR_WRITE:
-        {
         // We are expecting one of:
         //   MESSAGE_TYPE_READ_ERROR
         //   MESSAGE_TYPE_READ_SUCCESS
@@ -1287,16 +1331,18 @@ static void process_event_for_write(TinyDFS *tdfs,
         // The client should not try any random N chunk servers
         // for upload. It must try all chunk servers until N respond
 
+    } else if (request_tag >= TAG_UPLOAD_CHUNK_MIN && request_tag <= TAG_UPLOAD_CHUNK_MAX) {
+
+        // TODO
+
+    } else {
+
+        assert(request_tag == TAG_COMMIT_WRITE);
+
+        // TODO
+
         tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
     }
-        return;
-
-        default:
-        break;
-    }
-
-    // Write operation processing not fully implemented
-    tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR };
 }
 
 static void process_event(TinyDFS *tdfs,
@@ -1354,23 +1400,67 @@ int tinydfs_process_events(TinyDFS *tdfs, void **contexts, struct pollfd *polled
         switch (events[i].type) {
 
             case EVENT_CONNECT:
+            {
+                int tag = tcp_get_tag(&tdfs->tcp, conn_idx);
+                if (tag != TAG_METADATA_SERVER)
+                    tdfs->chunk_servers[tag].connected = true;
+            }
             break;
 
             case EVENT_DISCONNECT:
             {
-                RequestQueue *reqs;
+                // A TCP connection was just dropped.
+                // For clients, connections can be:
+                //   1. To the metadata server
+                //   2. or to a chunk server
+                // If requests were buffered for the metadata
+                // or chunk server, they are considered as failed
+                // and their failure event is processed.
+                //
+                // If a chunk server was never connected,
+                // then it's possible that using a different
+                // address will allow connecting succesfully
+                // and send the buffered messages. Therefore,
+                // if a chunk server wasn't connected and
+                // there are addresses to try, the messages
+                // are not dropped and a new connect process
+                // is started.
+
+                RequestQueue *reqs = NULL;
 
                 int tag = tcp_get_tag(&tdfs->tcp, conn_idx);
                 if (tag == TAG_METADATA_SERVER)
                     reqs = &tdfs->metadata_server.reqs;
                 else {
                     assert(tag > -1);
+
+                    if (tdfs->chunk_servers[tag].connected)
                         reqs = &tdfs->chunk_servers[tag].reqs;
+                    else {
+
+                        tdfs->chunk_servers[tag].current_addr_idx++;
+
+                        bool started = false;
+                        while (tdfs->chunk_servers[tag].current_addr_idx < tdfs->chunk_servers[tag].num_addrs) {
+
+                            if (tcp_connect(&tdfs->tcp, tdfs->chunk_servers[tag].addrs[addr_idx], tag, NULL) == 0) {
+                                started = true;
+                                break;
                             }
 
+                            tdfs->chunk_servers[tag].current_addr_idx++;
+                        }
+
+                        if (started)
+                            reqs = &tdfs->chunk_servers[tag].reqs;
+                    }
+                }
+
+                if (reqs) {
                     for (Request req; request_queue_pop(reqs, &req) == 0; )
                         process_event(tdfs, req.opidx, req.tag, (ByteView) { NULL, 0 });
                 }
+            }
             break;
 
             case EVENT_MESSAGE: