Merge pull request #27 from SidoShiro/network-read-1

Network read 1

Author: SidoShiro

include/globals.h

@@ -1,7 +1,7 @@
 #ifndef DISTRIBUTEDMALLOC_GLOBALS_H
 #define DISTRIBUTEDMALLOC_GLOBALS_H
 
-#define DEF_NODE_SIZE 32
+#define DEF_NODE_SIZE 8
 #define DEF_NODE_USER 0
 #define DEF_NODE_LEADER 1
 

src/cli/user.c

@@ -3,10 +3,11 @@
 #include "message.h"
 #include "globals.h"
 
-#include <mpi.h>
 #include <debug.h>
-#include <stdint-gcc.h>
-#include <event.h>
+#include <mpi.h>
+#include <stdint.h>
+#include <sys/types.h>
+#include <unistd.h>
 
 void send_write(void *data, unsigned short leader) {
     MPI_Request r;

src/network/leader.c

@@ -10,6 +10,7 @@
 #include <mpi.h>
 #include <stdint.h>
 #include <communication.h>
+#include <string.h>
 
 size_t get_message_size() {
     return (sizeof(unsigned short) * 3 + 2 * sizeof(size_t) + sizeof(enum operation));
@@ -262,37 +263,54 @@ void execute_read(struct leader_resources *l_r) {
         return;
     }
 
-    // Readed bytes
-    char *read_buff = NULL;
+    // Buffer of all READS bytes
+    char *read_buff = malloc(sizeof(char) * (d_r->size + 1));
 
     // 2 Get the block to write to (Warning to multiple parts allocation)
     //                             (Warning to size bigger than block)
     size_t to_write_address_v = d_r->address;
     size_t nb_read = 0;
-    for (size_t i = part_s; i < c_a->number_parts; i++) {
+    size_t nb_read_size = 0;
+    size_t x = d_r->size;
+    size_t offset = 0;
+    for (size_t i = part_s; x > 0 && i < c_a->number_parts; i++) {
         // TODO handle size overflow
-
-
         struct block *b = c_a->parts[i];
-        // compute size to write for this block
-        size_t to_read_size = d_r->size - b->size;
-        d_r->size -= to_read_size;
+        // compute size to read for this block
+        size_t to_read_size = 0;
+        if (x <= b->size) {
+            to_read_size = x;
+            x = 0;
+        } else {
+            x -= b->size;
+            to_read_size = b->size;
+        }
+
         // compute local address to write
-        size_t local_address = b->virtual_address - to_write_address_v;
-        to_write_address_v += local_address;
+        size_t local_address = 0;
+        if (b->virtual_address == to_write_address_v) {
+            local_address = 0;
+            to_write_address_v += to_read_size;
+        } else {
+            local_address += to_write_address_v - b->virtual_address;
+            to_write_address_v += to_read_size;
+        }
 
+        nb_read_size += to_read_size;
         // 3 Send READ OP to each node (Warning to the local address of the node, not the virtual)
         struct message *m = generate_message(l_r->id, b->id, b->id, local_address, to_read_size, OP_READ);
-        MPI_Send(m, sizeof(struct message), MPI_BYTE, b->id, 5, MPI_COMM_WORLD);
+        debug("Send OP Read", l_r->id);
+        MPI_Send(m, sizeof(struct message), MPI_BYTE, b->id, 3, MPI_COMM_WORLD);
+        void *buff = malloc(sizeof(char) * (to_read_size + 1));
+        MPI_Status st;
+        MPI_Recv(buff, to_read_size, MPI_BYTE, b->id, 4, MPI_COMM_WORLD, &st);
+        memcpy((void*)(read_buff + (offset * sizeof(char))), buff, to_read_size);
+        offset += to_read_size;
         nb_read++;
+        free(m);
     }
-
-    // 4 Receive bytes, append data from all nodes
-    for (size_t i = 0; i < nb_read; i++) {
-        // MPI_Irecv()
-    }
-    (void) read_buff;
-
+    debug("READ AND ASSEMBLED", l_r->id);
+    debug_n(read_buff, l_r->id, d_r->size);
 }
 
 void execute_write(struct leader_resources *l_r) {
@@ -316,6 +334,7 @@ void execute_write(struct leader_resources *l_r) {
     //                             (Warning to size bigger than block)
     size_t to_write_address_v = d_w->address;
     size_t x = d_w->size;
+    size_t offset = 0;
     for (size_t i = part_s; x > 0 && i < c_a->number_parts; i++) {
         // TODO handle size overflow
         struct block *b = c_a->parts[i];
@@ -350,7 +369,8 @@ void execute_write(struct leader_resources *l_r) {
         MPI_Recv(&m2, sizeof(struct message), MPI_BYTE, b->id, 3, MPI_COMM_WORLD, &st);
         debug("Send Data", l_r->id);
         // debug_n(d_w->data, l_r->id, d_w->size);
-        MPI_Send(d_w->data, to_write_size, MPI_BYTE, b->id, 4, MPI_COMM_WORLD);
+        MPI_Send((void*)((char*)d_w->data + offset), to_write_size, MPI_BYTE, b->id, 4, MPI_COMM_WORLD);
+        offset += to_write_size;
     }
 
     if (x > 0) {

src/network/node.c

@@ -32,6 +32,23 @@ void write_on_node(struct node *n, size_t address, char *data, size_t size) {
     }
 }
 
+/**
+ *
+ * @param n
+ * @param address
+ * @param data
+ * @param size
+ */
+void read_on_node(struct node *n, size_t address, char *data, size_t size) {
+    if (address + size <= n->size) {
+        void *mem_op_ptr = (n->memory + address);
+        memcpy((void *) data, (void *) mem_op_ptr, size);
+    }
+    else {
+        debug("OP READ FAILED", n->id);
+    }
+}
+
 void node_cycle(struct node *n) {
     while (1) {
         // cycle of node
@@ -57,6 +74,12 @@ void node_cycle(struct node *n) {
             }
                 break;
             case OP_READ:
+                debug("Read OP", n->id);
+                char *data = malloc(m->size * sizeof(char));
+                read_on_node(n, m->address, data, m->size);
+                debug("Send Read: Data", n->id);
+                MPI_Send(data, m->size, MPI_BYTE, m->id_s, 4, MPI_COMM_WORLD);
+                free(data);
                 break;
             default:
                 break;