Use futexes for linux bridges, closes #234

(needs a full rebuild of carla and bridges)

source/backend/engine/CarlaEngineBridge.cpp

@@ -175,7 +175,7 @@ struct BridgeRtClientControl : public CarlaRingBufferControl<SmallStackBuffer> {
 
         WaitHelper(BridgeRtClientControl& c) noexcept
             : data(c.data),
-              ok(jackbridge_sem_timedwait(&data->sem.server, 5)) {}
+              ok(jackbridge_sem_timedwait(&data->sem.server, 5000)) {}
 
         ~WaitHelper() noexcept
         {
@@ -494,7 +494,7 @@ public:
         carla_stdout("  sizeof(BridgeNonRtClientData): %i/" P_SIZE, shmNonRtClientDataSize, sizeof(BridgeNonRtClientData));
         carla_stdout("  sizeof(BridgeNonRtServerData): %i/" P_SIZE, shmNonRtServerDataSize, sizeof(BridgeNonRtServerData));
 
-        if (shmRtClientDataSize != sizeof(BridgeRtClientData) || shmNonRtClientDataSize != sizeof(BridgeNonRtClientData)  || shmNonRtServerDataSize != sizeof(BridgeNonRtServerData))
+        if (shmRtClientDataSize != sizeof(BridgeRtClientData) || shmNonRtClientDataSize != sizeof(BridgeNonRtClientData) || shmNonRtServerDataSize != sizeof(BridgeNonRtServerData))
             return false;
 
         // tell backend we're live

source/backend/plugin/CarlaPluginBridge.cpp

@@ -240,13 +240,13 @@ struct BridgeRtClientControl : public CarlaRingBufferControl<SmallStackBuffer> {
         setRingBuffer(nullptr, false);
     }
 
-    bool waitForClient(const uint secs) noexcept
+    bool waitForClient(const uint msecs) noexcept
     {
         CARLA_SAFE_ASSERT_RETURN(data != nullptr, false);
 
         jackbridge_sem_post(&data->sem.server);
 
-        return jackbridge_sem_timedwait(&data->sem.client, secs);
+        return jackbridge_sem_timedwait(&data->sem.client, msecs);
     }
 
     void writeOpcode(const PluginBridgeRtClientOpcode opcode) noexcept
@@ -738,6 +738,7 @@ public:
           fSaved(true),
           fTimedOut(false),
           fTimedError(false),
+          fProcWaitTime(0),
           fLastPongTime(-1),
           fBridgeBinary(),
           fBridgeThread(engine, this),
@@ -783,7 +784,7 @@ public:
             fShmRtClientControl.commitWrite();
 
             if (! fTimedOut)
-                waitForClient("stopping", 3);
+                waitForClient("stopping", 3000);
         }
 
         fBridgeThread.stopThread(3000);
@@ -1379,7 +1380,7 @@ public:
         fTimedOut = false;
 
         try {
-            waitForClient("activate", 2);
+            waitForClient("activate", 2000);
         } CARLA_SAFE_EXCEPTION("activate - waitForClient");
     }
 
@@ -1397,7 +1398,7 @@ public:
         fTimedOut = false;
 
         try {
-            waitForClient("deactivate", 2);
+            waitForClient("deactivate", 2000);
         } CARLA_SAFE_EXCEPTION("deactivate - waitForClient");
     }
 
@@ -1780,7 +1781,7 @@ public:
             fShmRtClientControl.commitWrite();
         }
 
-        waitForClient("process", 1);
+        waitForClient("process", fProcWaitTime);
 
         if (fTimedOut)
         {
@@ -1875,7 +1876,10 @@ public:
             fShmNonRtClientControl.commitWrite();
         }
 
-        waitForClient("buffersize", 1);
+        //fProcWaitTime = newBufferSize*1000/pData->engine->getSampleRate();
+        fProcWaitTime = 1000;
+
+        waitForClient("buffersize", 1000);
     }
 
     void sampleRateChanged(const double newSampleRate) override
@@ -1887,7 +1891,10 @@ public:
             fShmNonRtClientControl.commitWrite();
         }
 
-        waitForClient("samplerate", 1);
+        //fProcWaitTime = pData->engine->getBufferSize()*1000/newSampleRate;
+        fProcWaitTime = 1000;
+
+        waitForClient("samplerate", 1000);
     }
 
     void offlineModeChanged(const bool isOffline) override
