[concurrent-blocking-queue-fix] ConcurrentBlockingQueue::Pop's return…

src/common/concurrent_blocking_queue.h

@@ -31,7 +31,7 @@ template<typename T> class ConcurrentBlockingQueue {
         queue_.pop_front();
         if (queue_.empty())
           has_elmt_ = false;
-        return false;
+        return true;
       }
     }
     {
@@ -44,9 +44,9 @@ template<typename T> class ConcurrentBlockingQueue {
         queue_.pop_front();
         if (queue_.empty())
           has_elmt_ = false;
-        return false;
+        return true;
       } else {
-        return true;        
+        return false;        
       }
     }
   }

src/common/spinlock.h

@@ -0,0 +1,44 @@
+#ifndef MXNET_COMMON_SPINLOCK_H_
+#define MXNET_COMMON_SPINLOCK_H_
+
+#include <atomic>
+
+namespace mxnet {
+namespace common {
+
+/*!
+ * \brief Simple userspace spinlock implementation.
+ */
+class Spinlock {
+ public:
+  Spinlock() = default;
+  /*!
+   * \brief Disable copy and move.
+   */
+  Spinlock(Spinlock const&) = delete;
+  Spinlock(Spinlock&&) = delete;
+  Spinlock& operator=(Spinlock const&) = delete;
+  Spinlock& operator=(Spinlock&&) = delete;
+  ~Spinlock() = default;
+  /*!
+   * \brief Acquire lock.
+   */
+  void lock() noexcept {
+    while (lock_.test_and_set(std::memory_order_acquire));
+  }
+  /*!
+   * \brief Release lock.
+   */
+  void unlock() noexcept {
+    lock_.clear(std::memory_order_release);
+  }
+
+ private:
+  std::atomic_flag lock_ = ATOMIC_FLAG_INIT;
+};
+
+}  // namespace common
+}  // namespace mxnet
+
+#endif  // MXNET_COMMON_SPINLOCK_H_
+

src/dag_engine/threaded_engine.cc

@@ -159,7 +159,7 @@ class ThreadedEngine : public DAGEngine {
   }
   void WorkerRoutine(int thrid) {
     OpDescr* opd = nullptr;
-    while(! worker_queues_[thrid]->Pop(opd)) {
+    while(worker_queues_[thrid]->Pop(opd)) {
       LOG(INFO) << "worker thread #" << thrid << " got operator " << opd;
       opd->op(GetRunContext(opd->exec_ctx), [this, opd] () { this->OnOpFinished(opd); });
       opd = nullptr;