added shared memory with support for a single client

README.md

@@ -21,3 +21,5 @@ $ cd client && make run
 type `.help` for an overview of commands.
 
 That's it!
+
+## 

client/include/client.h

@@ -9,12 +9,12 @@ typedef enum BackendType {
     MQ=0,
     UDS=1,
     TCP=2,
+    SHM=3,
 } backend_type_t;
 
 typedef struct Client {
-    backend_t backends[3];
+    backend_t backends[4];
     backend_type_t selected_backend;
-
 } client_t;
 
 

client/include/shm.h

@@ -0,0 +1,3 @@
+#pragma once
+
+int shm_request(char* in, char* out);

client/src/client.c

@@ -4,6 +4,7 @@
 #include "mq.h"
 #include "tcp.h"
 #include "uds.h"
+#include "shm.h"
 #include <readline/history.h>
 #include <readline/readline.h>
 #include <stdio.h>
@@ -16,6 +17,7 @@ client_t client_create(void) {
             mq_request,
             uds_request,
             tcp_request,
+            shm_request,
         },
         .selected_backend=TCP,
     };
@@ -71,7 +73,10 @@ int client_parse_and_execute(client_t* client) {
         client_select_backend(client, MQ);
         return 0;
     }
-
+    if (!strcmp(line, PROMPT_C "ishm")) {
+        client_select_backend(client, SHM);
+        return 0;
+    }
     if (!strcmp(line, PROMPT_C "clear") || !strcmp(line, PROMPT_C "c")) {
         system("clear");
         return 0;
@@ -109,6 +114,10 @@ char* get_backend_name(backend_type_t type) {
             return "MQ";
         case TCP:
             return "TCP";
-        default: return NULL;
+        case SHM:
+            return "SHM";
+        default: 
+            log_warn_m("Backend name string not defined!");
+            return NULL;
     }
 }

client/src/misc.c