@@ -1898,7 +1905,7 @@ public:
             fShmNonRtClientControl.commitWrite();
         }
 
-        waitForClient("offline", 1);
+        waitForClient("offline", 1000);
     }
 
     // -------------------------------------------------------------------
@@ -2509,6 +2516,10 @@ public:
 
         fShmNonRtClientControl.commitWrite();
 
+        // testing dummy message
+        fShmRtClientControl.writeOpcode(kPluginBridgeRtClientNull);
+        fShmRtClientControl.commitWrite();
+
         // init bridge thread
         {
             char shmIdsStr[6*4+1];
@@ -2599,6 +2610,7 @@ private:
     bool fSaved;
     bool fTimedOut;
     bool fTimedError;
+    uint fProcWaitTime;
 
     int64_t fLastPongTime;
 
@@ -2650,15 +2662,15 @@ private:
         fShmRtClientControl.writeULong(static_cast<uint64_t>(fShmAudioPool.size));
         fShmRtClientControl.commitWrite();
 
-        waitForClient("resize-pool", 5);
+        waitForClient("resize-pool", 5000);
     }
 
-    void waitForClient(const char* const action, const uint secs)
+    void waitForClient(const char* const action, const uint msecs)
     {
         CARLA_SAFE_ASSERT_RETURN(! fTimedOut,);
         CARLA_SAFE_ASSERT_RETURN(! fTimedError,);
 
-        if (fShmRtClientControl.waitForClient(secs))
+        if (fShmRtClientControl.waitForClient(msecs))
             return;
 
         fTimedOut = true;

source/jackbridge/JackBridge.hpp

@@ -397,7 +397,7 @@ JACKBRIDGE_API bool jackbridge_set_property_change_callback(jack_client_t* clien
 JACKBRIDGE_API bool jackbridge_sem_init(void* sem) noexcept;
 JACKBRIDGE_API void jackbridge_sem_destroy(void* sem) noexcept;
 JACKBRIDGE_API void jackbridge_sem_post(void* sem) noexcept;
-JACKBRIDGE_API bool jackbridge_sem_timedwait(void* sem, uint secs) noexcept;
+JACKBRIDGE_API bool jackbridge_sem_timedwait(void* sem, uint msecs) noexcept;
 
 JACKBRIDGE_API bool  jackbridge_shm_is_valid(const void* shm) noexcept;
 JACKBRIDGE_API void  jackbridge_shm_init(void* shm) noexcept;

source/jackbridge/JackBridge2.cpp

@@ -52,14 +52,14 @@ void jackbridge_sem_post(void* sem) noexcept
 #endif
 }
 
-bool jackbridge_sem_timedwait(void* sem, uint secs) noexcept
+bool jackbridge_sem_timedwait(void* sem, uint msecs) noexcept
 {
     CARLA_SAFE_ASSERT_RETURN(sem != nullptr, false);
 
 #ifdef JACKBRIDGE_DUMMY
     return false;
 #else
-    return carla_sem_timedwait(*(carla_sem_t*)sem, secs);
+    return carla_sem_timedwait(*(carla_sem_t*)sem, msecs);
 #endif
 }
 

source/jackbridge/JackBridgeExport.cpp

@@ -548,9 +548,9 @@ void jackbridge_sem_post(void* sem) noexcept
     getBridgeInstance().sem_post_ptr(sem);
 }
 
-bool jackbridge_sem_timedwait(void* sem, uint secs) noexcept
+bool jackbridge_sem_timedwait(void* sem, uint msecs) noexcept
 {
-    return getBridgeInstance().sem_timedwait_ptr(sem, secs);
+    return getBridgeInstance().sem_timedwait_ptr(sem, msecs);
 }
 
 bool jackbridge_shm_is_valid(const void* shm) noexcept

source/utils/CarlaSemUtils.hpp

@@ -22,15 +22,24 @@
 
 #include <ctime>
 
+#define CARLA_USE_FUTEXES
+
 #ifdef CARLA_OS_WIN
 struct carla_sem_t { HANDLE handle; };
 #elif defined(CARLA_OS_MAC)
 // TODO
 struct carla_sem_t { char dummy; };
