chore: small fixes post #605

util/fibers/fiberqueue_threadpool.cc

@@ -38,22 +38,26 @@ void FiberQueue::Run() {
     return false;
   };
 
-  while (true) {
-    pull_ec_.await(cb);
+  try {
+    while (true) {
+      pull_ec_.await(cb);
 
-    if (is_closed)
-      break;
+      if (is_closed)
+        break;
 
-    while (count < kBatchSize && queue_.try_dequeue_sc(batch[count])) {
-      ++count;
-    }
+      while (count < kBatchSize && queue_.try_dequeue_sc(batch[count])) {
+        ++count;
+      }
 
-    push_ec_.notifyAll();
+      push_ec_.notifyAll();
 
-    for (unsigned i = 0; i < count; ++i) {
-      batch[i]();
+      for (unsigned i = 0; i < count; ++i) {
+        batch[i]();
+      }
+      count = 0;
     }
-    count = 0;
+  } catch (const std::exception& ex) {
+    LOG(FATAL) << "Exception " << ex.what();
   }
 }
 

util/fibers/fiberqueue_threadpool.h

@@ -108,7 +108,7 @@ class FiberQueue {
 
   void Run();
 
-  unsigned blocked_submitters() const {
+  static unsigned blocked_submitters() {
     return blocked_submitters_;
   }
 
@@ -173,7 +173,9 @@ class FiberQueueThreadPool {
         break;
       }
 
+      main_w.q->blocked_submitters_++;
       main_w.q->push_ec_.wait(key.epoch());
+      main_w.q->blocked_submitters_--;
     }
   }