Merge pull request #36 from SidoShiro/dev

Dev : Master update

Author: SidoShiro

CMakeLists.txt

@@ -22,6 +22,7 @@ set(SRCS
     src/utils/debug.c
     src/cli/user.c
     src/network/communication.c
+    src/network/data_communication.c
     src/utils/queue.c)
 
 

include/cli.h

@@ -10,7 +10,7 @@
 #include "message.h"
 #include "utils.h"
 
-void start_cli();
+void start_cli(unsigned size);
 
 void send_command(enum operation op, void *data, unsigned short leader);
 

include/communication.h

@@ -8,16 +8,33 @@
 // Number of attempts failed before considering the node as dead.
 #define NB_ITER 60
 
+#define TAG_MSG 3
+#define TAG_DATA 4
+#define TAG_ELECTION 5
+
 /*
  * Send the given message 'm' to the destination node
  * Return 1 if it's a success, 0 if the destination node didn't respond in time
  */
-int send_safe_message(struct message *m_send, struct queue *queue);
+int send_safe_message(struct message *m_send, struct queue *queue, int tag);
 
 /*
  * Receive a message sent with 'send_safe_message'
  * then send a OP_OK to the source node
  */
-struct message *receive_message(struct queue *message_queue);
+struct message *receive_message(struct queue *message_queue, int tag);
+
+/*
+ * Recieve a specific data from a specific node.
+ * All others messages recieved are put in the queue. If needed, a IS_ALIVE is send.
+ * Return the wanted data, which was calloc.
+ */
+void *recieve_data(size_t size, struct queue *queue, unsigned source);
+
+/*
+ * Send message to all nodes except user and itself
+ * m_send->id_t will change according to the target node
+ */
+void broadcast_message(struct message *m_send, unsigned id, unsigned network_size, int tag);
 
 #endif /* !DISTRIBUTEDMALLOC_COMMUNICATION_H */

include/globals.h

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

include/leader.h