+#elif defined(CARLA_USE_FUTEXES)
+# include <cerrno>
+# include <syscall.h>
+# include <sys/time.h>
+# include <linux/futex.h>
+struct carla_sem_t { int count; };
 #else
+# include <cerrno>
+# include <semaphore.h>
 # include <sys/time.h>
 # include <sys/types.h>
-# include <semaphore.h>
 struct carla_sem_t { sem_t sem; };
 #endif
 
@@ -40,6 +49,7 @@ struct carla_sem_t { sem_t sem; };
 static inline
 bool carla_sem_create2(carla_sem_t& sem) noexcept
 {
+    carla_zeroStruct(sem);
 #if defined(CARLA_OS_WIN)
     SECURITY_ATTRIBUTES sa;
     carla_zeroStruct(sa);
@@ -51,6 +61,9 @@ bool carla_sem_create2(carla_sem_t& sem) noexcept
     return (sem.handle != INVALID_HANDLE_VALUE);
 #elif defined(CARLA_OS_MAC)
     return false; // TODO
+#elif defined(CARLA_USE_FUTEXES)
+    sem.count = 0;
+    return true;
 #else
     return (::sem_init(&sem.sem, 1, 0) == 0);
 #endif
@@ -83,6 +96,9 @@ void carla_sem_destroy2(carla_sem_t& sem) noexcept
     ::CloseHandle(sem.handle);
 #elif defined(CARLA_OS_MAC)
     // TODO
+#elif defined(CARLA_USE_FUTEXES)
+    // nothing to do
+    (void)sem;
 #else
     ::sem_destroy(&sem.sem);
 #endif
@@ -110,6 +126,10 @@ void carla_sem_post(carla_sem_t& sem) noexcept
     ::ReleaseSemaphore(sem.handle, 1, nullptr);
 #elif defined(CARLA_OS_MAC)
     // TODO
+#elif defined(CARLA_USE_FUTEXES)
+    const bool unlocked = __sync_bool_compare_and_swap(&sem.count, 0, 1);
+    CARLA_SAFE_ASSERT_RETURN(unlocked,);
+    ::syscall(__NR_futex, &sem.count, FUTEX_WAKE, 1, nullptr, nullptr, 0);
 #else
     ::sem_post(&sem.sem);
 #endif
@@ -119,21 +139,42 @@ void carla_sem_post(carla_sem_t& sem) noexcept
  * Wait for a semaphore (lock).
  */
 static inline
-bool carla_sem_timedwait(carla_sem_t& sem, const uint secs) noexcept
+bool carla_sem_timedwait(carla_sem_t& sem, const uint msecs) noexcept
 {
-    CARLA_SAFE_ASSERT_RETURN(secs > 0, false);
+    CARLA_SAFE_ASSERT_RETURN(msecs > 0, false);
 
 #if defined(CARLA_OS_WIN)
     return (::WaitForSingleObject(sem.handle, secs*1000) == WAIT_OBJECT_0);
 #elif defined(CARLA_OS_MAC)
-    // TODO
+    return false; // TODO
+#elif defined(CARLA_USE_FUTEXES)
+    timespec timeout;
+    timeout.tv_sec  = static_cast<time_t>(msecs / 1000);
+    timeout.tv_nsec = static_cast<time_t>(msecs % 1000);
+
+    for (; ! __sync_bool_compare_and_swap(&sem.count, 1, 0);)
+    {
+        if (syscall(__NR_futex, &sem.count, FUTEX_WAIT, 0, &timeout, nullptr, 0) == 0)
+        {
+            __sync_fetch_and_sub(&sem.count, 1);
+            return true;
+        }
+
+        //if (errno == EAGAIN)
+        //   continue;
+
+        return false;
+    }
+
+    return true;
 #else
     if (::sem_trywait(&sem.sem) == 0)
         return true;
 
     timespec timeout;
     ::clock_gettime(CLOCK_REALTIME, &timeout);
-    timeout.tv_sec += static_cast<time_t>(secs);
+    timeout.tv_sec  += static_cast<time_t>(msecs / 1000);
+    timeout.tv_nsec += static_cast<time_t>(msecs % 1000);
 
     try {
         return (::sem_timedwait(&sem.sem, &timeout) == 0);