@@ -33,6 +33,7 @@ const char* get_op_help_str() {
         PROMPT_C "itcp  |        :: set IPC backend to " C_YELLOW "'TCP'" C_GREEN "\n"
         PROMPT_C "iuds  |        :: set IPC backend to " C_YELLOW "'UDS'" C_GREEN "\n"
         PROMPT_C "imq   |        :: set IPC backend to " C_YELLOW "'MQ'" C_GREEN "\n"
+        PROMPT_C "ishm  |        :: set IPC backend to " C_YELLOW "'SHM'" C_GREEN "\n"
         PROMPT_C "clear | " PROMPT_C "c     :: clear screen" "\n"
         PROMPT_C "exit  | " PROMPT_C "e     :: close interactive shell" "\n" "\n"
         "|>Work in progress:"

client/src/shm.c

@@ -0,0 +1,92 @@
+#include "shm.h"
+#include "protocol.h"
+#include "log.h"
+
+#include <errno.h>
+#include <stdbool.h>
+#include <unistd.h>
+#include <string.h>
+#include <sys/mman.h>
+#include <sys/stat.h>
+#include <fcntl.h> 
+
+
+#define MAX_RESP_WAIT_TIME_ms (1000*1000) // 1s
+
+void close_and_cleanup(shared_memory_buf_t* buffer, int fd) {
+    atomic_store(&buffer->shared_status, SHM_STAT_DONE);
+    munmap(buffer, sizeof(shared_memory_buf_t));
+    close(fd);
+}
+
+int shm_request(char* in, char* out) {
+    // bind shared memory
+    int fd = shm_open(SHM_PATH, O_RDWR, 0664);
+    int errn = errno;
+    if (fd < 0) {
+        log_err_m("Opening shared memory failed with (err: %s). Is the server running?", strerror(errn));
+        return 1;
+    }
+
+    shared_memory_buf_t *buf = mmap(NULL, sizeof(shared_memory_buf_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
+    if (buf == MAP_FAILED) {
+        log_err_m("Mapping shared memory failed with (err: %s). Is the server running?", strerror(errn));
+        return 1;
+    }
+
+
+    const long retry_delay_ms = 200;
+    bool client_free = false;
+
+    for (int i = 0; i < MAX_RESP_WAIT_TIME_ms; i += retry_delay_ms) {
+        if (buf->shared_status == SHM_STAT_AVAIL) { 
+            // increment status
+            atomic_store(&buf->shared_status, SHM_STAT_NEW_CONNECT);
+            client_free=true;
+            break; 
+        };
+        usleep(retry_delay_ms);
+    }
+    if (!client_free) {
+        log_err_m("Timeout: Server is busy and SHM does not support multiple clients!");
+        return 1;
+    }
+
+    // <== connected  ==> 
+    
+    // now write 
+    buf->message = (mq_message_t){ 
+        .request_id=67,
+        .client_pid = getpid(),
+    };
+
+    memcpy(buf->message.payload, in, READ_BUFFER_SIZE);
+
+    // and set flag
+    atomic_store(&buf->shared_status, SHM_STAT_REQ_WRITTEN);
+
+    // now wait for response
+    bool responded = false;
+
+    for (int i = 0; i < MAX_RESP_WAIT_TIME_ms; i += retry_delay_ms) {
+        if (buf->shared_status == SHM_STAT_RES_WRITTEN) { 
+            // 
+            responded=true;
+            break; 
+        };
+        usleep(retry_delay_ms);
+    }
+    if (!responded) {
+        log_err_m("Timeout: Server didn't respond!");
+        close_and_cleanup(buf, fd);
+        return 1;
+    }
+    
+    // read
+    memcpy(out, buf->response.payload, WRITE_BUFFER_SIZE);
+
+    // done
+    close_and_cleanup(buf, fd);
+
+    return 0;
+}

server/Makefile

@@ -34,7 +34,7 @@ cloc:
 	@cloc --exclude-ext=d,json --fullpath --not-match-d='(.*)deps(.*)' .
 
 monitor: 
-	@watch -t -n 0.1 "ls /dev/mqueue/ /tmp | grep -E 'store|:' && echo tcp: && ss -tlnp | grep store | awk '{printf \"%s --> %s\\n\",$$5, $$6;}'"
+	@watch -t -n 0.1 "ls /dev/mqueue/ /tmp /dev/shm | grep -E 'store|:' && echo tcp: && ss -tlnp | grep store"
 
 -include $(DEP_FILES)
 .PHONY: all clean re run cloc monitor

server/include/client.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "shm_client.h"
 #include "tcp_client.h"
 #include "uds_client.h"
 #include "mq_client.h"
@@ -34,6 +35,7 @@ typedef union ClientData {
     uds_client_data_t uds_data;
     mq_client_data_t mq_data;
     tcp_client_data_t tcp_data;
+    shm_client_data_t shm_data;
 } client_data_t;
 
 typedef struct Client client_t;

server/include/connection_handler.h

@@ -3,6 +3,7 @@
 #include "tcp_conn_handle.h"
 #include "uds_conn_handle.h"
 #include "mq_conn_handle.h"
+#include "shm_conn_handle.h"
 
 typedef struct ConnectionResult {
     enum { CONN_NONE, CONN_ERR, CONN_NEW } status;
@@ -16,6 +17,7 @@ typedef union ConnHandlData {
     uds_handl_data_t uds_data;
     mq_handl_data_t   mq_data;
     tcp_handl_data_t tcp_data;
+    shm_handl_data_t   shm_data;
 } conn_handl_data_t;
 
 typedef struct ConnectionHandler connection_handler_t;

server/include/server.h

@@ -11,7 +11,7 @@
 typedef struct Server {
     volatile int running;
 
-    connection_handler_t conn_handlers[3];
+    connection_handler_t conn_handlers[4];
     size_t num_conn_handlers;
 
     database_t* db;

server/include/shm_client.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include "protocol.h"
+#include <stdlib.h>
+#include <stdint.h>
+
+// forward declare so no circular deps with client.h
+typedef struct Client client_t;
+typedef struct ClientReadResult client_read_result_t;
+typedef struct ClientWriteResult client_write_result_t;
+
+client_t shm_client_create(shared_memory_buf_t* buffer, int fd);
+
+typedef struct SHMClientdata {
+    int fd;
+    shared_memory_buf_t* buffer;
+} shm_client_data_t;
+
+
+// api functs
+client_read_result_t shm_client_read(client_t* client);
+client_write_result_t  shm_client_write(client_t* client, uint8_t* data, size_t data_size);
+void shm_client_close(client_t* client);

server/include/shm_conn_handle.h

@@ -0,0 +1,17 @@
+#pragma once
+
+
+#include "protocol.h"
+typedef struct ConnectionHandler connection_handler_t;
+typedef struct ConnectionResult connection_result_t;
+typedef struct SHMHandldata {
+    int fd;
+    shared_memory_buf_t* buffer;
+} shm_handl_data_t;
+
+connection_handler_t create_shm_handler(void);
+
+int shm_init(connection_handler_t*);
+connection_result_t shm_check_conn(connection_handler_t*);
+void shm_cleanup(connection_handler_t*);
+

server/src/mq_conn_handle.c

@@ -4,8 +4,8 @@
 #include "connection_handler.h"
 #include "mq_conn_handle.h"
 #include "client.h"
+#include "shm_client.h"
 #include "client_thread.h"
-
 #include "log.h"
 
 connection_handler_t create_mq_handler(void) {

server/src/server.c

@@ -5,6 +5,7 @@
 #include "server.h"
 #include "db.h"
 #include "connection_handler.h"
+#include "shm_conn_handle.h"
 #include "uds_conn_handle.h"
 #include "mq_conn_handle.h"
 
@@ -20,6 +21,7 @@ server_t* server_create(void) {
             create_uds_handler(),
             create_mq_handler(),
             create_tcp_handler(),
+            create_shm_handler(),
         },
     };
     srv->num_conn_handlers = sizeof(srv->conn_handlers) / sizeof(connection_handler_t);

server/src/shm_client.c

@@ -0,0 +1,60 @@
+#include "shm_client.h"
+#include "client.h"
+#include "log.h"
+#include "protocol.h"
+#include <stdatomic.h>
+#include <string.h>
+
+
+client_t shm_client_create(shared_memory_buf_t* buffer, int fd) {
+    client_t client = {
+        .client_data.shm_data = (shm_client_data_t) {
+            .fd = fd,
+            .buffer = buffer,
+        },
+        .read  = shm_client_read,
+        .write = shm_client_write,
+        .close = shm_client_close,
+    };
+    log_info_m("SHM client connected");
+
+    return client;
+}
+
+
+client_read_result_t shm_client_read(client_t* client) {
+    shared_memory_buf_t* buffer = client->client_data.shm_data.buffer; 
+
+    while (buffer->shared_status != SHM_STAT_REQ_WRITTEN) { };
+
+    // read 
+
+    client_read_result_t result = {
+        .status = READ_OK,
+        .len = READ_BUFFER_SIZE,
+    };
+    memcpy(result.data, &buffer->message.payload, READ_BUFFER_SIZE);
+
+    return result;
+}
+
+client_write_result_t shm_client_write(client_t* client, uint8_t* data, size_t data_size) {
+    shared_memory_buf_t* buffer = client->client_data.shm_data.buffer; 
+    
+    memcpy(&buffer->response.payload, data, data_size);
+    atomic_store(&buffer->shared_status, SHM_STAT_RES_WRITTEN);
+
+    return (client_write_result_t) {
+        .status = WRITE_OK
+    };
+}
+
+void shm_client_close(client_t* client) {
+    shared_memory_buf_t* buffer = client->client_data.shm_data.buffer; 
+
+    while (buffer->shared_status != SHM_STAT_DONE) { };
+
+    memset(buffer, 0, sizeof(shared_memory_buf_t));
+
+    atomic_store(&buffer->shared_status, SHM_STAT_AVAIL);
+}

server/src/shm_conn_handle.c

@@ -0,0 +1,93 @@
+#include "shm_conn_handle.h"
+#include "client.h"
+#include "shm_client.h"
+#include "client_thread.h"
+#include "connection_handler.h"
+#include "log.h"
+#include "protocol.h"
+
+#include <errno.h>
+#include <stdatomic.h>
+#include <string.h>
+#include <sys/mman.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+
+connection_handler_t create_shm_handler(void) {
+    return (connection_handler_t) {
+        .conn_data.shm_data = (shm_handl_data_t) {
+            .fd = -1,
+        },
+        .init = shm_init,
+        .check_conn = shm_check_conn,
+        .cleanup = shm_cleanup,
+    };
+}
+
+
+int shm_init(connection_handler_t* handle) { 
+
+    // initialize shared memory
+    int fd =  shm_open(SHM_PATH, O_NONBLOCK | O_RDWR | O_CREAT, 0664);
+    int errn = errno;
+    if (fd < 0) {
+        log_err_m("Opening shared memory failed with (err: %s)", strerror(errn));
+        return 1;
+    }
+
+    int err = ftruncate(fd, sizeof(shared_memory_buf_t));
+    errn = errno;
+    if (err < 0) {
+        log_err_m("Truncating shared memory failed with (err: %s)", strerror(errn));
+        return 1;
+    }
+
+    shared_memory_buf_t* buffer = mmap(NULL, sizeof(shared_memory_buf_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
+    errn = errno;
+    if (buffer == MAP_FAILED) {
+        log_err_m("Mapping shared memory failed with (err: %s)", strerror(errn));
+        return 1;
+    }
+
+    handle->conn_data.shm_data.fd = fd;
+    handle->conn_data.shm_data.buffer = buffer;
+
+    memset(buffer, 0, sizeof(shared_memory_buf_t));
+
+    atomic_store(&buffer->shared_status, SHM_STAT_AVAIL);
+
+    log_info_m("Initiaized SHM buffer");
+    return 0;
+}
+
+connection_result_t shm_check_conn(connection_handler_t* handl) {
+    shared_memory_buf_t* shm_buffer = handl->conn_data.shm_data.buffer;
+    
+    shm_status_t status = shm_buffer->shared_status;
+
+    if (status != SHM_STAT_NEW_CONNECT && status != SHM_STAT_REQ_WRITTEN) {
+        return (connection_result_t) {
+            .status = CONN_NONE,
+        };
+    };
+
+    log_ok_m("New client connected via MessageQueue!");
+
+    // create client object and hand to thread 
+    client_t new_client = shm_client_create(shm_buffer, handl->conn_data.shm_data.fd);
+    pthread_t tid = client_thread_start(new_client);
+
+    return (connection_result_t) {
+        .status = CONN_NEW,
+        .tid = tid,
+    };
+
+}
+
+void shm_cleanup(connection_handler_t* handl) {
+    (void) handl;
+    munmap(handl->conn_data.shm_data.buffer, sizeof(shared_memory_buf_t));
+    close(handl->conn_data.shm_data.fd);
+    shm_unlink(SHM_PATH);
+}

shared/include/log.h

@@ -18,6 +18,7 @@ static inline void set_log_level(char level) {
 #define LL_INFO 0b00100000
 #define LL_OK   0b00010000
 #define LL_ALL  0b11110000 
+
 #define LL_NONE 0
 
 #define C_NORMAL  "\033[0m"

shared/include/protocol.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <stdatomic.h>
 #include <stdint.h>
 #include <sys/types.h>
 
@@ -20,6 +21,8 @@
 
 #define TCP_PORT 6767
 
+#define SHM_PATH "/storebus"
+
 // message for shared incoming queue
 typedef struct MQMessage {
     pid_t client_pid;
@@ -32,3 +35,28 @@ typedef struct MQResponse {
     unsigned int request_id;
     uint8_t payload[MQ_PAYLOAD_SIZE_OUT];
 } mq_response_t;
+
+
+// shared memory
+
+/*   
+ * 0 = availible for new connections
+ * 1 = connection reserved
+ * 2 = client wrote request 
+ * 3 = server wrote response
+ * 4 = client finished      
+*/
+typedef enum SHMStatus {
+	SHM_STAT_AVAIL        = 0,
+	SHM_STAT_NEW_CONNECT  = 1,
+	SHM_STAT_REQ_WRITTEN  = 2,
+	SHM_STAT_RES_WRITTEN  = 3,
+	SHM_STAT_DONE         = 4,
+} shm_status_t;
+
+typedef struct SharedMemoryBuf {
+    _Atomic shm_status_t shared_status;
+
+    mq_message_t message;
+    mq_response_t response;
+} shared_memory_buf_t ;