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