GRPC C++  1.39.1
thread_manager.h
Go to the documentation of this file.
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 
28 #include "src/core/lib/gprpp/thd.h"
30 
31 namespace grpc {
32 
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
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.
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 
125  bool created() const { return created_; }
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
Definition: thread_manager.h:33
virtual void Shutdown()
Definition: thread_manager.cc:82
ThreadManager(const char *name, grpc_resource_quota *resource_quota, int min_pollers, int max_pollers)
Definition: thread_manager.cc:52
int GetMaxActiveThreadsSoFar()
Definition: thread_manager.cc:92
WorkStatus
Definition: thread_manager.h:43
@ TIMEOUT
Definition: thread_manager.h:43
@ WORK_FOUND
Definition: thread_manager.h:43
@ SHUTDOWN
Definition: thread_manager.h:43
virtual ~ThreadManager()
Definition: thread_manager.cc:64
void Initialize()
Definition: thread_manager.cc:126
virtual void DoWork(void *tag, bool ok, bool resources)=0
virtual WorkStatus PollForWork(void **tag, bool *ok)=0
virtual void Wait()
Definition: thread_manager.cc:75
bool IsShutdown()
Definition: thread_manager.cc:87
Definition: sync.h:123
Definition: sync.h:59
Definition: thd.h:46
An Alarm posts the user-provided tag to its associated completion queue or invokes the user-provided ...
Definition: alarm.h:33
Definition: resource_quota.cc:127
Definition: resource_quota.cc:65