Add semaphore

clin99 · clin99 · commit 4dae4ccbf8db · 2019-05-02T14:06:45.000-05:00
diff --git a/taskflow/executor/semaphore.hpp b/taskflow/executor/semaphore.hpp
@@ -0,0 +1,113 @@
+//  - modified the thread monitor from TBB
+
+#pragma once
+
+#include "../predef/compiler.hpp"
+#include "../predef/os.hpp"
+
+#if TF_COMP_MSVC || TF_COMP_MSVC_EMULATED 
+  #include <windows.h>
+#elif TF_COMP_GNUC || TF_COMP_GNUC_EMULATED
+  #include <semaphore.h> 
+#elif TF_OS_MACOS
+  #include <mach/semaphore.h>
+  #include <mach/task.h>
+  #include <mach/mach_init.h>
+  #include <mach/error.h>
+#else
+#  error "Unsupported compiler"
+#endif
+
+
+#include <cassert>
+
+namespace tf {
+
+#if TF_COMP_MSVC || TF_COMP_MSVC_EMULATED 
+struct BinarySemaphore {
+  BinarySemaphore() { 
+    sem = CreateSemaphore(NULL, 0, 1, NULL);
+  }
+
+  ~BinarySemaphore() { 
+    CloseHandle(sem); 
+  }
+
+  void P() { 
+    WaitForSingleObject(sem, INFINITE); 
+  }
+
+  void V() { 
+    ReleaseSemaphore(sem, 1, NULL);
+  }
+
+private:  
+  HANDLE sem;
+  BinarySemaphore(const BinarySemaphore& other) = delete;
+  BinarySemaphore& operator=(const BinarySemaphore& other) = delete;
+};
+
+#elif TF_COMP_GNUC || TF_COMP_GNUC_EMULATED 
+struct BinarySemaphore {
+  BinarySemaphore() {
+    int ret = sem_init( &sem, 0, 0 );
+    // Return 0 on success; -1 on error
+    assert(ret == 0);
+  }
+
+  ~BinarySemaphore() {
+    int ret = sem_destroy( &sem );
+    // Return 0 on success; -1 on error
+    assert(ret == 0);
+  }
+
+  void P() {
+    while( sem_wait( &sem )!=0 );
+  }
+
+  void V() { 
+    sem_post( &sem ); 
+  }
+ 
+private:  
+  sem_t sem;
+  BinarySemaphore(const BinarySemaphore& other) = delete;
+  BinarySemaphore& operator=(const BinarySemaphore& other) = delete;
+};
+
+#elif TF_OS_MACOS
+struct BinarySemaphore {
+
+  BinarySemaphore() : sem(0) {
+    kern_return_t ret = semaphore_create(mach_task_self(), &sem, SYNC_POLICY_FIFO, 0);
+  }
+
+  ~BinarySemaphore() {
+    kern_return_t ret = semaphore_destroy(mach_task_self(), sem);
+  }
+
+  void P() {
+    int ret;
+    do {
+      ret = semaphore_wait( sem );
+    } while( ret==KERN_ABORTED );
+  }
+
+  void V() { 
+    semaphore_signal( sem );  
+  }
+
+private:  
+  semaphore_t sem;
+  BinarySemaphore(const BinarySemaphore& other) = delete;
+  BinarySemaphore& operator=(const BinarySemaphore& other) = delete;
+};
+
+#else
+#  error "Unsupported compiler"
+#endif
+
+
+}  // end of namespace tf. ---------------------------------------------------
+
+
diff --git a/unittest/executor.cpp b/unittest/executor.cpp
@@ -45,6 +45,9 @@
 #include <doctest.h>
 #include <taskflow/executor/executor.hpp>
 
+#include <chrono>
+#include <taskflow/executor/semaphore.hpp>
+
 // ============================================================================
 // WorkStealingQueue tests
 // ============================================================================
@@ -447,6 +450,8 @@ void test_executor() {
   }
 }
 
+
+
 // ----------------------------------------------------------------------------
 // Testcase: SimpleExecutor
 // ----------------------------------------------------------------------------
@@ -482,3 +487,46 @@ TEST_CASE("EigenWorkStealingExecutor" * doctest::timeout(300)) {
   test_executor<tf::EigenWorkStealingExecutor<std::function<void()>>>();
 }
 
+// ----------------------------------------------------------------------------
+// Testcase: BinarySemaphore
+// ----------------------------------------------------------------------------
+TEST_CASE("BinarySemaphore" * doctest::timeout(300)) {
+  tf::BinarySemaphore sema;
+
+  size_t count {0};
+
+  std::thread t1([&](){
+    using namespace std::chrono_literals;
+    for(int i=0; i<100; i++) {
+      std::this_thread::sleep_for(2ms);
+      sema.P();
+      ++ count;
+    }
+  });
+
+  {
+    using namespace std::chrono_literals;
+    for(int i=0; i<100; i++) {
+      sema.V();
+      std::this_thread::sleep_for(10ms);
+      REQUIRE(count == i+1);
+    }
+  }
+
+  t1.join();
+
+
+  count = 0;
+  std::thread t2([&](){
+    using namespace std::chrono_literals;
+    for(int i=0; i<10; i++) {
+      std::this_thread::sleep_for(10ms);
+      sema.V();
+      count += 1;
+    }
+  });
+
+  t2.join();
+  REQUIRE(count == 10);
+}
+
