blob: 6e68f1a679dffec1f82b45d80ce386eef702bb4a [file] [log] [blame]
henrike@webrtc.orgf0488722014-05-13 18:00:261/*
2 * Copyright 2004 The WebRTC Project Authors. All rights reserved.
3 *
4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
10
Mirko Bonadei92ea95e2017-09-15 04:47:3111#ifndef RTC_BASE_THREAD_H_
12#define RTC_BASE_THREAD_H_
henrike@webrtc.orgf0488722014-05-13 18:00:2613
Yves Gerey988cc082018-10-23 10:03:0114#include <stdint.h>
Jonas Olssona4d87372019-07-05 17:08:3315
Henrik Kjellanderec78f1c2017-06-29 05:52:5016#include <list>
Sebastian Janssonda7267a2020-03-03 09:48:0517#include <map>
Henrik Kjellanderec78f1c2017-06-29 05:52:5018#include <memory>
Sebastian Jansson6ea2c6a2020-01-13 13:07:2219#include <queue>
Sebastian Janssonda7267a2020-03-03 09:48:0520#include <set>
Henrik Kjellanderec78f1c2017-06-29 05:52:5021#include <string>
Yves Gerey988cc082018-10-23 10:03:0122#include <type_traits>
Sebastian Jansson6ea2c6a2020-01-13 13:07:2223#include <vector>
henrike@webrtc.orgf0488722014-05-13 18:00:2624
Henrik Kjellanderec78f1c2017-06-29 05:52:5025#if defined(WEBRTC_POSIX)
26#include <pthread.h>
27#endif
Danil Chapovalov89313452019-11-29 11:56:4328#include "api/function_view.h"
Danil Chapovalov912b3b82019-11-22 14:52:4029#include "api/task_queue/queued_task.h"
30#include "api/task_queue/task_queue_base.h"
Steve Anton10542f22019-01-11 17:11:0031#include "rtc_base/constructor_magic.h"
Markus Handell3cb525b2020-07-16 14:16:0932#include "rtc_base/deprecated/recursive_critical_section.h"
Yves Gerey988cc082018-10-23 10:03:0133#include "rtc_base/location.h"
Steve Anton10542f22019-01-11 17:11:0034#include "rtc_base/message_handler.h"
Mirko Bonadei92ea95e2017-09-15 04:47:3135#include "rtc_base/platform_thread_types.h"
Steve Anton10542f22019-01-11 17:11:0036#include "rtc_base/socket_server.h"
Mirko Bonadei35214fc2019-09-23 12:54:2837#include "rtc_base/system/rtc_export.h"
Yves Gerey988cc082018-10-23 10:03:0138#include "rtc_base/thread_annotations.h"
Sebastian Jansson6ea2c6a2020-01-13 13:07:2239#include "rtc_base/thread_message.h"
Henrik Kjellanderec78f1c2017-06-29 05:52:5040
41#if defined(WEBRTC_WIN)
Mirko Bonadei92ea95e2017-09-15 04:47:3142#include "rtc_base/win32.h"
Henrik Kjellanderec78f1c2017-06-29 05:52:5043#endif
44
Tommife041642021-04-07 08:08:2845#if RTC_DCHECK_IS_ON
46// Counts how many blocking Thread::Invoke or Thread::Send calls are made from
47// within a scope and logs the number of blocking calls at the end of the scope.
48#define RTC_LOG_THREAD_BLOCK_COUNT() \
49 rtc::Thread::ScopedCountBlockingCalls blocked_call_count_printer( \
50 [func = __func__](uint32_t actual_block, uint32_t could_block) { \
51 auto total = actual_block + could_block; \
52 if (total) { \
53 RTC_LOG(LS_WARNING) << "Blocking " << func << ": total=" << total \
54 << " (actual=" << actual_block \
55 << ", could=" << could_block << ")"; \
56 } \
57 })
58
59// Adds an RTC_DCHECK_LE that checks that the number of blocking calls are
60// less than or equal to a specific value. Use to avoid regressing in the
61// number of blocking thread calls.
62// Note: Use of this macro, requires RTC_LOG_THREAD_BLOCK_COUNT() to be called
63// first.
Tomas Gunnarsson89f3dd52021-04-14 10:54:1064#define RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(x) \
65 do { \
66 blocked_call_count_printer.set_minimum_call_count_for_callback(x + 1); \
67 RTC_DCHECK_LE(blocked_call_count_printer.GetTotalBlockedCallCount(), x); \
68 } while (0)
Tommife041642021-04-07 08:08:2869#else
70#define RTC_LOG_THREAD_BLOCK_COUNT()
71#define RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(x)
72#endif
73
Henrik Kjellanderec78f1c2017-06-29 05:52:5074namespace rtc {
75
76class Thread;
77
Henrik Boströmba4dcc32019-02-28 08:34:0678namespace rtc_thread_internal {
79
Niels Möllerf13a0962019-05-17 08:15:0680class MessageLikeTask : public MessageData {
Henrik Boströmba4dcc32019-02-28 08:34:0681 public:
Niels Möllerf13a0962019-05-17 08:15:0682 virtual void Run() = 0;
83};
84
85template <class FunctorT>
86class MessageWithFunctor final : public MessageLikeTask {
87 public:
88 explicit MessageWithFunctor(FunctorT&& functor)
Henrik Boströmba4dcc32019-02-28 08:34:0689 : functor_(std::forward<FunctorT>(functor)) {}
90
Niels Möllerf13a0962019-05-17 08:15:0691 void Run() override { functor_(); }
Henrik Boströmba4dcc32019-02-28 08:34:0692
93 private:
Niels Möllerf13a0962019-05-17 08:15:0694 ~MessageWithFunctor() override {}
Henrik Boströmba4dcc32019-02-28 08:34:0695
96 typename std::remove_reference<FunctorT>::type functor_;
97
Niels Möllerf13a0962019-05-17 08:15:0698 RTC_DISALLOW_COPY_AND_ASSIGN(MessageWithFunctor);
99};
100
Henrik Boströmba4dcc32019-02-28 08:34:06101} // namespace rtc_thread_internal
102
Mirko Bonadei35214fc2019-09-23 12:54:28103class RTC_EXPORT ThreadManager {
Henrik Kjellanderec78f1c2017-06-29 05:52:50104 public:
105 static const int kForever = -1;
106
107 // Singleton, constructor and destructor are private.
108 static ThreadManager* Instance();
109
Sebastian Jansson6ea2c6a2020-01-13 13:07:22110 static void Add(Thread* message_queue);
111 static void Remove(Thread* message_queue);
112 static void Clear(MessageHandler* handler);
113
Sebastian Jansson6ea2c6a2020-01-13 13:07:22114 // For testing purposes, for use with a simulated clock.
115 // Ensures that all message queues have processed delayed messages
116 // up until the current point in time.
117 static void ProcessAllMessageQueuesForTesting();
118
Henrik Kjellanderec78f1c2017-06-29 05:52:50119 Thread* CurrentThread();
120 void SetCurrentThread(Thread* thread);
Sebastian Jansson178a6852020-01-14 10:12:26121 // Allows changing the current thread, this is intended for tests where we
122 // want to simulate multiple threads running on a single physical thread.
123 void ChangeCurrentThreadForTest(Thread* thread);
Henrik Kjellanderec78f1c2017-06-29 05:52:50124
125 // Returns a thread object with its thread_ ivar set
126 // to whatever the OS uses to represent the thread.
127 // If there already *is* a Thread object corresponding to this thread,
128 // this method will return that. Otherwise it creates a new Thread
129 // object whose wrapped() method will return true, and whose
130 // handle will, on Win32, be opened with only synchronization privileges -
131 // if you need more privilegs, rather than changing this method, please
132 // write additional code to adjust the privileges, or call a different
133 // factory method of your own devising, because this one gets used in
134 // unexpected contexts (like inside browser plugins) and it would be a
135 // shame to break it. It is also conceivable on Win32 that we won't even
136 // be able to get synchronization privileges, in which case the result
137 // will have a null handle.
Yves Gerey665174f2018-06-19 13:03:05138 Thread* WrapCurrentThread();
Henrik Kjellanderec78f1c2017-06-29 05:52:50139 void UnwrapCurrentThread();
140
Niels Moller9d1840c2019-05-21 07:26:37141 bool IsMainThread();
142
Sebastian Janssonda7267a2020-03-03 09:48:05143#if RTC_DCHECK_IS_ON
144 // Registers that a Send operation is to be performed between |source| and
145 // |target|, while checking that this does not cause a send cycle that could
146 // potentially cause a deadlock.
147 void RegisterSendAndCheckForCycles(Thread* source, Thread* target);
148#endif
149
Henrik Kjellanderec78f1c2017-06-29 05:52:50150 private:
151 ThreadManager();
152 ~ThreadManager();
153
Sebastian Jansson178a6852020-01-14 10:12:26154 void SetCurrentThreadInternal(Thread* thread);
Sebastian Jansson6ea2c6a2020-01-13 13:07:22155 void AddInternal(Thread* message_queue);
156 void RemoveInternal(Thread* message_queue);
157 void ClearInternal(MessageHandler* handler);
158 void ProcessAllMessageQueuesInternal();
Sebastian Janssonda7267a2020-03-03 09:48:05159#if RTC_DCHECK_IS_ON
160 void RemoveFromSendGraph(Thread* thread) RTC_EXCLUSIVE_LOCKS_REQUIRED(crit_);
161#endif
Sebastian Jansson6ea2c6a2020-01-13 13:07:22162
163 // This list contains all live Threads.
164 std::vector<Thread*> message_queues_ RTC_GUARDED_BY(crit_);
165
166 // Methods that don't modify the list of message queues may be called in a
167 // re-entrant fashion. "processing_" keeps track of the depth of re-entrant
168 // calls.
Markus Handell3cb525b2020-07-16 14:16:09169 RecursiveCriticalSection crit_;
Sebastian Jansson6ea2c6a2020-01-13 13:07:22170 size_t processing_ RTC_GUARDED_BY(crit_) = 0;
Sebastian Janssonda7267a2020-03-03 09:48:05171#if RTC_DCHECK_IS_ON
172 // Represents all thread seand actions by storing all send targets per thread.
173 // This is used by RegisterSendAndCheckForCycles. This graph has no cycles
174 // since we will trigger a CHECK failure if a cycle is introduced.
175 std::map<Thread*, std::set<Thread*>> send_graph_ RTC_GUARDED_BY(crit_);
176#endif
Sebastian Jansson6ea2c6a2020-01-13 13:07:22177
Henrik Kjellanderec78f1c2017-06-29 05:52:50178#if defined(WEBRTC_POSIX)
179 pthread_key_t key_;
180#endif
181
182#if defined(WEBRTC_WIN)
Tommi51492422017-12-04 14:18:23183 const DWORD key_;
Henrik Kjellanderec78f1c2017-06-29 05:52:50184#endif
185
Niels Moller9d1840c2019-05-21 07:26:37186 // The thread to potentially autowrap.
187 const PlatformThreadRef main_thread_ref_;
188
Henrik Kjellanderec78f1c2017-06-29 05:52:50189 RTC_DISALLOW_COPY_AND_ASSIGN(ThreadManager);
190};
191
Henrik Kjellanderec78f1c2017-06-29 05:52:50192// WARNING! SUBCLASSES MUST CALL Stop() IN THEIR DESTRUCTORS! See ~Thread().
193
Sebastian Jansson6ea2c6a2020-01-13 13:07:22194class RTC_LOCKABLE RTC_EXPORT Thread : public webrtc::TaskQueueBase {
tommia8a35152017-07-13 12:47:25195 public:
Sebastian Jansson6ea2c6a2020-01-13 13:07:22196 static const int kForever = -1;
197
198 // Create a new Thread and optionally assign it to the passed
199 // SocketServer. Subclasses that override Clear should pass false for
200 // init_queue and call DoInit() from their constructor to prevent races
201 // with the ThreadManager using the object while the vtable is still
202 // being created.
Henrik Kjellanderec78f1c2017-06-29 05:52:50203 explicit Thread(SocketServer* ss);
204 explicit Thread(std::unique_ptr<SocketServer> ss);
Sebastian Jansson6ea2c6a2020-01-13 13:07:22205
Taylor Brandstetter08672602018-03-02 23:20:33206 // Constructors meant for subclasses; they should call DoInit themselves and
207 // pass false for |do_init|, so that DoInit is called only on the fully
208 // instantiated class, which avoids a vptr data race.
209 Thread(SocketServer* ss, bool do_init);
210 Thread(std::unique_ptr<SocketServer> ss, bool do_init);
Henrik Kjellanderec78f1c2017-06-29 05:52:50211
212 // NOTE: ALL SUBCLASSES OF Thread MUST CALL Stop() IN THEIR DESTRUCTORS (or
213 // guarantee Stop() is explicitly called before the subclass is destroyed).
214 // This is required to avoid a data race between the destructor modifying the
215 // vtable, and the Thread::PreRun calling the virtual method Run().
Sebastian Jansson6ea2c6a2020-01-13 13:07:22216
217 // NOTE: SUBCLASSES OF Thread THAT OVERRIDE Clear MUST CALL
218 // DoDestroy() IN THEIR DESTRUCTORS! This is required to avoid a data race
219 // between the destructor modifying the vtable, and the ThreadManager
220 // calling Clear on the object from a different thread.
Henrik Kjellanderec78f1c2017-06-29 05:52:50221 ~Thread() override;
222
223 static std::unique_ptr<Thread> CreateWithSocketServer();
224 static std::unique_ptr<Thread> Create();
225 static Thread* Current();
226
227 // Used to catch performance regressions. Use this to disallow blocking calls
228 // (Invoke) for a given scope. If a synchronous call is made while this is in
229 // effect, an assert will be triggered.
230 // Note that this is a single threaded class.
231 class ScopedDisallowBlockingCalls {
232 public:
233 ScopedDisallowBlockingCalls();
Sebastian Jansson9debe5a2019-03-22 14:42:38234 ScopedDisallowBlockingCalls(const ScopedDisallowBlockingCalls&) = delete;
235 ScopedDisallowBlockingCalls& operator=(const ScopedDisallowBlockingCalls&) =
236 delete;
Henrik Kjellanderec78f1c2017-06-29 05:52:50237 ~ScopedDisallowBlockingCalls();
Yves Gerey665174f2018-06-19 13:03:05238
Henrik Kjellanderec78f1c2017-06-29 05:52:50239 private:
240 Thread* const thread_;
241 const bool previous_state_;
242 };
243
Tommife041642021-04-07 08:08:28244#if RTC_DCHECK_IS_ON
245 class ScopedCountBlockingCalls {
246 public:
247 ScopedCountBlockingCalls(std::function<void(uint32_t, uint32_t)> callback);
248 ScopedCountBlockingCalls(const ScopedDisallowBlockingCalls&) = delete;
249 ScopedCountBlockingCalls& operator=(const ScopedDisallowBlockingCalls&) =
250 delete;
251 ~ScopedCountBlockingCalls();
252
253 uint32_t GetBlockingCallCount() const;
254 uint32_t GetCouldBeBlockingCallCount() const;
255 uint32_t GetTotalBlockedCallCount() const;
256
Tomas Gunnarsson89f3dd52021-04-14 10:54:10257 void set_minimum_call_count_for_callback(uint32_t minimum) {
258 min_blocking_calls_for_callback_ = minimum;
259 }
260
Tommife041642021-04-07 08:08:28261 private:
262 Thread* const thread_;
263 const uint32_t base_blocking_call_count_;
264 const uint32_t base_could_be_blocking_call_count_;
Tomas Gunnarsson89f3dd52021-04-14 10:54:10265 // The minimum number of blocking calls required in order to issue the
266 // result_callback_. This is used by RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN to
267 // tame log spam.
268 // By default we always issue the callback, regardless of callback count.
269 uint32_t min_blocking_calls_for_callback_ = 0;
Tommife041642021-04-07 08:08:28270 std::function<void(uint32_t, uint32_t)> result_callback_;
271 };
272
273 uint32_t GetBlockingCallCount() const;
274 uint32_t GetCouldBeBlockingCallCount() const;
275#endif
276
Sebastian Jansson6ea2c6a2020-01-13 13:07:22277 SocketServer* socketserver();
278
279 // Note: The behavior of Thread has changed. When a thread is stopped,
280 // futher Posts and Sends will fail. However, any pending Sends and *ready*
281 // Posts (as opposed to unexpired delayed Posts) will be delivered before
282 // Get (or Peek) returns false. By guaranteeing delivery of those messages,
283 // we eliminate the race condition when an MessageHandler and Thread
284 // may be destroyed independently of each other.
285 virtual void Quit();
286 virtual bool IsQuitting();
287 virtual void Restart();
288 // Not all message queues actually process messages (such as SignalThread).
289 // In those cases, it's important to know, before posting, that it won't be
290 // Processed. Normally, this would be true until IsQuitting() is true.
291 virtual bool IsProcessingMessagesForTesting();
292
293 // Get() will process I/O until:
294 // 1) A message is available (returns true)
295 // 2) cmsWait seconds have elapsed (returns false)
296 // 3) Stop() is called (returns false)
297 virtual bool Get(Message* pmsg,
298 int cmsWait = kForever,
299 bool process_io = true);
300 virtual bool Peek(Message* pmsg, int cmsWait = 0);
Sebastian Jansson61380c02020-01-17 13:46:08301 // |time_sensitive| is deprecated and should always be false.
Sebastian Jansson6ea2c6a2020-01-13 13:07:22302 virtual void Post(const Location& posted_from,
303 MessageHandler* phandler,
304 uint32_t id = 0,
305 MessageData* pdata = nullptr,
306 bool time_sensitive = false);
307 virtual void PostDelayed(const Location& posted_from,
Sebastian Jansson61380c02020-01-17 13:46:08308 int delay_ms,
Sebastian Jansson6ea2c6a2020-01-13 13:07:22309 MessageHandler* phandler,
310 uint32_t id = 0,
311 MessageData* pdata = nullptr);
312 virtual void PostAt(const Location& posted_from,
Sebastian Jansson61380c02020-01-17 13:46:08313 int64_t run_at_ms,
Sebastian Jansson6ea2c6a2020-01-13 13:07:22314 MessageHandler* phandler,
315 uint32_t id = 0,
316 MessageData* pdata = nullptr);
317 virtual void Clear(MessageHandler* phandler,
318 uint32_t id = MQID_ANY,
319 MessageList* removed = nullptr);
320 virtual void Dispatch(Message* pmsg);
Sebastian Jansson6ea2c6a2020-01-13 13:07:22321
322 // Amount of time until the next message can be retrieved
323 virtual int GetDelay();
324
325 bool empty() const { return size() == 0u; }
326 size_t size() const {
Sebastian Jansson61380c02020-01-17 13:46:08327 CritScope cs(&crit_);
328 return messages_.size() + delayed_messages_.size() + (fPeekKeep_ ? 1u : 0u);
Sebastian Jansson6ea2c6a2020-01-13 13:07:22329 }
330
331 // Internally posts a message which causes the doomed object to be deleted
332 template <class T>
333 void Dispose(T* doomed) {
334 if (doomed) {
335 Post(RTC_FROM_HERE, nullptr, MQID_DISPOSE, new DisposeData<T>(doomed));
336 }
337 }
338
Henrik Kjellanderec78f1c2017-06-29 05:52:50339 bool IsCurrent() const;
340
341 // Sleeps the calling thread for the specified number of milliseconds, during
342 // which time no processing is performed. Returns false if sleeping was
343 // interrupted by a signal (POSIX only).
344 static bool SleepMs(int millis);
345
346 // Sets the thread's name, for debugging. Must be called before Start().
347 // If |obj| is non-null, its value is appended to |name|.
348 const std::string& name() const { return name_; }
349 bool SetName(const std::string& name, const void* obj);
350
Harald Alvestrandba694422021-01-27 21:52:14351 // Sets the expected processing time in ms. The thread will write
352 // log messages when Invoke() takes more time than this.
353 // Default is 50 ms.
354 void SetDispatchWarningMs(int deadline);
355
Henrik Kjellanderec78f1c2017-06-29 05:52:50356 // Starts the execution of the thread.
Niels Möllerd2e50132019-06-11 07:24:14357 bool Start();
Henrik Kjellanderec78f1c2017-06-29 05:52:50358
359 // Tells the thread to stop and waits until it is joined.
360 // Never call Stop on the current thread. Instead use the inherited Quit
Sebastian Jansson6ea2c6a2020-01-13 13:07:22361 // function which will exit the base Thread without terminating the
Henrik Kjellanderec78f1c2017-06-29 05:52:50362 // underlying OS thread.
363 virtual void Stop();
364
365 // By default, Thread::Run() calls ProcessMessages(kForever). To do other
366 // work, override Run(). To receive and dispatch messages, call
367 // ProcessMessages occasionally.
368 virtual void Run();
369
370 virtual void Send(const Location& posted_from,
371 MessageHandler* phandler,
372 uint32_t id = 0,
373 MessageData* pdata = nullptr);
374
375 // Convenience method to invoke a functor on another thread. Caller must
376 // provide the |ReturnT| template argument, which cannot (easily) be deduced.
377 // Uses Send() internally, which blocks the current thread until execution
378 // is complete.
379 // Ex: bool result = thread.Invoke<bool>(RTC_FROM_HERE,
380 // &MyFunctionReturningBool);
381 // NOTE: This function can only be called when synchronous calls are allowed.
382 // See ScopedDisallowBlockingCalls for details.
Henrik Boströmba4dcc32019-02-28 08:34:06383 // NOTE: Blocking invokes are DISCOURAGED, consider if what you're doing can
384 // be achieved with PostTask() and callbacks instead.
Danil Chapovalov89313452019-11-29 11:56:43385 template <
386 class ReturnT,
387 typename = typename std::enable_if<!std::is_void<ReturnT>::value>::type>
388 ReturnT Invoke(const Location& posted_from, FunctionView<ReturnT()> functor) {
389 ReturnT result;
390 InvokeInternal(posted_from, [functor, &result] { result = functor(); });
391 return result;
392 }
393
394 template <
395 class ReturnT,
396 typename = typename std::enable_if<std::is_void<ReturnT>::value>::type>
397 void Invoke(const Location& posted_from, FunctionView<void()> functor) {
398 InvokeInternal(posted_from, functor);
Henrik Kjellanderec78f1c2017-06-29 05:52:50399 }
400
Artem Titovdfc5f0d2020-07-03 10:09:26401 // Allows invoke to specified |thread|. Thread never will be dereferenced and
402 // will be used only for reference-based comparison, so instance can be safely
403 // deleted. If NDEBUG is defined and DCHECK_ALWAYS_ON is undefined do nothing.
404 void AllowInvokesToThread(Thread* thread);
Tomas Gunnarssonabdb4702020-09-05 16:43:36405
Artem Titovdfc5f0d2020-07-03 10:09:26406 // If NDEBUG is defined and DCHECK_ALWAYS_ON is undefined do nothing.
407 void DisallowAllInvokes();
408 // Returns true if |target| was allowed by AllowInvokesToThread() or if no
409 // calls were made to AllowInvokesToThread and DisallowAllInvokes. Otherwise
410 // returns false.
411 // If NDEBUG is defined and DCHECK_ALWAYS_ON is undefined always returns true.
412 bool IsInvokeToThreadAllowed(rtc::Thread* target);
413
Henrik Boströmba4dcc32019-02-28 08:34:06414 // Posts a task to invoke the functor on |this| thread asynchronously, i.e.
415 // without blocking the thread that invoked PostTask(). Ownership of |functor|
Niels Möllerf13a0962019-05-17 08:15:06416 // is passed and (usually, see below) destroyed on |this| thread after it is
417 // invoked.
Henrik Boströmba4dcc32019-02-28 08:34:06418 // Requirements of FunctorT:
419 // - FunctorT is movable.
420 // - FunctorT implements "T operator()()" or "T operator()() const" for some T
421 // (if T is not void, the return value is discarded on |this| thread).
422 // - FunctorT has a public destructor that can be invoked from |this| thread
423 // after operation() has been invoked.
424 // - The functor must not cause the thread to quit before PostTask() is done.
425 //
Niels Möllerf13a0962019-05-17 08:15:06426 // Destruction of the functor/task mimics what TaskQueue::PostTask does: If
427 // the task is run, it will be destroyed on |this| thread. However, if there
428 // are pending tasks by the time the Thread is destroyed, or a task is posted
429 // to a thread that is quitting, the task is destroyed immediately, on the
430 // calling thread. Destroying the Thread only blocks for any currently running
431 // task to complete. Note that TQ abstraction is even vaguer on how
432 // destruction happens in these cases, allowing destruction to happen
433 // asynchronously at a later time and on some arbitrary thread. So to ease
434 // migration, don't depend on Thread::PostTask destroying un-run tasks
435 // immediately.
436 //
Henrik Boströmba4dcc32019-02-28 08:34:06437 // Example - Calling a class method:
438 // class Foo {
439 // public:
440 // void DoTheThing();
441 // };
442 // Foo foo;
443 // thread->PostTask(RTC_FROM_HERE, Bind(&Foo::DoTheThing, &foo));
444 //
445 // Example - Calling a lambda function:
446 // thread->PostTask(RTC_FROM_HERE,
447 // [&x, &y] { x.TrackComputations(y.Compute()); });
448 template <class FunctorT>
449 void PostTask(const Location& posted_from, FunctorT&& functor) {
Steve Antonbcc1a762019-12-11 19:21:53450 Post(posted_from, GetPostTaskMessageHandler(), /*id=*/0,
Niels Möllerf13a0962019-05-17 08:15:06451 new rtc_thread_internal::MessageWithFunctor<FunctorT>(
Henrik Boströmba4dcc32019-02-28 08:34:06452 std::forward<FunctorT>(functor)));
Henrik Boströmba4dcc32019-02-28 08:34:06453 }
Steve Antonbcc1a762019-12-11 19:21:53454 template <class FunctorT>
455 void PostDelayedTask(const Location& posted_from,
456 FunctorT&& functor,
457 uint32_t milliseconds) {
458 PostDelayed(posted_from, milliseconds, GetPostTaskMessageHandler(),
459 /*id=*/0,
460 new rtc_thread_internal::MessageWithFunctor<FunctorT>(
461 std::forward<FunctorT>(functor)));
462 }
Henrik Boströmba4dcc32019-02-28 08:34:06463
Danil Chapovalov912b3b82019-11-22 14:52:40464 // From TaskQueueBase
465 void PostTask(std::unique_ptr<webrtc::QueuedTask> task) override;
466 void PostDelayedTask(std::unique_ptr<webrtc::QueuedTask> task,
467 uint32_t milliseconds) override;
468 void Delete() override;
469
Henrik Kjellanderec78f1c2017-06-29 05:52:50470 // ProcessMessages will process I/O and dispatch messages until:
471 // 1) cms milliseconds have elapsed (returns true)
472 // 2) Stop() is called (returns false)
473 bool ProcessMessages(int cms);
474
475 // Returns true if this is a thread that we created using the standard
476 // constructor, false if it was created by a call to
477 // ThreadManager::WrapCurrentThread(). The main thread of an application
478 // is generally not owned, since the OS representation of the thread
479 // obviously exists before we can get to it.
480 // You cannot call Start on non-owned threads.
481 bool IsOwned();
482
Tommi51492422017-12-04 14:18:23483 // Expose private method IsRunning() for tests.
Henrik Kjellanderec78f1c2017-06-29 05:52:50484 //
485 // DANGER: this is a terrible public API. Most callers that might want to
486 // call this likely do not have enough control/knowledge of the Thread in
487 // question to guarantee that the returned value remains true for the duration
488 // of whatever code is conditionally executing because of the return value!
Tommi51492422017-12-04 14:18:23489 bool RunningForTest() { return IsRunning(); }
Henrik Kjellanderec78f1c2017-06-29 05:52:50490
Henrik Kjellanderec78f1c2017-06-29 05:52:50491 // These functions are public to avoid injecting test hooks. Don't call them
492 // outside of tests.
493 // This method should be called when thread is created using non standard
494 // method, like derived implementation of rtc::Thread and it can not be
495 // started by calling Start(). This will set started flag to true and
496 // owned to false. This must be called from the current thread.
497 bool WrapCurrent();
498 void UnwrapCurrent();
499
Karl Wiberg32562252019-02-21 12:38:30500 // Sets the per-thread allow-blocking-calls flag to false; this is
501 // irrevocable. Must be called on this thread.
502 void DisallowBlockingCalls() { SetAllowBlockingCalls(false); }
503
Henrik Kjellanderec78f1c2017-06-29 05:52:50504 protected:
Sebastian Jansson6ce033a2020-01-22 09:12:56505 class CurrentThreadSetter : CurrentTaskQueueSetter {
506 public:
507 explicit CurrentThreadSetter(Thread* thread)
508 : CurrentTaskQueueSetter(thread),
509 manager_(rtc::ThreadManager::Instance()),
510 previous_(manager_->CurrentThread()) {
511 manager_->ChangeCurrentThreadForTest(thread);
512 }
513 ~CurrentThreadSetter() { manager_->ChangeCurrentThreadForTest(previous_); }
514
515 private:
516 rtc::ThreadManager* const manager_;
517 rtc::Thread* const previous_;
518 };
519
Sebastian Jansson61380c02020-01-17 13:46:08520 // DelayedMessage goes into a priority queue, sorted by trigger time. Messages
521 // with the same trigger time are processed in num_ (FIFO) order.
522 class DelayedMessage {
523 public:
524 DelayedMessage(int64_t delay,
525 int64_t run_time_ms,
526 uint32_t num,
527 const Message& msg)
528 : delay_ms_(delay),
529 run_time_ms_(run_time_ms),
530 message_number_(num),
531 msg_(msg) {}
532
533 bool operator<(const DelayedMessage& dmsg) const {
534 return (dmsg.run_time_ms_ < run_time_ms_) ||
535 ((dmsg.run_time_ms_ == run_time_ms_) &&
536 (dmsg.message_number_ < message_number_));
537 }
538
539 int64_t delay_ms_; // for debugging
540 int64_t run_time_ms_;
541 // Monotonicaly incrementing number used for ordering of messages
542 // targeted to execute at the same time.
543 uint32_t message_number_;
544 Message msg_;
545 };
546
Sebastian Jansson6ea2c6a2020-01-13 13:07:22547 class PriorityQueue : public std::priority_queue<DelayedMessage> {
548 public:
549 container_type& container() { return c; }
550 void reheap() { make_heap(c.begin(), c.end(), comp); }
551 };
552
553 void DoDelayPost(const Location& posted_from,
554 int64_t cmsDelay,
555 int64_t tstamp,
556 MessageHandler* phandler,
557 uint32_t id,
558 MessageData* pdata);
559
560 // Perform initialization, subclasses must call this from their constructor
561 // if false was passed as init_queue to the Thread constructor.
562 void DoInit();
563
564 // Does not take any lock. Must be called either while holding crit_, or by
565 // the destructor (by definition, the latter has exclusive access).
566 void ClearInternal(MessageHandler* phandler,
567 uint32_t id,
568 MessageList* removed) RTC_EXCLUSIVE_LOCKS_REQUIRED(&crit_);
569
570 // Perform cleanup; subclasses must call this from the destructor,
571 // and are not expected to actually hold the lock.
572 void DoDestroy() RTC_EXCLUSIVE_LOCKS_REQUIRED(&crit_);
573
574 void WakeUpSocketServer();
575
Henrik Kjellanderec78f1c2017-06-29 05:52:50576 // Same as WrapCurrent except that it never fails as it does not try to
577 // acquire the synchronization access of the thread. The caller should never
578 // call Stop() or Join() on this thread.
579 void SafeWrapCurrent();
580
581 // Blocks the calling thread until this thread has terminated.
582 void Join();
583
584 static void AssertBlockingIsAllowedOnCurrentThread();
585
586 friend class ScopedDisallowBlockingCalls;
587
Markus Handell3cb525b2020-07-16 14:16:09588 RecursiveCriticalSection* CritForTest() { return &crit_; }
Sebastian Jansson6ea2c6a2020-01-13 13:07:22589
Henrik Kjellanderec78f1c2017-06-29 05:52:50590 private:
Harald Alvestrandba694422021-01-27 21:52:14591 static const int kSlowDispatchLoggingThreshold = 50; // 50 ms
592
Danil Chapovalov912b3b82019-11-22 14:52:40593 class QueuedTaskHandler final : public MessageHandler {
594 public:
Tomas Gunnarsson77baeee2020-09-24 20:39:21595 QueuedTaskHandler() {}
Danil Chapovalov912b3b82019-11-22 14:52:40596 void OnMessage(Message* msg) override;
597 };
Steve Antonbcc1a762019-12-11 19:21:53598
Karl Wiberg32562252019-02-21 12:38:30599 // Sets the per-thread allow-blocking-calls flag and returns the previous
600 // value. Must be called on this thread.
601 bool SetAllowBlockingCalls(bool allow);
602
Henrik Kjellanderec78f1c2017-06-29 05:52:50603#if defined(WEBRTC_WIN)
604 static DWORD WINAPI PreRun(LPVOID context);
605#else
Yves Gerey665174f2018-06-19 13:03:05606 static void* PreRun(void* pv);
Henrik Kjellanderec78f1c2017-06-29 05:52:50607#endif
608
609 // ThreadManager calls this instead WrapCurrent() because
610 // ThreadManager::Instance() cannot be used while ThreadManager is
611 // being created.
612 // The method tries to get synchronization rights of the thread on Windows if
613 // |need_synchronize_access| is true.
614 bool WrapCurrentWithThreadManager(ThreadManager* thread_manager,
615 bool need_synchronize_access);
616
Tommi51492422017-12-04 14:18:23617 // Return true if the thread is currently running.
618 bool IsRunning();
Henrik Kjellanderec78f1c2017-06-29 05:52:50619
Danil Chapovalov89313452019-11-29 11:56:43620 void InvokeInternal(const Location& posted_from,
621 rtc::FunctionView<void()> functor);
Henrik Kjellanderec78f1c2017-06-29 05:52:50622
Tommi6866dc72020-05-15 08:11:56623 // Called by the ThreadManager when being set as the current thread.
624 void EnsureIsCurrentTaskQueue();
625
626 // Called by the ThreadManager when being unset as the current thread.
627 void ClearCurrentTaskQueue();
628
Steve Antonbcc1a762019-12-11 19:21:53629 // Returns a static-lifetime MessageHandler which runs message with
630 // MessageLikeTask payload data.
631 static MessageHandler* GetPostTaskMessageHandler();
632
Sebastian Jansson6ea2c6a2020-01-13 13:07:22633 bool fPeekKeep_;
634 Message msgPeek_;
Sebastian Jansson61380c02020-01-17 13:46:08635 MessageList messages_ RTC_GUARDED_BY(crit_);
636 PriorityQueue delayed_messages_ RTC_GUARDED_BY(crit_);
637 uint32_t delayed_next_num_ RTC_GUARDED_BY(crit_);
Tommife041642021-04-07 08:08:28638#if RTC_DCHECK_IS_ON
639 uint32_t blocking_call_count_ RTC_GUARDED_BY(this) = 0;
640 uint32_t could_be_blocking_call_count_ RTC_GUARDED_BY(this) = 0;
Artem Titovdfc5f0d2020-07-03 10:09:26641 std::vector<Thread*> allowed_threads_ RTC_GUARDED_BY(this);
642 bool invoke_policy_enabled_ RTC_GUARDED_BY(this) = false;
643#endif
Markus Handell3cb525b2020-07-16 14:16:09644 RecursiveCriticalSection crit_;
Sebastian Jansson6ea2c6a2020-01-13 13:07:22645 bool fInitialized_;
646 bool fDestroyed_;
647
648 volatile int stop_;
649
650 // The SocketServer might not be owned by Thread.
651 SocketServer* const ss_;
652 // Used if SocketServer ownership lies with |this|.
653 std::unique_ptr<SocketServer> own_ss_;
654
Henrik Kjellanderec78f1c2017-06-29 05:52:50655 std::string name_;
Tommi51492422017-12-04 14:18:23656
Jonas Olssona4d87372019-07-05 17:08:33657 // TODO(tommi): Add thread checks for proper use of control methods.
658 // Ideally we should be able to just use PlatformThread.
Henrik Kjellanderec78f1c2017-06-29 05:52:50659
660#if defined(WEBRTC_POSIX)
Tommi6cea2b02017-12-04 17:51:16661 pthread_t thread_ = 0;
Henrik Kjellanderec78f1c2017-06-29 05:52:50662#endif
663
664#if defined(WEBRTC_WIN)
Tommi6cea2b02017-12-04 17:51:16665 HANDLE thread_ = nullptr;
666 DWORD thread_id_ = 0;
Henrik Kjellanderec78f1c2017-06-29 05:52:50667#endif
668
Tommi51492422017-12-04 14:18:23669 // Indicates whether or not ownership of the worker thread lies with
670 // this instance or not. (i.e. owned_ == !wrapped).
671 // Must only be modified when the worker thread is not running.
672 bool owned_ = true;
673
674 // Only touched from the worker thread itself.
675 bool blocking_calls_allowed_ = true;
Henrik Kjellanderec78f1c2017-06-29 05:52:50676
Danil Chapovalov912b3b82019-11-22 14:52:40677 // Runs webrtc::QueuedTask posted to the Thread.
678 QueuedTaskHandler queued_task_handler_;
Tommi6866dc72020-05-15 08:11:56679 std::unique_ptr<TaskQueueBase::CurrentTaskQueueSetter>
680 task_queue_registration_;
Danil Chapovalov912b3b82019-11-22 14:52:40681
Henrik Kjellanderec78f1c2017-06-29 05:52:50682 friend class ThreadManager;
683
Harald Alvestrandba694422021-01-27 21:52:14684 int dispatch_warning_ms_ RTC_GUARDED_BY(this) = kSlowDispatchLoggingThreshold;
685
Henrik Kjellanderec78f1c2017-06-29 05:52:50686 RTC_DISALLOW_COPY_AND_ASSIGN(Thread);
687};
688
689// AutoThread automatically installs itself at construction
690// uninstalls at destruction, if a Thread object is
691// _not already_ associated with the current OS thread.
Tomas Gunnarsson0fd4c4e2020-09-04 14:33:25692//
693// NOTE: *** This class should only be used by tests ***
694//
Henrik Kjellanderec78f1c2017-06-29 05:52:50695class AutoThread : public Thread {
696 public:
697 AutoThread();
698 ~AutoThread() override;
699
700 private:
701 RTC_DISALLOW_COPY_AND_ASSIGN(AutoThread);
702};
703
704// AutoSocketServerThread automatically installs itself at
705// construction and uninstalls at destruction. If a Thread object is
706// already associated with the current OS thread, it is temporarily
707// disassociated and restored by the destructor.
708
709class AutoSocketServerThread : public Thread {
710 public:
711 explicit AutoSocketServerThread(SocketServer* ss);
712 ~AutoSocketServerThread() override;
713
714 private:
715 rtc::Thread* old_thread_;
716
717 RTC_DISALLOW_COPY_AND_ASSIGN(AutoSocketServerThread);
718};
Henrik Kjellanderec78f1c2017-06-29 05:52:50719} // namespace rtc
henrike@webrtc.orgf0488722014-05-13 18:00:26720
Mirko Bonadei92ea95e2017-09-15 04:47:31721#endif // RTC_BASE_THREAD_H_