@@ -31,4 +31,6 @@ void leader_loop(struct node *n, unsigned short terminal_id, unsigned short nb_n
 
 struct allocation *give_for_v_address(struct leader_resources *l_r, size_t v_address, size_t *part);
 
+size_t size_of_allocation(struct allocation *a);
+
 #endif /* !DISTRIBUTEDMALLOC_LEADER_H */

include/message.h

@@ -9,8 +9,11 @@ enum operation {
     OP_FREE,
     OP_WRITE,
     OP_READ,
+    OP_READ_FILE,
+    OP_TABLE,
     OP_SNAP,
     OP_LEADER,
+    OP_START_LEADER,
     OP_WHOISLEADER,
     OP_REVIVE,
     OP_KILL,
@@ -19,6 +22,7 @@ enum operation {
     OP_DUMP,
     OP_DUMP_ALL,
     OP_LEADER_OK,
+    OP_IS_ALIVE,
     OP_ALIVE,
     OP_LEADER_AGAIN
 };

src/cli/cli.c

@@ -3,7 +3,10 @@
 #include <mpi.h>
 #include <globals.h>
 #include <debug.h>
+#include <communication.h>
+#include <sys/stat.h>
 #include "cli.h"
+#include "communication.h"
 
 char *read_cmd() {
     ssize_t buffer_size = 256;
@@ -84,10 +87,15 @@ void execute(char **args, unsigned short leader) {
     if (0 == strcmp(args[0], "h")) {
         printf("dmalloc commands:\n"
                " h                             | display all available commands with their description   |\n"
+               " t                             | show table of allocations                               |\n"
                " m `size`                      | return `address` to cmd user of the required allocation |\n"
                " f `address`                   | free address, Warning if already free                   |\n"
                " w `address` `datasize` `data` | write at the address the data of size datasize          |\n"
+               " w `address` `file`            | write all content of file at address                    |\n"
+               " w `address` `file` `datasize` | write datasize bytes from file to the address           |\n"
                " r `address` `datasize`        | read datasize bytes at address                          |\n"
+               " r `address` `file`            | read all bytes of the block at address into file        |\n"
+               " r `address` `file` `datasize` | read datasize bytes at address into file                |\n"
                " d `address`                   | dump in as text all data of the block stored in address |\n"
                " d net                         | dump all allocation                                     |\n"
                " d `address` `file`            | dump address data in file                               |\n"
@@ -131,13 +139,17 @@ void execute(char **args, unsigned short leader) {
         size_t address = 0;
         if (1 == sscanf(args[1], "%zu", &address)) {
             printf("Execute Free of address : %zu\n", address);
+            struct data_address *d_a = generate_data_address(address);
+            send_command(OP_FREE, d_a, leader);
+            free(d_a);
         } else {
             error_msg("f  require an argument 'address' which can be casted as a positive integer");
         }
     } else if (0 == strcmp(args[0], "w")) {
         // ERRORS
-        if (l <= 3) {
-            error_msg("w require 3 arguments : 'address', 'datasize' and 'data'");
+        if (l <= 2) {
+            error_msg(
+                    "w require 2-3 arguments : 'address', 'datasize' and 'data'\n OR 'address' 'file' + 'datasize' optional, check h");
             return;
         } else if (l >= 5) {
             error_msg("w do not support more than 3 arguments, check command h");
@@ -149,23 +161,56 @@ void execute(char **args, unsigned short leader) {
         size_t datasize = 0;
         if (1 == sscanf(args[1], "%zu", &address)) {
             if (1 == sscanf(args[2], "%zu", &datasize)) {
+                if (args[3] == NULL) {
+                    error_msg("w cannot write nothing");
+                    return;
+                }
                 printf("Execute Write at %zu of %s : %zu bytes\n", address, args[3], datasize);
                 struct data_write *d_w = generate_data_write(address, datasize, args[3]);
                 send_command(OP_WRITE, d_w, leader);
                 free(d_w);
             } else {
-                error_msg("w requires an argument 'datasize' which can be casted as a positive integer");
+                // Write from File
+                FILE *file_write = fopen(args[2], "r");
+                if (!file_write) {
+                    error_msg("w 'address' 'file', file was not possible to open, or doesn't exist");
+                    return;
+                }
+                struct stat st;
+                /*get the size using stat()*/
+                if (stat(args[2], &st) != 0)
+                    return;
+                if (l == 4) {
+                    // file max data read
+                    if (1 == sscanf(args[3], "%zu", &datasize)) {
+                        if ((ssize_t) datasize >= st.st_size)
+                            datasize = st.st_size;
+                    } else {
+                        error_msg("w requires an argument 'datasize' which can be casted as a positive integer");
+                        return;
+                    }
+                } else {
+                    datasize = st.st_size;
+                }
+                // Now read datasize
+                char *buffer_write_file = malloc(sizeof(char) * (2 + datasize));
+                size_t r_c = fread(buffer_write_file, sizeof(char), datasize, file_write);
+                if (r_c != datasize)
+                    error_msg("WARNING: file bytes read different from bytes asked");
+                struct data_write *d_w = generate_data_write(address, datasize, buffer_write_file);
+                send_command(OP_WRITE, d_w, leader);
+                // OLD error_msg("w requires an argument 'datasize' which can be casted as a positive integer");
             }
         } else {
             error_msg("w requires an argument 'address' which can be casted as a positive integer");
         }
     } else if (0 == strcmp(args[0], "r")) {
         // ERRORS
         if (l <= 2) {
-            error_msg("r requires 2 arguments : 'address' and 'datasize'");
+            error_msg("r requires 2-3 arguments : 'address' and 'datasize'\n OR 'address' 'file'\n OR 'address' 'file' 'datasize");
             return;
-        } else if (l >= 4) {
-            error_msg("r do not support more than 2 arguments, check command h");
+        } else if (l >= 5) {
+            error_msg("r do not support more than 2-3 arguments, check command h");
             return;
         }
 
@@ -179,7 +224,38 @@ void execute(char **args, unsigned short leader) {
                 send_command(OP_READ, d_r, leader);
                 free(d_r);
             } else {
-                error_msg("r requires an argument 'datasize' which can be casted as a positive integer");
+                // Read to File
+                FILE *file_read = fopen(args[2], "w");
+                if (!file_read) {
+                    error_msg("r 'address' 'file', file was not possible to open, or doesn't exist");
+                    return;
+                }
+                struct stat st;
+                if (stat(args[2], &st) != 0)
+                    return;
+                if (l == 4) {
+                    // file max data read
+                    if (1 == sscanf(args[3], "%zu", &datasize)) {
+                        if ((ssize_t) datasize >= st.st_size)
+                            datasize = st.st_size;
+                    } else {
+                        error_msg("r requires an argument 'datasize' which can be casted as a positive integer");
+                        return;
+                    }
+                } else {
+                    datasize = st.st_size;
+                }
+                // Now read datasize bytes
+                char *buffer_read_file = NULL;
+                struct data_write *d_w = generate_data_write(address, datasize, buffer_read_file);
+                send_command(OP_READ_FILE, d_w, leader);
+                // /!\ buffer_read_file was allocated in send_command
+                debug("Write data in file (READ OP)", 0);
+                printf("Data Read :: %zu", d_w->size);
+                size_t r_c = fwrite(d_w->data, sizeof(char), d_w->size, file_read);
+                if (r_c != d_w->size)
+                    error_msg("WARNING: file bytes write different from bytes asked");
+                // OLD error_msg("r requires an argument 'datasize' which can be casted as a positive integer");
             }
         } else {
             error_msg("r require an argument 'address' which can be casted as a positive integer");
@@ -239,28 +315,14 @@ void execute(char **args, unsigned short leader) {
         } else {
             error_msg("w require an argument 'address' which can be casted as a positive integer");
         }
-    } else if (0 == strcmp(args[0], "w")) {
+    } else if (0 == strcmp(args[0], "t")) {
         // ERRORS
-        if (l <= 3) {
-            error_msg("w require 3 arguments : 'address', 'datasize' and 'data'");
-            return;
-        } else if (l >= 5) {
-            error_msg("w do not support more than 3 arguments, check command h");
+        if (l != 1) {
+            error_msg("t require no arguments");
             return;
         }
+        send_command(OP_TABLE, NULL, leader);
 
-        // Execution
-        size_t address = 0;
-        size_t datasize = 0;
-        if (1 == sscanf(args[1], "%zu", &address)) {
-            if (1 == sscanf(args[2], "%zu", &datasize)) {
-                printf("Execute Write at %zu of %s : %zu bytes\n", address, args[3], datasize);
-            } else {
-                error_msg("w require an argument 'datasize' which can be casted as a positive integer");
-            }
-        } else {
-            error_msg("w require an argument 'address' which can be casted as a positive integer");
-        }
     } else if (0 == strcmp(args[0], "w")) {
         // ERRORS
         if (l <= 3) {
@@ -283,23 +345,31 @@ void execute(char **args, unsigned short leader) {
         } else {
             error_msg("w require an argument 'address' which can be casted as a positive integer");
         }
+    } else {
+        error_msg("command not found, see 'h' for help");
     }
 
 }
 
-unsigned short get_leader() {
+unsigned short start_leader_election(unsigned size) {
+    struct message *b = generate_message(0, 0, 0, 0, 0, OP_START_LEADER);
+    broadcast_message(b, 0, size, TAG_MSG);
+    free(b);
     struct message *m = generate_message(0, 0, 0, 0, 0, OP_NONE);
-    MPI_Recv(m, sizeof(*m), MPI_BYTE, MPI_ANY_SOURCE, 201, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
+    MPI_Recv(m, sizeof(*m), MPI_BYTE, MPI_ANY_SOURCE, TAG_MSG, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
     debug("Got Leader !", DEF_NODE_USER);
-    return m->id_o;
+    unsigned short leader = m->id_o;
+    free(m);
+    return leader;
 }
 
-void start_cli() {
+void start_cli(unsigned size) {
     int no_quit = 1;
     char *cmd;
     char **args;
 
-    unsigned short leader = get_leader();
+    struct queue *q = queue_init();
+    unsigned short leader = start_leader_election(size);
 
     while (no_quit) {
         fflush(0);
@@ -315,7 +385,10 @@ void start_cli() {
         }
 
         // DEBUG print_args(args);
-
+        struct message *m_verif = generate_message_a(0, leader, 0, 0, 0, OP_IS_ALIVE, 1);
+        if (!send_safe_message(m_verif, q, TAG_MSG))
+            leader = start_leader_election(size);
+        free(m_verif);
         execute(args, leader);
 
         free(cmd);