src/cpp/thread_manager/thread_manager.h

   1 /*
   2  *
   3  * Copyright 2016 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #ifndef GRPC_INTERNAL_CPP_THREAD_MANAGER_H
  20 #define GRPC_INTERNAL_CPP_THREAD_MANAGER_H
  21
  22 #include <condition_variable>
  23 #include <list>
  24 #include <memory>
  25 #include <mutex>
  26
  27 #include <grpcpp/support/config.h>
  28
  29 #include "src/core/lib/gprpp/sync.h"
  30 #include "src/core/lib/gprpp/thd.h"
  31 #include "src/core/lib/iomgr/resource_quota.h"
  32
  33 namespace grpc {
  34
  35 class ThreadManager {
  36  public:
  37   explicit ThreadManager(const char* name, grpc_resource_quota* resource_quota,
  38                          int min_pollers, int max_pollers);
  39   virtual ~ThreadManager();
  40
  41   // Initializes and Starts the Rpc Manager threads
  42   void Initialize();
  43
  44   // The return type of PollForWork() function
  45   enum WorkStatus { WORK_FOUND, SHUTDOWN, TIMEOUT };
  46
  47   // "Polls" for new work.
  48   // If the return value is WORK_FOUND:
  49   //  - The implementaion of PollForWork() MAY set some opaque identifier to
  50   //    (identify the work item found) via the '*tag' parameter
  51   //  - The implementaion MUST set the value of 'ok' to 'true' or 'false'. A
  52   //    value of 'false' indicates some implemenation specific error (that is
  53   //    neither SHUTDOWN nor TIMEOUT)
  54   //  - ThreadManager does not interpret the values of 'tag' and 'ok'
  55   //  - ThreadManager WILL call DoWork() and pass '*tag' and 'ok' as input to
  56   //    DoWork()
  57   //
  58   // If the return value is SHUTDOWN:,
  59   //  - ThreadManager WILL NOT call DoWork() and terminates the thread
  60   //
  61   // If the return value is TIMEOUT:,
  62   //  - ThreadManager WILL NOT call DoWork()
  63   //  - ThreadManager MAY terminate the thread depending on the current number
  64   //    of active poller threads and mix_pollers/max_pollers settings
  65   //  - Also, the value of timeout is specific to the derived class
  66   //    implementation
  67   virtual WorkStatus PollForWork(void** tag, bool* ok) = 0;
  68
  69   // The implementation of DoWork() is supposed to perform the work found by
  70   // PollForWork(). The tag and ok parameters are the same as returned by
  71   // PollForWork(). The resources parameter indicates that the call actually
  72   // has the resources available for performing the RPC's work. If it doesn't,
  73   // the implementation should fail it appropriately.
  74   //
  75   // The implementation of DoWork() should also do any setup needed to ensure
  76   // that the next call to PollForWork() (not necessarily by the current thread)
  77   // actually finds some work
  78   virtual void DoWork(void* tag, bool ok, bool resources) = 0;
  79
  80   // Mark the ThreadManager as shutdown and begin draining the work. This is a
  81   // non-blocking call and the caller should call Wait(), a blocking call which
  82   // returns only once the shutdown is complete
  83   virtual void Shutdown();
  84
  85   // Has Shutdown() been called
  86   bool IsShutdown();
  87
  88   // A blocking call that returns only after the ThreadManager has shutdown and
  89   // all the threads have drained all the outstanding work
  90   virtual void Wait();
  91
  92   // Max number of concurrent threads that were ever active in this thread
  93   // manager so far. This is useful for debugging purposes (and in unit tests)
  94   // to check if resource_quota is properly being enforced.
  95   int GetMaxActiveThreadsSoFar();
  96
  97  private:
  98   // Helper wrapper class around grpc_core::Thread. Takes a ThreadManager object
  99   // and starts a new grpc_core::Thread to calls the Run() function.
 100   //
 101   // The Run() function calls ThreadManager::MainWorkLoop() function and once
 102   // that completes, it marks the WorkerThread completed by calling
 103   // ThreadManager::MarkAsCompleted()
 104   //
 105   // WHY IS THIS NEEDED?:
 106   // When a thread terminates, some other thread *must* call Join() on that
 107   // thread so that the resources are released. Having a WorkerThread wrapper
 108   // will make this easier. Once Run() completes, each thread calls the
 109   // following two functions:
 110   //    ThreadManager::CleanupCompletedThreads()
 111   //    ThreadManager::MarkAsCompleted()
 112   //
 113   //  - MarkAsCompleted() puts the WorkerThread object in the ThreadManger's
 114   //    completed_threads_ list
 115   //  - CleanupCompletedThreads() calls "Join()" on the threads that are already
 116   //    in the completed_threads_ list  (since a thread cannot call Join() on
 117   //    itself, it calls CleanupCompletedThreads() *before* calling
 118   //    MarkAsCompleted())
 119   //
 120   // TODO(sreek): Consider creating the threads 'detached' so that Join() need
 121   // not be called (and the need for this WorkerThread class is eliminated)
 122   class WorkerThread {
 123    public:
 124     WorkerThread(ThreadManager* thd_mgr);
 125     ~WorkerThread();
 126
 127    private:
 128     // Calls thd_mgr_->MainWorkLoop() and once that completes, calls
 129     // thd_mgr_>MarkAsCompleted(this) to mark the thread as completed
 130     void Run();
 131
 132     ThreadManager* const thd_mgr_;
 133     grpc_core::Thread thd_;
 134   };
 135
 136   // The main function in ThreadManager
 137   void MainWorkLoop();
 138
 139   void MarkAsCompleted(WorkerThread* thd);
 140   void CleanupCompletedThreads();
 141
 142   // Protects shutdown_, num_pollers_, num_threads_ and
 143   // max_active_threads_sofar_
 144   grpc_core::Mutex mu_;
 145
 146   bool shutdown_;
 147   grpc_core::CondVar shutdown_cv_;
 148
 149   // The resource user object to use when requesting quota to create threads
 150   //
 151   // Note: The user of this ThreadManager object must create grpc_resource_quota
 152   // object (that contains the actual max thread quota) and a grpc_resource_user
 153   // object through which quota is requested whenver new threads need to be
 154   // created
 155   grpc_resource_user* resource_user_;
 156
 157   // Number of threads doing polling
 158   int num_pollers_;
 159
 160   // The minimum and maximum number of threads that should be doing polling
 161   int min_pollers_;
 162   int max_pollers_;
 163
 164   // The total number of threads currently active (includes threads includes the
 165   // threads that are currently polling i.e num_pollers_)
 166   int num_threads_;
 167
 168   // See GetMaxActiveThreadsSoFar()'s description.
 169   // To be more specific, this variable tracks the max value num_threads_ was
 170   // ever set so far
 171   int max_active_threads_sofar_;
 172
 173   grpc_core::Mutex list_mu_;
 174   std::list<WorkerThread*> completed_threads_;
 175 };
 176
 177 }  // namespace grpc
 178
 179 #endif  // GRPC_INTERNAL_CPP_THREAD_MANAGER_H