Saturation detector. Naming the queues. Diff 3/7

Summary: we should report the name of the executore/threadpool that is saturated. propagate names of the executor/threadpool onto the queue we are monitoring.

Reviewed By: andriigrynenko

Differential Revision: D20773871

fbshipit-source-id: f7f78edd62d515cca19e30a63773180404843c1b

folly/concurrency/QueueObserver.cpp

@@ -18,16 +18,17 @@
 
 namespace {
 std::unique_ptr<folly::QueueObserverFactory>
-make_queue_observer_factory_fallback() noexcept {
-  return {};
+make_queue_observer_factory_fallback(const std::string&, size_t) noexcept {
+  return std::unique_ptr<folly::QueueObserverFactory>();
 }
 } // namespace
 
 namespace folly {
-/* static */ std::unique_ptr<QueueObserverFactory>
-QueueObserverFactory::make() {
+/* static */ std::unique_ptr<QueueObserverFactory> QueueObserverFactory::make(
+    const std::string& context,
+    size_t numPriorities) {
   auto f = make_queue_observer_factory ? make_queue_observer_factory
                                        : make_queue_observer_factory_fallback;
-  return f();
+  return f(context, numPriorities);
 }
 } // namespace folly

folly/concurrency/QueueObserver.h

@@ -18,6 +18,7 @@
 
 #include <stdint.h>
 #include <memory>
+#include <string>
 
 #include <folly/Portability.h>
 
@@ -34,12 +35,15 @@ class QueueObserver {
 class QueueObserverFactory {
  public:
   virtual ~QueueObserverFactory() {}
-  virtual std::unique_ptr<QueueObserver> create() = 0;
+  virtual std::unique_ptr<QueueObserver> create(size_t pri) = 0;
 
-  static std::unique_ptr<QueueObserverFactory> make();
+  static std::unique_ptr<QueueObserverFactory> make(
+      const std::string& context,
+      size_t numPriorities);
 };
 
-using MakeQueueObserverFactory = std::unique_ptr<QueueObserverFactory>();
+using MakeQueueObserverFactory =
+    std::unique_ptr<QueueObserverFactory>(const std::string&, size_t);
 #if FOLLY_HAVE_WEAK_SYMBOLS
 FOLLY_ATTR_WEAK MakeQueueObserverFactory make_queue_observer_factory;
 #else

folly/executors/CPUThreadPoolExecutor.cpp

@@ -146,7 +146,7 @@ CPUThreadPoolExecutor::getQueueObserver(int8_t pri) {
     return getQueueObserver(0);
   }
   QueueObserver* existingObserver = nullptr;
-  QueueObserver* newObserver = queueObserverFactory_->create().release();
+  QueueObserver* newObserver = queueObserverFactory_->create(pri).release();
   if (!slot.compare_exchange_strong(existingObserver, newObserver)) {
     delete newObserver;
     return existingObserver;
@@ -287,7 +287,8 @@ CPUThreadPoolExecutor::createQueueObserverFactory() {
   for (auto& observer : queueObservers_) {
     observer.store(nullptr, std::memory_order_release);
   }
-  return QueueObserverFactory::make();
+  return QueueObserverFactory::make(
+      "cpu." + getName(), taskQueue_->getNumPriorities());
 }
 
 } // namespace folly