implemented parts of shared memory usage

CMakeLists.txt

@@ -1,7 +1,9 @@
 cmake_minimum_required(VERSION 3.3)
 project (Mycorrhiza)
 #add_definitions(--std=c99)
+set(CMAKE_EXE_LINKER_FLAGS "-lrt")
 
 add_executable(mycoagent test/mycoagent.c)
 add_executable(mycoagent2 test/mycoagent2.c)
 add_executable(mycodaemon test/mycodaemon.c)
+#target_link_libraries(mycoagent rt)

src/myco-agent.c

@@ -63,7 +63,11 @@ int myco_agent_unregister (const char *agent_name, int agent_message_queue_id) {
     return 0;
 }
 
-int myco_agent_register_resource (int agent_message_queue_id, const char *agent_name, const char *resource_name, int resource_transactional, pid_t pid, int size, void* pointer) {
+void *myco_agent_register_resource (int agent_message_queue_id, const char *agent_name, const char *resource_name, int resource_transactional, pid_t pid, int size) {
+    int file_descriptor = shm_open(resource_name, O_CREAT | O_TRUNC | O_RDWR, 0666);
+    int r = ftruncate(file_descriptor, size);
+    void *pointer = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, file_descriptor, 0);
+
     message msg;
     msg.agent_message_queue_id = agent_message_queue_id;
     sprintf(msg.agent_name, "%s", agent_name);
@@ -81,10 +85,10 @@ int myco_agent_register_resource (int agent_message_queue_id, const char *agent_
     }
 
     if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
-        return -1;
+        return NULL;
     }
 
-    return 0;
+    return pointer;
 }
 
 int myco_agent_unregister_resource (int agent_message_queue_id, const char *resource_name) {
@@ -103,13 +107,16 @@ int myco_agent_unregister_resource (int agent_message_queue_id, const char *reso
         return -1;
     }
 
+    int r = shm_unlink(resource_name);
+
     return 0;
 }
 
