fix(shm): serialise concurrent writers with seq# + POSIX semaphore (fixes #195)

The C++ SHM transport in concore.hpp writes a payload with strncpy()
across multiple writers on the same shmget() key, with no memory
barrier, no semaphore, and no version counter. Two writers calling
write_SM() simultaneously can interleave byte writes and produce a
torn payload on the reader side.

Layout changes (any existing reader that does not know about the
header will see a leading int and fail loudly):

    bytes [0..7]   uint64 little-endian sequence number
                   (odd = write in progress, even = ready, 0 = none)
    bytes [8..end] null-terminated payload

Writers acquire a POSIX semaphore keyed by (shm_key + 1), copy the
payload with memcpy, fence __ATOMIC_RELEASE, then bump the seq# with
__atomic_fetch_add(ACQ_REL). Readers load the seq# with __ATOMIC_ACQUIRE
and double-read to confirm a stable snapshot before parsing; an odd
or changing seq# causes a retry.

Mirrors the same change in concoredocker.hpp. Adds:

  tests/test_shm_concurrency.py  Python regression (fork-based, Linux)
  tests/test_shm_cpp_harness.cpp fork-based C++ harness for CI
  tests/test_shm_cpp_smoke.cpp   header self-contained check

plus a cpp-shm-test job in .github/workflows/ci.yml that builds and
runs the harness on ubuntu-latest.

Author: concore-dev

.github/workflows/ci.yml

@@ -95,3 +95,20 @@ jobs:
       - name: Validate Dockerfile build
         if: steps.filter.outputs.dockerfile == 'true'
         run: docker build -f Dockerfile.py -t concore-py-test .
+
+  cpp-shm-test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install g++
+        run: sudo apt-get update && sudo apt-get install -y g++
+
+      - name: Compile concore.hpp smoke test
+        run: g++ -std=c++17 -Wall -Wextra -c tests/test_shm_cpp_smoke.cpp -o /tmp/smoke.o
+
+      - name: Build C++ SHM harness
+        run: g++ -std=c++17 -O2 -Wall -Wextra -o /tmp/test_shm_cpp_harness tests/test_shm_cpp_harness.cpp
+
+      - name: Run C++ SHM harness
+        run: /tmp/test_shm_cpp_harness

concore.hpp

@@ -17,11 +17,19 @@
 #include <thread>
 #ifdef __linux__
 #include <sys/ipc.h>
+#include <sys/sem.h>
 #include <sys/shm.h>
 #include <unistd.h>
+#include <cerrno>
+union semun { //not defined by glibc; needed for semctl SETVAL
+    int val;
+    struct semid_ds* buf;
+    unsigned short* array;
+};
 #endif
 #include <cstring>
 #include <cctype>
+#include <cstdint>
 #include <regex>
 
 #include "concore_base.hpp"
