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