-int myco_agent_lock_resource (int agent_message_queue_id, const char *resource_name) {
+int myco_agent_lock_resource (int agent_message_queue_id, const char *agent_name, const char *resource_name) {
     message msg;
     msg.agent_message_queue_id = agent_message_queue_id;
     sprintf(msg.resource_name, "%s", resource_name);
+    sprintf(msg.agent_name, "%s", agent_name);
     sprintf(msg.message, "LOCK RESOURCE\n");
 
     msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);
@@ -125,7 +132,7 @@ int myco_agent_lock_resource (int agent_message_queue_id, const char *resource_n
     return 0;
 }
 
-int myco_agent_release_resource (int agent_message_queue_id, const char *resource_name) {
+int myco_agent_release_resource (int agent_message_queue_id, const char *resource_name, void *resource_pointer, int resource_size) {
     message msg;
     msg.agent_message_queue_id = agent_message_queue_id;
     sprintf(msg.resource_name, "%s", resource_name);
@@ -141,13 +148,16 @@ int myco_agent_release_resource (int agent_message_queue_id, const char *resourc
         return -1;
     }
 
+    int r = munmap(resource_pointer, resource_size);
+
     return 0;
 }
 
-int myco_agent_unlock_resource (int agent_message_queue_id, const char *resource_name) {
+int myco_agent_unlock_resource (int agent_message_queue_id, const char *agent_name, const char *resource_name) {
     message msg;
     msg.agent_message_queue_id = agent_message_queue_id;
     sprintf(msg.resource_name, "%s", resource_name);
+    sprintf(msg.agent_name, "%s", agent_name);
     sprintf(msg.message, "UNLOCK RESOURCE\n");
 
     msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);
@@ -184,8 +194,8 @@ struct resource myco_agent_request_resource(int agent_message_queue_id, char *re
     myresource.pointer = myco_malloc(msg.resource_size);
     myresource.size = msg.resource_size;
 
-    if (myco_read_transactional(msg.sender_pid, msg.resource_pointer, msg.resource_size, myresource.pointer, myresource.size) == -1) {
-        fprintf(stderr, "FATAL ERROR in myco_read_transactional() %s\n", strerror(errno));
+    if (myco_copy_memory(msg.sender_pid, msg.resource_pointer, msg.resource_size, myresource.pointer, myresource.size) == -1) {
+        fprintf(stderr, "FATAL ERROR in myco_copy_memory() %s\n", strerror(errno));
         myresource.size = -1;
         return myresource;
     }
@@ -208,7 +218,7 @@ int myco_agent_request_resource_list(int agent_message_queue_id) {
     msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);
 
     if(DEBUG) {
-        printf("%s", msg.message);
+        //printf("%s", msg.message);
     }
 
     if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
@@ -217,7 +227,7 @@ int myco_agent_request_resource_list(int agent_message_queue_id) {
 
     resource_pointer = myco_malloc(msg.resource_size);
 
-    if (myco_read_transactional(msg.sender_pid, msg.resource_pointer, msg.resource_size, (void *)resource_pointer, msg.resource_size) == -1) {
+    if (myco_copy_memory(msg.sender_pid, msg.resource_pointer, msg.resource_size, (void *)resource_pointer, msg.resource_size) == -1) {
         printf("ERROR: %s\n", strerror(errno));
         return -1;
     }
@@ -278,8 +288,8 @@ struct resource myco_agent_read_remote_resource(int agent_message_queue_id, char
     myresource.pointer = myco_malloc(msg.resource_size);
     myresource.size = msg.resource_size;
 
-    if (myco_read_transactional(msg.sender_pid, msg.resource_pointer, msg.resource_size, myresource.pointer, myresource.size) == -1) {
-        fprintf(stderr, "FATAL ERROR in myco_read_transactional() %s\n", strerror(errno));
+    if (myco_copy_memory(msg.sender_pid, msg.resource_pointer, msg.resource_size, myresource.pointer, myresource.size) == -1) {
+        fprintf(stderr, "FATAL ERROR in myco_copy_memory() %s\n", strerror(errno));
         myresource.size = -1;
         return myresource;
     }

src/myco-agent.h

@@ -34,6 +34,10 @@
 #include <errno.h>
 #include <string.h>
 #include "myco-ipc.c"
+#include <stdlib.h>
+#include <sys/file.h>
+#include <sys/mman.h>
+#include <sys/wait.h>
 
 /**
  * myco_agent_register:
@@ -60,7 +64,7 @@ int myco_agent_unregister (const char *agent_name, int message_queue_id);
  *
  * Returns: message_queue_id on success, -1 on error 
  */
-int myco_agent_register_resource (int message_queue_id, const char *agent_name, const char *resource_name, int resource_transactional, pid_t pid, int size, void* pointer);
+void *myco_agent_register_resource (int agent_message_queue_id, const char *agent_name, const char *resource_name, int resource_transactional, pid_t pid, int size);
 
 /**
  * myco_agent_unregister_resource:

src/myco-daemon.h

@@ -50,8 +50,8 @@ struct myco_resource_ {
     pid_t pid;
     void* pointer;
     int size;
-    int read_locked;
-    int transfer_locked;
+    int read_locked; // 0 = unlocked, 1 = locked
+    int transfer_locked; // 0 = unlocked, 1 = locked
     myco_resource *next;
     myco_resource *prev;
 };

src/myco-ipc.c

@@ -103,7 +103,7 @@ message myco_send_and_receive(message msg, int send_message_queue_id, int receiv
     return msg;
 }
 
-int myco_read_transactional(pid_t source_pid, void *source_pointer, int source_length, void *target_pointer, int target_length) {
+int myco_copy_memory(pid_t source_pid, void *source_pointer, int source_length, void *target_pointer, int target_length) {
     struct iovec local[1];
     struct iovec remote[1];
     ssize_t nread;

test/mycoagent.c

@@ -19,30 +19,38 @@
 int main() {
     // At the beginning every agent needs to register itself with the agent server.
     char *agent_name = "MYAGENT";
-    int agent_message_queue_id;
-    agent_message_queue_id = myco_agent_register(agent_name);
+    int agent_message_queue_id = myco_agent_register(agent_name);
 
     // MYCO create
     // #MYCO create("RESOURCE_1", 1024, transactional)
+    char *resource_name = "RESOURCE_1";
     int resource_size = 1024;
-    void *resource_pointer;
-    resource_pointer = myco_malloc(resource_size);
-    myco_agent_register_resource(agent_message_queue_id, agent_name, "RESOURCE_1", RESOURCE_TRANSACTIONAL, getpid(), resource_size, resource_pointer);
+    void *resource_pointer = myco_agent_register_resource(agent_message_queue_id, agent_name, resource_name, RESOURCE_TRANSACTIONAL, getpid(), resource_size);
+
+    myco_agent_request_resource_list(agent_message_queue_id);
 
     // MYCO lock
-    myco_agent_lock_resource(agent_message_queue_id, "RESOURCE_1");
+    myco_agent_lock_resource(agent_message_queue_id, agent_name, resource_name);
+
+    myco_agent_request_resource_list(agent_message_queue_id);
 
     // MYCO unlock
-    myco_agent_unlock_resource(agent_message_queue_id, "RESOURCE_1");
+    myco_agent_unlock_resource(agent_message_queue_id, agent_name, resource_name);
+
+    myco_agent_request_resource_list(agent_message_queue_id);
 
     // MYCO release
-    myco_agent_release_resource(agent_message_queue_id, "RESOURCE_1");
+    // unmap memory
+    myco_agent_release_resource(agent_message_queue_id, resource_name, resource_pointer, resource_size);
 
     // View resources
     myco_agent_request_resource_list(agent_message_queue_id);
 
     // MYCO fetch
     // #MYCO fetch("RESOURCE_2")
+    // open shared memory
+    // mmap memory
+    /*
     struct resource myresource2;
     myresource2 = myco_agent_request_resource(agent_message_queue_id, "RESOURCE_2");
     if (myresource2.size != -1) {
@@ -50,9 +58,11 @@ int main() {
     } else {
         fprintf(stderr, "Transfer did not work!\n");
     }
+    */
 
     // MYCO read
     // #MYCO read("RESOURCE_3")
+    /*
     struct resource myresource3;
     myresource3 = myco_agent_read_remote_resource(agent_message_queue_id, "RESOURCE_3");
     if (myresource3.size != -1) {
@@ -60,90 +70,24 @@ int main() {
     } else {
         fprintf(stderr, "Transfer did not work!\n");
     }
+    */
 
     // MYCO push
     // TODO
 
     // MYCO free
     // #MYCO free("RESOURCE_1")
-    myco_agent_unregister_resource(agent_message_queue_id, "RESOURCE_1");
-    myco_free(resource_pointer);
+    myco_agent_unregister_resource(agent_message_queue_id, resource_name);
 
     // MYCO free
     // #MYCO free("RESOURCE_2")
+    /*
     myco_agent_unregister_resource(agent_message_queue_id, "RESOURCE_2");
     myco_free(myresource2.pointer);
+    */
 
-    // At the end of its runtime the agent unregisters with the agent.
+    // At the end of its runtime the agent unregisters with the daemon.
     myco_agent_unregister(agent_name, agent_message_queue_id);
 
     return EXIT_SUCCESS;
 }
-
-/*
-   struct resource myresource;
-   pid_t worker_pid;
-   int agent_message_queue_id, worker_message_queue_id;
-   message msg;
-   char *split;
-   int split_count;
-
-   if (argv[1] == NULL) {
-   fprintf(stderr, "ERROR: executed mycoagent without worker (try ./mycoagent mycoworker)\n");
-   return 1;
-   }
-
-   worker_message_queue_id = myco_create_private_message_queue();
-   worker_pid = myco_worker_start(worker_message_queue_id, argv[1]);
-
-   agent_message_queue_id = myco_agent_register(argv[1]);
-
-//while (myco_worker_is_running(worker_pid) == 0) {
-while (1) {
-msg = myco_receive(worker_message_queue_id);
-if (strcmp(msg.message, "WORKER: ALLOCATED MEMORY") == 0) {
-if (DEBUG) {
-printf("%s\n", msg.message);
-}
-myco_agent_register_resource(agent_message_queue_id, argv[1], msg.resource_name, msg.resource_transactional, worker_pid, msg.resource_size, msg.resource_pointer);
-}
-if (strcmp(msg.message, "WORKER: FREED MEMORY") == 0) {
-if (DEBUG) {
-printf("%s %s\n", msg.message, msg.resource_name);
-}
-myco_agent_unregister_resource(agent_message_queue_id, msg.resource_name);
-}
-if (strcmp(msg.message, "WORKER: REQUESTING RESOURCE") == 0) {
-if (DEBUG) {
-printf("%s\n", msg.message);
-}
-// Stop this worker
-
-// Create shared memory region
-
-// Start transfer
-
-// Continue this worker
-
-//myresource = myco_agent_request_resource(agent_message_queue_id, "RESOURCE_1");
-}
-if (strcmp(msg.message, "WORKER: REQUESTING RESOURCE LIST") == 0) {
-if (DEBUG) {
-printf("%s\n", msg.message);
-}
-myco_agent_request_resource_list(agent_message_queue_id);
-}
-if (strcmp(msg.message, "WORKER: SHUTTING DOWN") == 0) {
-if (DEBUG) {
-printf("%s\n", msg.message);
-}
-break;
-}
-}
-
-myco_agent_unregister(argv[1], agent_message_queue_id);
-
-
-return EXIT_SUCCESS;
-}
-*/

test/mycoagent2.c

@@ -17,6 +17,7 @@
 // MYCO unlock:         Allows read access again.
 
 int main() {
+    /*
     // At the beginning every agent needs to register itself with the agent server.
     char *agent_name = "MYAGENT2";
     int agent_message_queue_id;
@@ -69,6 +70,7 @@ int main() {
 
     // At the end of its runtime the agent unregisters with the agent.
     myco_agent_unregister(agent_name, agent_message_queue_id);
+    */
 
     return EXIT_SUCCESS;
 }