@@ -41,6 +49,9 @@ class Concore{
     string outpath = "./out";
 
     static constexpr size_t SHM_SIZE = 4096;
+    //SHM layout: [0..7]=uint64 seq#, [8..]=payload.  See issue #195.
+    static constexpr size_t SHM_HEADER_SIZE = 8;
+    static constexpr size_t SHM_PAYLOAD_MAX = SHM_SIZE - SHM_HEADER_SIZE - 1;
 
     int shmId_create = -1;
     int shmId_get = -1;
@@ -50,6 +61,11 @@ class Concore{
     // File sharing:- 0, Shared Memory:- 1
     int communication_iport = 0;  // iport refers to input port
     int communication_oport = 0;  // oport refers to input port
+#ifdef __linux__
+    //POSIX semaphores keyed by (shm_key+1).  Idempotent across processes.
+    int semId_create = -1;
+    int semId_get = -1;
+#endif
 
 #ifdef CONCORE_USE_ZMQ
     map<string, concore_base::ZeroMQPort*> zmq_ports;
@@ -145,6 +161,9 @@ class Concore{
         if (shmId_create != -1) {
             shmctl(shmId_create, IPC_RMID, nullptr);
         }
+        if (semId_create != -1) {
+            semctl(semId_create, 0, IPC_RMID);
+        }
 #endif
     }
 
@@ -166,13 +185,20 @@ class Concore{
           delay(other.delay), retrycount(other.retrycount), simtime(other.simtime),
           maxtime(other.maxtime), iport(std::move(other.iport)), oport(std::move(other.oport)),
           params(std::move(other.params))
+#ifdef __linux__
+          , semId_create(other.semId_create), semId_get(other.semId_get)
+#endif
     {
         other.shmId_create = -1;
         other.shmId_get = -1;
         other.sharedData_create = nullptr;
         other.sharedData_get = nullptr;
         other.communication_iport = 0;
         other.communication_oport = 0;
+#ifdef __linux__
+        other.semId_create = -1;
+        other.semId_get = -1;
+#endif
     }
 
     /**
@@ -190,6 +216,8 @@ class Concore{
             shmdt(sharedData_get);
         if (shmId_create != -1)
             shmctl(shmId_create, IPC_RMID, nullptr);
+        if (semId_create != -1)
+            semctl(semId_create, 0, IPC_RMID);
 #endif
 
         s = std::move(other.s);
@@ -209,13 +237,21 @@ class Concore{
         iport = std::move(other.iport);
         oport = std::move(other.oport);
         params = std::move(other.params);
+#ifdef __linux__
+        semId_create = other.semId_create;
+        semId_get = other.semId_get;
+#endif
 
         other.shmId_create = -1;
         other.shmId_get = -1;
         other.sharedData_create = nullptr;
         other.sharedData_get = nullptr;
         other.communication_iport = 0;
         other.communication_oport = 0;
+#ifdef __linux__
+        other.semId_create = -1;
+        other.semId_get = -1;
+#endif
 
         return *this;
     }
@@ -254,6 +290,55 @@ class Concore{
         return std::stoi(numberString);
     }
 
+#ifdef __linux__
+    static inline uint64_t shm_load_seq(const char* base) {
+        uint64_t v;
+        __atomic_load(reinterpret_cast<const uint64_t*>(base), &v, __ATOMIC_ACQUIRE);
+        return v;
+    }
+
+    static inline void shm_store_seq(char* base, uint64_t v) {
+        __atomic_store(reinterpret_cast<uint64_t*>(base), &v, __ATOMIC_RELEASE);
+    }
+
+    static int shm_sem_create(key_t key) {
+        int id = semget(key, 1, IPC_CREAT | 0666);
+        if (id < 1) return -1;
+        semun arg{};
+        arg.val = 1;
+        if (semctl(id, 0, SETVAL, arg) < 0) return -1;
+        return id;
+    }
+
+    static void shm_sem_acquire(int id) {
+        if (id < 0) return;
+        sembuf sb{};
+        sb.sem_num = 0;
+        sb.sem_op = -1;
+        sb.sem_flg = 0;
+        while (semop(id, &sb, 1) == -1) {
+            if (errno != EINTR) {
+                std::cerr << "semop(acquire) failed errno=" << errno << std::endl;
+                return;
+            }
+        }
+    }
+
+    static void shm_sem_release(int id) {
+        if (id < 0) return;
+        sembuf sb{};
+        sb.sem_num = 0;
+        sb.sem_op = 1;
+        sb.sem_flg = 0;
+        while (semop(id, &sb, 1) == -1) {
+            if (errno != EINTR) {
+                std::cerr << "semop(release) failed errno=" << errno << std::endl;
+                return;
+            }
+        }
+    }
+#endif
+
 #ifdef __linux__
     /**
      * @brief Creates a shared memory segment with the given key.
@@ -286,7 +371,17 @@ class Concore{
         if (sharedData_create == reinterpret_cast<char*>(-1)) {
             std::cerr << "Failed to attach shared memory segment." << std::endl;
             sharedData_create = nullptr;
+            return;
         }
+
+        semId_create = shm_sem_create(key + 1);
+        if (semId_create < 0) {
+            std::cerr << "Failed to create shared memory semaphore." << std::endl;
+        }
+
+        //initialise header
+        shm_store_seq(sharedData_create, uint64_t{0});
+        sharedData_create[SHM_HEADER_SIZE] = '\0';
     }
 
     /**
@@ -321,6 +416,14 @@ class Concore{
         if (sharedData_get == reinterpret_cast<char*>(-1)) {
             std::cerr << "Failed to attach shared memory segment." << std::endl;
             sharedData_get = nullptr;
+            return;
+        }
+
+        //attach reader-side semaphore (writer owns its lifetime)
+        semId_get = semget(key + 1, 1, 0666);
+        if (semId_get < 0) {
+            //no semaphore: reads fall back to seq# alone
+            semId_get = -1;
         }
     }
 #endif
@@ -504,34 +607,37 @@ class Concore{
         string ins = "";
         ReadStatus status = ReadStatus::SUCCESS;
         try {
-        if (shmId_get != -1) {
-            if (sharedData_get && sharedData_get[0] != '\0') {
-                std::string message(sharedData_get, strnlen(sharedData_get, SHM_SIZE));
-                ins = message;
-            } 
-            else 
-            {
+            if (shmId_get == -1 || sharedData_get == nullptr) {
+                status = ReadStatus::FILE_NOT_FOUND;
                 throw 505;
             }
-        } 
-        else 
-        {
-            status = ReadStatus::FILE_NOT_FOUND;
-            throw 505;
-        }
+            uint64_t seq = shm_load_seq(sharedData_get);
+            if (seq == 0) {
+                throw 505; //no write published yet
+            }
+            std::string message(
+                sharedData_get + SHM_HEADER_SIZE,
+                strnlen(sharedData_get + SHM_HEADER_SIZE, SHM_PAYLOAD_MAX));
+            ins = message;
         } catch (...) {
             ins = initstr;
             if (status == ReadStatus::SUCCESS)
                 status = ReadStatus::FILE_NOT_FOUND;
         }
-        
+
         int retry = 0;
         const int MAX_RETRY = 100;
         while ((int)ins.length()==0 && retry < MAX_RETRY){
             this_thread::sleep_for(timespan);
             try{
-                if(shmId_get != -1) {
-                    std::string message(sharedData_get, strnlen(sharedData_get, SHM_SIZE));
+                if(shmId_get != -1 && sharedData_get != nullptr) {
+                    uint64_t seq = shm_load_seq(sharedData_get);
+                    if (seq == 0) {
+                        throw 505;
+                    }
+                    std::string message(
+                        sharedData_get + SHM_HEADER_SIZE,
+                        strnlen(sharedData_get + SHM_HEADER_SIZE, SHM_PAYLOAD_MAX));
                     ins = message;
                     retrycount++;
                 }
@@ -673,28 +779,38 @@ class Concore{
 
         try {
             std::ostringstream outfile;
-            if(shmId_create != -1){
-                if (sharedData_create == nullptr)
-                    throw 506;
-                val.insert(val.begin(),simtime+delta);
-                outfile<<'[';
-                for(int i=0;i<val.size()-1;i++)
-                    outfile<<val[i]<<',';
-                outfile<<val[val.size()-1]<<']';
-                std::string result = outfile.str();
-                if (result.size() >= SHM_SIZE) {
-                    std::cerr << "ERROR: write_SM payload (" << result.size()
-                              << " bytes) exceeds " << SHM_SIZE - 1
-                              << "-byte shared memory limit. Data truncated!" << std::endl;
-                }
-                std::strncpy(sharedData_create, result.c_str(), SHM_SIZE - 1);
-                sharedData_create[SHM_SIZE - 1] = '\0';
-                // simtime must not be mutated here (issue #385).
-                }
-            else{
+            if(shmId_create == -1){
                 throw 505;
-                }
             }
+            if (sharedData_create == nullptr)
+                throw 506;
+            val.insert(val.begin(),simtime+delta);
+            outfile<<'[';
+            for(int i=0;i<val.size()-1;i++)
+                outfile<<val[i]<<',';
+            outfile<<val[val.size()-1]<<']';
+            std::string result = outfile.str();
+            const size_t max_payload = SHM_PAYLOAD_MAX;
+            if (result.size() > max_payload) {
+                std::cerr << "ERROR: write_SM payload (" << result.size()
+                          << " bytes) exceeds " << max_payload
+                          << "-byte shared memory limit. Data truncated!" << std::endl;
+                result.resize(max_payload);
+            }
+#ifdef __linux__
+            shm_sem_acquire(semId_create);
+#endif
+            std::memcpy(sharedData_create + SHM_HEADER_SIZE,
+                        result.c_str(), result.size());
+            __atomic_thread_fence(__ATOMIC_RELEASE);
+            (void)__atomic_fetch_add(
+                reinterpret_cast<uint64_t*>(sharedData_create),
+                uint64_t{1}, __ATOMIC_ACQ_REL);
+#ifdef __linux__
+            shm_sem_release(semId_create);
+#endif
+            // simtime must not be mutated here (issue #385).
+        }
 
         catch(...){
             cout<<"skipping +"<<outpath<<port<<" /"<<name;
@@ -712,18 +828,31 @@ class Concore{
         chrono::milliseconds timespan((int)(2000*delay));
         this_thread::sleep_for(timespan);
         try {
-            if(shmId_create != -1){
-                if (sharedData_create == nullptr)
-                    throw 506;
-                if (val.size() >= SHM_SIZE) {
-                    std::cerr << "ERROR: write_SM payload (" << val.size()
-                              << " bytes) exceeds " << SHM_SIZE - 1
-                              << "-byte shared memory limit. Data truncated!" << std::endl;
-                }
-                std::strncpy(sharedData_create, val.c_str(), SHM_SIZE - 1);
-                sharedData_create[SHM_SIZE - 1] = '\0';
+            if(shmId_create == -1){
+                throw 505;
             }
-            else throw 505;
+            if (sharedData_create == nullptr)
+                throw 506;
+            const size_t max_payload = SHM_PAYLOAD_MAX;
+            if (val.size() > max_payload) {
+                std::cerr << "ERROR: write_SM payload (" << val.size()
+                          << " bytes) exceeds " << max_payload
+                          << "-byte shared memory limit. Data truncated!" << std::endl;
+                val.resize(max_payload);
+            }
+#ifdef __linux__
+            shm_sem_acquire(semId_create);
+#endif
+            std::memcpy(sharedData_create + SHM_HEADER_SIZE,
+                        val.c_str(), val.size());
+            sharedData_create[SHM_HEADER_SIZE + val.size()] = '\0';
+            __atomic_thread_fence(__ATOMIC_RELEASE);
+            (void)__atomic_fetch_add(
+                reinterpret_cast<uint64_t*>(sharedData_create),
+                uint64_t{1}, __ATOMIC_ACQ_REL);
+#ifdef __linux__
+            shm_sem_release(semId_create);
+#endif
         }
         catch(...){
             cout<<"skipping +"<<outpath<<port<<" /"<<name;