Windows PCQueue support without Boost (#106)

src/translator/pcqueue.h

@@ -10,12 +10,14 @@
 #include <mutex>
 
 #ifdef __APPLE__
-#include <mach/mach.h>
-#include <mach/mach_traps.h>
 #include <mach/semaphore.h>
 #include <mach/task.h>
+#include <mach/mach_traps.h>
+#include <mach/mach.h>
 #elif defined(__linux)
 #include <semaphore.h>
+#elif defined(_WIN32) || defined(_WIN64)
+#include <windows.h>
 #else
 #include <boost/interprocess/sync/interprocess_semaphore.hpp>
 #endif
@@ -37,9 +39,7 @@ namespace bergamot {
 class Semaphore {
   public:
     explicit Semaphore(int value) : task_(mach_task_self()) {
-    ABORT_IF(KERN_SUCCESS !=
-                 semaphore_create(task_, &back_, SYNC_POLICY_FIFO, value),
-             "Could not create semaphore");
+      ABORT_IF(KERN_SUCCESS != semaphore_create(task_, &back_, SYNC_POLICY_FIFO, value), "Could not create semaphore");
     }
 
     ~Semaphore() {
@@ -50,13 +50,11 @@ public:
     }
 
     void wait() {
-    ABORT_IF(KERN_SUCCESS != semaphore_wait(back_),
-             "Wait for semaphore failed");
+      ABORT_IF(KERN_SUCCESS != semaphore_wait(back_), "Wait for semaphore failed");
     }
 
     void post() {
-    ABORT_IF(KERN_SUCCESS != semaphore_signal(back_),
-             "Could not post to semaphore");
+      ABORT_IF(KERN_SUCCESS != semaphore_signal(back_), "Could not post to semaphore");
     }
 
   private:
@@ -64,7 +62,9 @@ private:
     task_t task_;
 };
 
-inline void WaitSemaphore(Semaphore &semaphore) { semaphore.wait(); }
+inline void WaitSemaphore(Semaphore &semaphore) {
+  semaphore.wait();
+}
 
 #elif defined(__linux)
 
@@ -82,7 +82,7 @@ public:
     }
 
     void wait() {
-    while (UTIL_UNLIKELY(-1 == sem_wait(&sem_))) {
+      while (-1 == sem_wait(&sem_)) {
         ABORT_IF(errno != EINTR, "Wait for semaphore failed");
       }
     }
@@ -95,7 +95,49 @@ private:
     sem_t sem_;
 };
 
-inline void WaitSemaphore(Semaphore &semaphore) { semaphore.wait(); }
+inline void WaitSemaphore(Semaphore &semaphore) {
+  semaphore.wait();
+}
+
+#elif defined(_WIN32) || defined(_WIN64)
+
+class Semaphore {
+  public:
+    explicit Semaphore(LONG value) : sem_(CreateSemaphoreA(NULL, value, 2147483647, NULL)) {
+      ABORT_IF(!sem_, "Could not CreateSemaphore {}", GetLastError());
+    }
+
+    ~Semaphore() {
+      CloseHandle(sem_);
+    }
+
+
+    void wait() {
+      while (true) {
+        switch (WaitForSingleObject(sem_, 0L)) {
+          case WAIT_OBJECT_0:
+            return;
+          case WAIT_ABANDONED:
+            ABORT("A semaphore can't be abandoned, confused by Windows");
+          case WAIT_TIMEOUT:
+            continue;
+          case WAIT_FAILED:
+            ABORT("Waiting on Semaphore failed {}", GetLastError());
+        }
+      }
+    }
+
+    void post() {
+      ABORT_IF(!ReleaseSemaphore(sem_, 1, NULL), "Failed to release Semaphore {}", GetLastError());
+    }
+
+  private:
+    HANDLE sem_;
+};
+
+inline void WaitSemaphore(Semaphore &semaphore) {
+  semaphore.wait();
+}
 
 #else
 typedef boost::interprocess::interprocess_semaphore Semaphore;
@@ -113,7 +155,7 @@ inline void WaitSemaphore(Semaphore &on) {
   }
 }
 
-#endif // Apple
+#endif // Cases for semaphore support
 
 /**
  * Producer consumer queue safe for multiple producers and multiple consumers.
@@ -126,8 +168,10 @@ inline void WaitSemaphore(Semaphore &on) {
 template <class T> class PCQueue {
  public:
   explicit PCQueue(size_t size)
-      : empty_(size), used_(0), storage_(new T[size]),
-        end_(storage_.get() + size), produce_at_(storage_.get()),
+   : empty_(size), used_(0),
+     storage_(new T[size]),
+     end_(storage_.get() + size),
+     produce_at_(storage_.get()),
      consume_at_(storage_.get()) {}
 
   // Add a value to the queue.
@@ -141,8 +185,7 @@ public:
         empty_.post();
         throw;
       }
-      if (++produce_at_ == end_)
-        produce_at_ = storage_.get();
+      if (++produce_at_ == end_) produce_at_ = storage_.get();
     }
     used_.post();
   }
@@ -158,12 +201,12 @@ public:
         empty_.post();
         throw;
       }
-      if (++produce_at_ == end_)
-        produce_at_ = storage_.get();
+      if (++produce_at_ == end_) produce_at_ = storage_.get();
     }
     used_.post();
   }
 
+
   // Consume a value, assigning it to out.
   T& Consume(T &out) {
     WaitSemaphore(used_);
@@ -175,8 +218,7 @@ public:
         used_.post();
         throw;
       }
-      if (++consume_at_ == end_)
-        consume_at_ = storage_.get();
+      if (++consume_at_ == end_) consume_at_ = storage_.get();
     }
     empty_.post();
     return out;
@@ -193,13 +235,13 @@ public:
         used_.post();
         throw;
       }
-      if (++consume_at_ == end_)
-        consume_at_ = storage_.get();
+      if (++consume_at_ == end_) consume_at_ = storage_.get();
     }
     empty_.post();
     return out;
   }
 
+
   // Convenience version of Consume that copies the value to return.
   // The other version is faster.
   T Consume() {
@@ -250,7 +292,9 @@ public:
       valid_.post();
     }
 
-  void Produce(const T &val) { Produce(T(val)); }
+    void Produce(const T &val) {
+      Produce(T(val));
+    }
 
     T& Consume(T &out) {
       WaitSemaphore(valid_);
@@ -264,7 +308,9 @@ public:
     // Warning: very much a no-guarantees race-condition-rich implementation!
     // But sufficient for our specific purpose: The single thread that consumes
     // is also the only one that checks Empty, and knows that it's racing.
-  bool Empty() const { return reading_current_ == filling_current_; }
+    bool Empty() const {
+      return reading_current_ == filling_current_;
+    }
 
   private:
     void SetFilling(UnboundedPage<T> *to) {