intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

2 files changed, 446 insertions, 0 deletions

diff --git a/contrib/libs/grpc/src/cpp/thread_manager/thread_manager.cc b/contrib/libs/grpc/src/cpp/thread_manager/thread_manager.cc
new file mode 100644
index 00000000000..c8560aa81dd
--- /dev/null
+++ b/contrib/libs/grpc/src/cpp/thread_manager/thread_manager.cc
@@ -0,0 +1,265 @@
+/*
+ *
+ * Copyright 2016 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "src/cpp/thread_manager/thread_manager.h"
+
+#include <climits>
+
+#include <grpc/support/log.h>
+#include "src/core/lib/gprpp/thd.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+
+namespace grpc {
+
+ThreadManager::WorkerThread::WorkerThread(ThreadManager* thd_mgr)
+    : thd_mgr_(thd_mgr) {
+  // Make thread creation exclusive with respect to its join happening in
+  // ~WorkerThread().
+  thd_ = grpc_core::Thread(
+      "grpcpp_sync_server",
+      [](void* th) { static_cast<ThreadManager::WorkerThread*>(th)->Run(); },
+      this, &created_);
+  if (!created_) {
+    gpr_log(GPR_ERROR, "Could not create grpc_sync_server worker-thread");
+  }
+}
+
+void ThreadManager::WorkerThread::Run() {
+  thd_mgr_->MainWorkLoop();
+  thd_mgr_->MarkAsCompleted(this);
+}
+
+ThreadManager::WorkerThread::~WorkerThread() {
+  // Don't join until the thread is fully constructed.
+  thd_.Join();
+}
+
+ThreadManager::ThreadManager(const char* name,
+                             grpc_resource_quota* resource_quota,
+                             int min_pollers, int max_pollers)
+    : shutdown_(false),
+      num_pollers_(0),
+      min_pollers_(min_pollers),
+      max_pollers_(max_pollers == -1 ? INT_MAX : max_pollers),
+      num_threads_(0),
+      max_active_threads_sofar_(0) {
+  resource_user_ = grpc_resource_user_create(resource_quota, name);
+}
+
+ThreadManager::~ThreadManager() {
+  {
+    grpc_core::MutexLock lock(&mu_);
+    GPR_ASSERT(num_threads_ == 0);
+  }
+
+  grpc_core::ExecCtx exec_ctx;  // grpc_resource_user_unref needs an exec_ctx
+  grpc_resource_user_unref(resource_user_);
+  CleanupCompletedThreads();
+}
+
+void ThreadManager::Wait() {
+  grpc_core::MutexLock lock(&mu_);
+  while (num_threads_ != 0) {
+    shutdown_cv_.Wait(&mu_);
+  }
+}
+
+void ThreadManager::Shutdown() {
+  grpc_core::MutexLock lock(&mu_);
+  shutdown_ = true;
+}
+
+bool ThreadManager::IsShutdown() {
+  grpc_core::MutexLock lock(&mu_);
+  return shutdown_;
+}
+
+int ThreadManager::GetMaxActiveThreadsSoFar() {
+  grpc_core::MutexLock list_lock(&list_mu_);
+  return max_active_threads_sofar_;
+}
+
+void ThreadManager::MarkAsCompleted(WorkerThread* thd) {
+  {
+    grpc_core::MutexLock list_lock(&list_mu_);
+    completed_threads_.push_back(thd);
+  }
+
+  {
+    grpc_core::MutexLock lock(&mu_);
+    num_threads_--;
+    if (num_threads_ == 0) {
+      shutdown_cv_.Signal();
+    }
+  }
+
+  // Give a thread back to the resource quota
+  grpc_resource_user_free_threads(resource_user_, 1);
+}
+
+void ThreadManager::CleanupCompletedThreads() {
+  std::list<WorkerThread*> completed_threads;
+  {
+    // swap out the completed threads list: allows other threads to clean up
+    // more quickly
+    grpc_core::MutexLock lock(&list_mu_);
+    completed_threads.swap(completed_threads_);
+  }
+  for (auto thd : completed_threads) delete thd;
+}
+
+void ThreadManager::Initialize() {
+  if (!grpc_resource_user_allocate_threads(resource_user_, min_pollers_)) {
+    gpr_log(GPR_ERROR,
+            "No thread quota available to even create the minimum required "
+            "polling threads (i.e %d). Unable to start the thread manager",
+            min_pollers_);
+    abort();
+  }
+
+  {
+    grpc_core::MutexLock lock(&mu_);
+    num_pollers_ = min_pollers_;
+    num_threads_ = min_pollers_;
+    max_active_threads_sofar_ = min_pollers_;
+  }
+
+  for (int i = 0; i < min_pollers_; i++) {
+    WorkerThread* worker = new WorkerThread(this);
+    GPR_ASSERT(worker->created());  // Must be able to create the minimum
+    worker->Start();
+  }
+}
+
+void ThreadManager::MainWorkLoop() {
+  while (true) {
+    void* tag;
+    bool ok;
+    WorkStatus work_status = PollForWork(&tag, &ok);
+
+    grpc_core::ReleasableMutexLock lock(&mu_);
+    // Reduce the number of pollers by 1 and check what happened with the poll
+    num_pollers_--;
+    bool done = false;
+    switch (work_status) {
+      case TIMEOUT:
+        // If we timed out and we have more pollers than we need (or we are
+        // shutdown), finish this thread
+        if (shutdown_ || num_pollers_ > max_pollers_) done = true;
+        break;
+      case SHUTDOWN:
+        // If the thread manager is shutdown, finish this thread
+        done = true;
+        break;
+      case WORK_FOUND:
+        // If we got work and there are now insufficient pollers and there is
+        // quota available to create a new thread, start a new poller thread
+        bool resource_exhausted = false;
+        if (!shutdown_ && num_pollers_ < min_pollers_) {
+          if (grpc_resource_user_allocate_threads(resource_user_, 1)) {
+            // We can allocate a new poller thread
+            num_pollers_++;
+            num_threads_++;
+            if (num_threads_ > max_active_threads_sofar_) {
+              max_active_threads_sofar_ = num_threads_;
+            }
+            // Drop lock before spawning thread to avoid contention
+            lock.Unlock();
+            WorkerThread* worker = new WorkerThread(this);
+            if (worker->created()) {
+              worker->Start();
+            } else {
+              // Get lock again to undo changes to poller/thread counters.
+              grpc_core::MutexLock failure_lock(&mu_);
+              num_pollers_--;
+              num_threads_--;
+              resource_exhausted = true;
+              delete worker;
+            }
+          } else if (num_pollers_ > 0) {
+            // There is still at least some thread polling, so we can go on
+            // even though we are below the number of pollers that we would
+            // like to have (min_pollers_)
+            lock.Unlock();
+          } else {
+            // There are no pollers to spare and we couldn't allocate
+            // a new thread, so resources are exhausted!
+            lock.Unlock();
+            resource_exhausted = true;
+          }
+        } else {
+          // There are a sufficient number of pollers available so we can do
+          // the work and continue polling with our existing poller threads
+          lock.Unlock();
+        }
+        // Lock is always released at this point - do the application work
+        // or return resource exhausted if there is new work but we couldn't
+        // get a thread in which to do it.
+        DoWork(tag, ok, !resource_exhausted);
+        // Take the lock again to check post conditions
+        lock.Lock();
+        // If we're shutdown, we should finish at this point.
+        if (shutdown_) done = true;
+        break;
+    }
+    // If we decided to finish the thread, break out of the while loop
+    if (done) break;
+
+    // Otherwise go back to polling as long as it doesn't exceed max_pollers_
+    //
+    // **WARNING**:
+    // There is a possibility of threads thrashing here (i.e excessive thread
+    // shutdowns and creations than the ideal case). This happens if max_poller_
+    // count is small and the rate of incoming requests is also small. In such
+    // scenarios we can possibly configure max_pollers_ to a higher value and/or
+    // increase the cq timeout.
+    //
+    // However, not doing this check here and unconditionally incrementing
+    // num_pollers (and hoping that the system will eventually settle down) has
+    // far worse consequences i.e huge number of threads getting created to the
+    // point of thread-exhaustion. For example: if the incoming request rate is
+    // very high, all the polling threads will return very quickly from
+    // PollForWork() with WORK_FOUND. They all briefly decrement num_pollers_
+    // counter thereby possibly - and briefly - making it go below min_pollers;
+    // This will most likely result in the creation of a new poller since
+    // num_pollers_ dipped below min_pollers_.
+    //
+    // Now, If we didn't do the max_poller_ check here, all these threads will
+    // go back to doing PollForWork() and the whole cycle repeats (with a new
+    // thread being added in each cycle). Once the total number of threads in
+    // the system crosses a certain threshold (around ~1500), there is heavy
+    // contention on mutexes (the mu_ here or the mutexes in gRPC core like the
+    // pollset mutex) that makes DoWork() take longer to finish thereby causing
+    // new poller threads to be created even faster. This results in a thread
+    // avalanche.
+    if (num_pollers_ < max_pollers_) {
+      num_pollers_++;
+    } else {
+      break;
+    }
+  };
+
+  // This thread is exiting. Do some cleanup work i.e delete already completed
+  // worker threads
+  CleanupCompletedThreads();
+
+  // If we are here, either ThreadManager is shutting down or it already has
+  // enough threads.
+}
+
+}  // namespace grpc
diff --git a/contrib/libs/grpc/src/cpp/thread_manager/thread_manager.h b/contrib/libs/grpc/src/cpp/thread_manager/thread_manager.h
new file mode 100644
index 00000000000..43f1fd5585f
--- /dev/null
+++ b/contrib/libs/grpc/src/cpp/thread_manager/thread_manager.h
@@ -0,0 +1,181 @@
+/*
+ *
+ * Copyright 2016 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CPP_THREAD_MANAGER_H
+#define GRPC_INTERNAL_CPP_THREAD_MANAGER_H
+
+#include <list>
+#include <memory>
+
+#include <grpcpp/support/config.h>
+
+#include "src/core/lib/gprpp/sync.h"
+#include "src/core/lib/gprpp/thd.h"
+#include "src/core/lib/iomgr/resource_quota.h"
+
+namespace grpc {
+
+class ThreadManager {
+ public:
+  explicit ThreadManager(const char* name, grpc_resource_quota* resource_quota,
+                         int min_pollers, int max_pollers);
+  virtual ~ThreadManager();
+
+  // Initializes and Starts the Rpc Manager threads
+  void Initialize();
+
+  // The return type of PollForWork() function
+  enum WorkStatus { WORK_FOUND, SHUTDOWN, TIMEOUT };
+
+  // "Polls" for new work.
+  // If the return value is WORK_FOUND:
+  //  - The implementaion of PollForWork() MAY set some opaque identifier to
+  //    (identify the work item found) via the '*tag' parameter
+  //  - The implementaion MUST set the value of 'ok' to 'true' or 'false'. A
+  //    value of 'false' indicates some implemenation specific error (that is
+  //    neither SHUTDOWN nor TIMEOUT)
+  //  - ThreadManager does not interpret the values of 'tag' and 'ok'
+  //  - ThreadManager WILL call DoWork() and pass '*tag' and 'ok' as input to
+  //    DoWork()
+  //
+  // If the return value is SHUTDOWN:,
+  //  - ThreadManager WILL NOT call DoWork() and terminates the thread
+  //
+  // If the return value is TIMEOUT:,
+  //  - ThreadManager WILL NOT call DoWork()
+  //  - ThreadManager MAY terminate the thread depending on the current number
+  //    of active poller threads and mix_pollers/max_pollers settings
+  //  - Also, the value of timeout is specific to the derived class
+  //    implementation
+  virtual WorkStatus PollForWork(void** tag, bool* ok) = 0;
+
+  // The implementation of DoWork() is supposed to perform the work found by
+  // PollForWork(). The tag and ok parameters are the same as returned by
+  // PollForWork(). The resources parameter indicates that the call actually
+  // has the resources available for performing the RPC's work. If it doesn't,
+  // the implementation should fail it appropriately.
+  //
+  // The implementation of DoWork() should also do any setup needed to ensure
+  // that the next call to PollForWork() (not necessarily by the current thread)
+  // actually finds some work
+  virtual void DoWork(void* tag, bool ok, bool resources) = 0;
+
+  // Mark the ThreadManager as shutdown and begin draining the work. This is a
+  // non-blocking call and the caller should call Wait(), a blocking call which
+  // returns only once the shutdown is complete
+  virtual void Shutdown();
+
+  // Has Shutdown() been called
+  bool IsShutdown();
+
+  // A blocking call that returns only after the ThreadManager has shutdown and
+  // all the threads have drained all the outstanding work
+  virtual void Wait();
+
+  // Max number of concurrent threads that were ever active in this thread
+  // manager so far. This is useful for debugging purposes (and in unit tests)
+  // to check if resource_quota is properly being enforced.
+  int GetMaxActiveThreadsSoFar();
+
+ private:
+  // Helper wrapper class around grpc_core::Thread. Takes a ThreadManager object
+  // and starts a new grpc_core::Thread to calls the Run() function.
+  //
+  // The Run() function calls ThreadManager::MainWorkLoop() function and once
+  // that completes, it marks the WorkerThread completed by calling
+  // ThreadManager::MarkAsCompleted()
+  //
+  // WHY IS THIS NEEDED?:
+  // When a thread terminates, some other thread *must* call Join() on that
+  // thread so that the resources are released. Having a WorkerThread wrapper
+  // will make this easier. Once Run() completes, each thread calls the
+  // following two functions:
+  //    ThreadManager::CleanupCompletedThreads()
+  //    ThreadManager::MarkAsCompleted()
+  //
+  //  - MarkAsCompleted() puts the WorkerThread object in the ThreadManger's
+  //    completed_threads_ list
+  //  - CleanupCompletedThreads() calls "Join()" on the threads that are already
+  //    in the completed_threads_ list  (since a thread cannot call Join() on
+  //    itself, it calls CleanupCompletedThreads() *before* calling
+  //    MarkAsCompleted())
+  //
+  // TODO(sreek): Consider creating the threads 'detached' so that Join() need
+  // not be called (and the need for this WorkerThread class is eliminated)
+  class WorkerThread {
+   public:
+    WorkerThread(ThreadManager* thd_mgr);
+    ~WorkerThread();
+
+    bool created() const { return created_; }
+    void Start() { thd_.Start(); }
+
+   private:
+    // Calls thd_mgr_->MainWorkLoop() and once that completes, calls
+    // thd_mgr_>MarkAsCompleted(this) to mark the thread as completed
+    void Run();
+
+    ThreadManager* const thd_mgr_;
+    grpc_core::Thread thd_;
+    bool created_;
+  };
+
+  // The main function in ThreadManager
+  void MainWorkLoop();
+
+  void MarkAsCompleted(WorkerThread* thd);
+  void CleanupCompletedThreads();
+
+  // Protects shutdown_, num_pollers_, num_threads_ and
+  // max_active_threads_sofar_
+  grpc_core::Mutex mu_;
+
+  bool shutdown_;
+  grpc_core::CondVar shutdown_cv_;
+
+  // The resource user object to use when requesting quota to create threads
+  //
+  // Note: The user of this ThreadManager object must create grpc_resource_quota
+  // object (that contains the actual max thread quota) and a grpc_resource_user
+  // object through which quota is requested whenever new threads need to be
+  // created
+  grpc_resource_user* resource_user_;
+
+  // Number of threads doing polling
+  int num_pollers_;
+
+  // The minimum and maximum number of threads that should be doing polling
+  int min_pollers_;
+  int max_pollers_;
+
+  // The total number of threads currently active (includes threads includes the
+  // threads that are currently polling i.e num_pollers_)
+  int num_threads_;
+
+  // See GetMaxActiveThreadsSoFar()'s description.
+  // To be more specific, this variable tracks the max value num_threads_ was
+  // ever set so far
+  int max_active_threads_sofar_;
+
+  grpc_core::Mutex list_mu_;
+  std::list<WorkerThread*> completed_threads_;
+};
+
+}  // namespace grpc
+
+#endif  // GRPC_INTERNAL_CPP_THREAD_MANAGER_H