blob: 6d3c39b8acfaf96e58f87cf46d71bb7e7046ba5d [file] [log] [blame]
henrike@webrtc.orgf0488722014-05-13 18:00:26 +00001/*
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 06:47:31 +020011#ifndef RTC_BASE_THREAD_H_
12#define RTC_BASE_THREAD_H_
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000013
Yves Gerey988cc082018-10-23 12:03:01 +020014#include <stdint.h>
Jonas Olssona4d87372019-07-05 19:08:33 +020015
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020016#include <list>
Sebastian Janssonda7267a2020-03-03 10:48:05 +010017#include <map>
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020018#include <memory>
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +010019#include <queue>
Sebastian Janssonda7267a2020-03-03 10:48:05 +010020#include <set>
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020021#include <string>
Yves Gerey988cc082018-10-23 12:03:01 +020022#include <type_traits>
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +010023#include <vector>
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000024
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020025#if defined(WEBRTC_POSIX)
26#include <pthread.h>
27#endif
Danil Chapovalov89313452019-11-29 12:56:43 +010028#include "api/function_view.h"
Danil Chapovalov912b3b82019-11-22 15:52:40 +010029#include "api/task_queue/queued_task.h"
30#include "api/task_queue/task_queue_base.h"
Steve Anton10542f22019-01-11 09:11:00 -080031#include "rtc_base/constructor_magic.h"
Markus Handell3cb525b2020-07-16 16:16:09 +020032#include "rtc_base/deprecated/recursive_critical_section.h"
Yves Gerey988cc082018-10-23 12:03:01 +020033#include "rtc_base/location.h"
Steve Anton10542f22019-01-11 09:11:00 -080034#include "rtc_base/message_handler.h"
Mirko Bonadei92ea95e2017-09-15 06:47:31 +020035#include "rtc_base/platform_thread_types.h"
Steve Anton10542f22019-01-11 09:11:00 -080036#include "rtc_base/socket_server.h"
Mirko Bonadei35214fc2019-09-23 14:54:28 +020037#include "rtc_base/system/rtc_export.h"
Yves Gerey988cc082018-10-23 12:03:01 +020038#include "rtc_base/thread_annotations.h"
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +010039#include "rtc_base/thread_message.h"
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020040
41#if defined(WEBRTC_WIN)
Mirko Bonadei92ea95e2017-09-15 06:47:31 +020042#include "rtc_base/win32.h"
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020043#endif
44
Tommife041642021-04-07 10:08:28 +020045#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.
64#define RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(x) \
65 RTC_DCHECK_LE(blocked_call_count_printer.GetTotalBlockedCallCount(), x)
66#else
67#define RTC_LOG_THREAD_BLOCK_COUNT()
68#define RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(x)
69#endif
70
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020071namespace rtc {
72
73class Thread;
74
Henrik Boströmba4dcc32019-02-28 09:34:06 +010075namespace rtc_thread_internal {
76
Niels Möllerf13a0962019-05-17 10:15:06 +020077class MessageLikeTask : public MessageData {
Henrik Boströmba4dcc32019-02-28 09:34:06 +010078 public:
Niels Möllerf13a0962019-05-17 10:15:06 +020079 virtual void Run() = 0;
80};
81
82template <class FunctorT>
83class MessageWithFunctor final : public MessageLikeTask {
84 public:
85 explicit MessageWithFunctor(FunctorT&& functor)
Henrik Boströmba4dcc32019-02-28 09:34:06 +010086 : functor_(std::forward<FunctorT>(functor)) {}
87
Niels Möllerf13a0962019-05-17 10:15:06 +020088 void Run() override { functor_(); }
Henrik Boströmba4dcc32019-02-28 09:34:06 +010089
90 private:
Niels Möllerf13a0962019-05-17 10:15:06 +020091 ~MessageWithFunctor() override {}
Henrik Boströmba4dcc32019-02-28 09:34:06 +010092
93 typename std::remove_reference<FunctorT>::type functor_;
94
Niels Möllerf13a0962019-05-17 10:15:06 +020095 RTC_DISALLOW_COPY_AND_ASSIGN(MessageWithFunctor);
96};
97
Henrik Boströmba4dcc32019-02-28 09:34:06 +010098} // namespace rtc_thread_internal
99
Mirko Bonadei35214fc2019-09-23 14:54:28 +0200100class RTC_EXPORT ThreadManager {
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200101 public:
102 static const int kForever = -1;
103
104 // Singleton, constructor and destructor are private.
105 static ThreadManager* Instance();
106
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100107 static void Add(Thread* message_queue);
108 static void Remove(Thread* message_queue);
109 static void Clear(MessageHandler* handler);
110
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100111 // For testing purposes, for use with a simulated clock.
112 // Ensures that all message queues have processed delayed messages
113 // up until the current point in time.
114 static void ProcessAllMessageQueuesForTesting();
115
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200116 Thread* CurrentThread();
117 void SetCurrentThread(Thread* thread);
Sebastian Jansson178a6852020-01-14 11:12:26 +0100118 // Allows changing the current thread, this is intended for tests where we
119 // want to simulate multiple threads running on a single physical thread.
120 void ChangeCurrentThreadForTest(Thread* thread);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200121
122 // Returns a thread object with its thread_ ivar set
123 // to whatever the OS uses to represent the thread.
124 // If there already *is* a Thread object corresponding to this thread,
125 // this method will return that. Otherwise it creates a new Thread
126 // object whose wrapped() method will return true, and whose
127 // handle will, on Win32, be opened with only synchronization privileges -
128 // if you need more privilegs, rather than changing this method, please
129 // write additional code to adjust the privileges, or call a different
130 // factory method of your own devising, because this one gets used in
131 // unexpected contexts (like inside browser plugins) and it would be a
132 // shame to break it. It is also conceivable on Win32 that we won't even
133 // be able to get synchronization privileges, in which case the result
134 // will have a null handle.
Yves Gerey665174f2018-06-19 15:03:05 +0200135 Thread* WrapCurrentThread();
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200136 void UnwrapCurrentThread();
137
Niels Moller9d1840c2019-05-21 07:26:37 +0000138 bool IsMainThread();
139
Sebastian Janssonda7267a2020-03-03 10:48:05 +0100140#if RTC_DCHECK_IS_ON
141 // Registers that a Send operation is to be performed between |source| and
142 // |target|, while checking that this does not cause a send cycle that could
143 // potentially cause a deadlock.
144 void RegisterSendAndCheckForCycles(Thread* source, Thread* target);
145#endif
146
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200147 private:
148 ThreadManager();
149 ~ThreadManager();
150
Sebastian Jansson178a6852020-01-14 11:12:26 +0100151 void SetCurrentThreadInternal(Thread* thread);
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100152 void AddInternal(Thread* message_queue);
153 void RemoveInternal(Thread* message_queue);
154 void ClearInternal(MessageHandler* handler);
155 void ProcessAllMessageQueuesInternal();
Sebastian Janssonda7267a2020-03-03 10:48:05 +0100156#if RTC_DCHECK_IS_ON
157 void RemoveFromSendGraph(Thread* thread) RTC_EXCLUSIVE_LOCKS_REQUIRED(crit_);
158#endif
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100159
160 // This list contains all live Threads.
161 std::vector<Thread*> message_queues_ RTC_GUARDED_BY(crit_);
162
163 // Methods that don't modify the list of message queues may be called in a
164 // re-entrant fashion. "processing_" keeps track of the depth of re-entrant
165 // calls.
Markus Handell3cb525b2020-07-16 16:16:09 +0200166 RecursiveCriticalSection crit_;
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100167 size_t processing_ RTC_GUARDED_BY(crit_) = 0;
Sebastian Janssonda7267a2020-03-03 10:48:05 +0100168#if RTC_DCHECK_IS_ON
169 // Represents all thread seand actions by storing all send targets per thread.
170 // This is used by RegisterSendAndCheckForCycles. This graph has no cycles
171 // since we will trigger a CHECK failure if a cycle is introduced.
172 std::map<Thread*, std::set<Thread*>> send_graph_ RTC_GUARDED_BY(crit_);
173#endif
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100174
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200175#if defined(WEBRTC_POSIX)
176 pthread_key_t key_;
177#endif
178
179#if defined(WEBRTC_WIN)
Tommi51492422017-12-04 15:18:23 +0100180 const DWORD key_;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200181#endif
182
Niels Moller9d1840c2019-05-21 07:26:37 +0000183 // The thread to potentially autowrap.
184 const PlatformThreadRef main_thread_ref_;
185
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200186 RTC_DISALLOW_COPY_AND_ASSIGN(ThreadManager);
187};
188
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200189// WARNING! SUBCLASSES MUST CALL Stop() IN THEIR DESTRUCTORS! See ~Thread().
190
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100191class RTC_LOCKABLE RTC_EXPORT Thread : public webrtc::TaskQueueBase {
tommia8a35152017-07-13 05:47:25 -0700192 public:
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100193 static const int kForever = -1;
194
195 // Create a new Thread and optionally assign it to the passed
196 // SocketServer. Subclasses that override Clear should pass false for
197 // init_queue and call DoInit() from their constructor to prevent races
198 // with the ThreadManager using the object while the vtable is still
199 // being created.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200200 explicit Thread(SocketServer* ss);
201 explicit Thread(std::unique_ptr<SocketServer> ss);
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100202
Taylor Brandstetter08672602018-03-02 15:20:33 -0800203 // Constructors meant for subclasses; they should call DoInit themselves and
204 // pass false for |do_init|, so that DoInit is called only on the fully
205 // instantiated class, which avoids a vptr data race.
206 Thread(SocketServer* ss, bool do_init);
207 Thread(std::unique_ptr<SocketServer> ss, bool do_init);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200208
209 // NOTE: ALL SUBCLASSES OF Thread MUST CALL Stop() IN THEIR DESTRUCTORS (or
210 // guarantee Stop() is explicitly called before the subclass is destroyed).
211 // This is required to avoid a data race between the destructor modifying the
212 // vtable, and the Thread::PreRun calling the virtual method Run().
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100213
214 // NOTE: SUBCLASSES OF Thread THAT OVERRIDE Clear MUST CALL
215 // DoDestroy() IN THEIR DESTRUCTORS! This is required to avoid a data race
216 // between the destructor modifying the vtable, and the ThreadManager
217 // calling Clear on the object from a different thread.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200218 ~Thread() override;
219
220 static std::unique_ptr<Thread> CreateWithSocketServer();
221 static std::unique_ptr<Thread> Create();
222 static Thread* Current();
223
224 // Used to catch performance regressions. Use this to disallow blocking calls
225 // (Invoke) for a given scope. If a synchronous call is made while this is in
226 // effect, an assert will be triggered.
227 // Note that this is a single threaded class.
228 class ScopedDisallowBlockingCalls {
229 public:
230 ScopedDisallowBlockingCalls();
Sebastian Jansson9debe5a2019-03-22 15:42:38 +0100231 ScopedDisallowBlockingCalls(const ScopedDisallowBlockingCalls&) = delete;
232 ScopedDisallowBlockingCalls& operator=(const ScopedDisallowBlockingCalls&) =
233 delete;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200234 ~ScopedDisallowBlockingCalls();
Yves Gerey665174f2018-06-19 15:03:05 +0200235
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200236 private:
237 Thread* const thread_;
238 const bool previous_state_;
239 };
240
Tommife041642021-04-07 10:08:28 +0200241#if RTC_DCHECK_IS_ON
242 class ScopedCountBlockingCalls {
243 public:
244 ScopedCountBlockingCalls(std::function<void(uint32_t, uint32_t)> callback);
245 ScopedCountBlockingCalls(const ScopedDisallowBlockingCalls&) = delete;
246 ScopedCountBlockingCalls& operator=(const ScopedDisallowBlockingCalls&) =
247 delete;
248 ~ScopedCountBlockingCalls();
249
250 uint32_t GetBlockingCallCount() const;
251 uint32_t GetCouldBeBlockingCallCount() const;
252 uint32_t GetTotalBlockedCallCount() const;
253
254 private:
255 Thread* const thread_;
256 const uint32_t base_blocking_call_count_;
257 const uint32_t base_could_be_blocking_call_count_;
258 std::function<void(uint32_t, uint32_t)> result_callback_;
259 };
260
261 uint32_t GetBlockingCallCount() const;
262 uint32_t GetCouldBeBlockingCallCount() const;
263#endif
264
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100265 SocketServer* socketserver();
266
267 // Note: The behavior of Thread has changed. When a thread is stopped,
268 // futher Posts and Sends will fail. However, any pending Sends and *ready*
269 // Posts (as opposed to unexpired delayed Posts) will be delivered before
270 // Get (or Peek) returns false. By guaranteeing delivery of those messages,
271 // we eliminate the race condition when an MessageHandler and Thread
272 // may be destroyed independently of each other.
273 virtual void Quit();
274 virtual bool IsQuitting();
275 virtual void Restart();
276 // Not all message queues actually process messages (such as SignalThread).
277 // In those cases, it's important to know, before posting, that it won't be
278 // Processed. Normally, this would be true until IsQuitting() is true.
279 virtual bool IsProcessingMessagesForTesting();
280
281 // Get() will process I/O until:
282 // 1) A message is available (returns true)
283 // 2) cmsWait seconds have elapsed (returns false)
284 // 3) Stop() is called (returns false)
285 virtual bool Get(Message* pmsg,
286 int cmsWait = kForever,
287 bool process_io = true);
288 virtual bool Peek(Message* pmsg, int cmsWait = 0);
Sebastian Jansson61380c02020-01-17 14:46:08 +0100289 // |time_sensitive| is deprecated and should always be false.
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100290 virtual void Post(const Location& posted_from,
291 MessageHandler* phandler,
292 uint32_t id = 0,
293 MessageData* pdata = nullptr,
294 bool time_sensitive = false);
295 virtual void PostDelayed(const Location& posted_from,
Sebastian Jansson61380c02020-01-17 14:46:08 +0100296 int delay_ms,
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100297 MessageHandler* phandler,
298 uint32_t id = 0,
299 MessageData* pdata = nullptr);
300 virtual void PostAt(const Location& posted_from,
Sebastian Jansson61380c02020-01-17 14:46:08 +0100301 int64_t run_at_ms,
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100302 MessageHandler* phandler,
303 uint32_t id = 0,
304 MessageData* pdata = nullptr);
305 virtual void Clear(MessageHandler* phandler,
306 uint32_t id = MQID_ANY,
307 MessageList* removed = nullptr);
308 virtual void Dispatch(Message* pmsg);
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100309
310 // Amount of time until the next message can be retrieved
311 virtual int GetDelay();
312
313 bool empty() const { return size() == 0u; }
314 size_t size() const {
Sebastian Jansson61380c02020-01-17 14:46:08 +0100315 CritScope cs(&crit_);
316 return messages_.size() + delayed_messages_.size() + (fPeekKeep_ ? 1u : 0u);
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100317 }
318
319 // Internally posts a message which causes the doomed object to be deleted
320 template <class T>
321 void Dispose(T* doomed) {
322 if (doomed) {
323 Post(RTC_FROM_HERE, nullptr, MQID_DISPOSE, new DisposeData<T>(doomed));
324 }
325 }
326
327 // When this signal is sent out, any references to this queue should
328 // no longer be used.
329 sigslot::signal0<> SignalQueueDestroyed;
330
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200331 bool IsCurrent() const;
332
333 // Sleeps the calling thread for the specified number of milliseconds, during
334 // which time no processing is performed. Returns false if sleeping was
335 // interrupted by a signal (POSIX only).
336 static bool SleepMs(int millis);
337
338 // Sets the thread's name, for debugging. Must be called before Start().
339 // If |obj| is non-null, its value is appended to |name|.
340 const std::string& name() const { return name_; }
341 bool SetName(const std::string& name, const void* obj);
342
Harald Alvestrandba694422021-01-27 21:52:14 +0000343 // Sets the expected processing time in ms. The thread will write
344 // log messages when Invoke() takes more time than this.
345 // Default is 50 ms.
346 void SetDispatchWarningMs(int deadline);
347
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200348 // Starts the execution of the thread.
Niels Möllerd2e50132019-06-11 09:24:14 +0200349 bool Start();
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200350
351 // Tells the thread to stop and waits until it is joined.
352 // Never call Stop on the current thread. Instead use the inherited Quit
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100353 // function which will exit the base Thread without terminating the
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200354 // underlying OS thread.
355 virtual void Stop();
356
357 // By default, Thread::Run() calls ProcessMessages(kForever). To do other
358 // work, override Run(). To receive and dispatch messages, call
359 // ProcessMessages occasionally.
360 virtual void Run();
361
362 virtual void Send(const Location& posted_from,
363 MessageHandler* phandler,
364 uint32_t id = 0,
365 MessageData* pdata = nullptr);
366
367 // Convenience method to invoke a functor on another thread. Caller must
368 // provide the |ReturnT| template argument, which cannot (easily) be deduced.
369 // Uses Send() internally, which blocks the current thread until execution
370 // is complete.
371 // Ex: bool result = thread.Invoke<bool>(RTC_FROM_HERE,
372 // &MyFunctionReturningBool);
373 // NOTE: This function can only be called when synchronous calls are allowed.
374 // See ScopedDisallowBlockingCalls for details.
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100375 // NOTE: Blocking invokes are DISCOURAGED, consider if what you're doing can
376 // be achieved with PostTask() and callbacks instead.
Danil Chapovalov89313452019-11-29 12:56:43 +0100377 template <
378 class ReturnT,
379 typename = typename std::enable_if<!std::is_void<ReturnT>::value>::type>
380 ReturnT Invoke(const Location& posted_from, FunctionView<ReturnT()> functor) {
381 ReturnT result;
382 InvokeInternal(posted_from, [functor, &result] { result = functor(); });
383 return result;
384 }
385
386 template <
387 class ReturnT,
388 typename = typename std::enable_if<std::is_void<ReturnT>::value>::type>
389 void Invoke(const Location& posted_from, FunctionView<void()> functor) {
390 InvokeInternal(posted_from, functor);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200391 }
392
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200393 // Allows invoke to specified |thread|. Thread never will be dereferenced and
394 // will be used only for reference-based comparison, so instance can be safely
395 // deleted. If NDEBUG is defined and DCHECK_ALWAYS_ON is undefined do nothing.
396 void AllowInvokesToThread(Thread* thread);
Tomas Gunnarssonabdb4702020-09-05 18:43:36 +0200397
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200398 // If NDEBUG is defined and DCHECK_ALWAYS_ON is undefined do nothing.
399 void DisallowAllInvokes();
400 // Returns true if |target| was allowed by AllowInvokesToThread() or if no
401 // calls were made to AllowInvokesToThread and DisallowAllInvokes. Otherwise
402 // returns false.
403 // If NDEBUG is defined and DCHECK_ALWAYS_ON is undefined always returns true.
404 bool IsInvokeToThreadAllowed(rtc::Thread* target);
405
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100406 // Posts a task to invoke the functor on |this| thread asynchronously, i.e.
407 // without blocking the thread that invoked PostTask(). Ownership of |functor|
Niels Möllerf13a0962019-05-17 10:15:06 +0200408 // is passed and (usually, see below) destroyed on |this| thread after it is
409 // invoked.
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100410 // Requirements of FunctorT:
411 // - FunctorT is movable.
412 // - FunctorT implements "T operator()()" or "T operator()() const" for some T
413 // (if T is not void, the return value is discarded on |this| thread).
414 // - FunctorT has a public destructor that can be invoked from |this| thread
415 // after operation() has been invoked.
416 // - The functor must not cause the thread to quit before PostTask() is done.
417 //
Niels Möllerf13a0962019-05-17 10:15:06 +0200418 // Destruction of the functor/task mimics what TaskQueue::PostTask does: If
419 // the task is run, it will be destroyed on |this| thread. However, if there
420 // are pending tasks by the time the Thread is destroyed, or a task is posted
421 // to a thread that is quitting, the task is destroyed immediately, on the
422 // calling thread. Destroying the Thread only blocks for any currently running
423 // task to complete. Note that TQ abstraction is even vaguer on how
424 // destruction happens in these cases, allowing destruction to happen
425 // asynchronously at a later time and on some arbitrary thread. So to ease
426 // migration, don't depend on Thread::PostTask destroying un-run tasks
427 // immediately.
428 //
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100429 // Example - Calling a class method:
430 // class Foo {
431 // public:
432 // void DoTheThing();
433 // };
434 // Foo foo;
435 // thread->PostTask(RTC_FROM_HERE, Bind(&Foo::DoTheThing, &foo));
436 //
437 // Example - Calling a lambda function:
438 // thread->PostTask(RTC_FROM_HERE,
439 // [&x, &y] { x.TrackComputations(y.Compute()); });
440 template <class FunctorT>
441 void PostTask(const Location& posted_from, FunctorT&& functor) {
Steve Antonbcc1a762019-12-11 11:21:53 -0800442 Post(posted_from, GetPostTaskMessageHandler(), /*id=*/0,
Niels Möllerf13a0962019-05-17 10:15:06 +0200443 new rtc_thread_internal::MessageWithFunctor<FunctorT>(
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100444 std::forward<FunctorT>(functor)));
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100445 }
Steve Antonbcc1a762019-12-11 11:21:53 -0800446 template <class FunctorT>
447 void PostDelayedTask(const Location& posted_from,
448 FunctorT&& functor,
449 uint32_t milliseconds) {
450 PostDelayed(posted_from, milliseconds, GetPostTaskMessageHandler(),
451 /*id=*/0,
452 new rtc_thread_internal::MessageWithFunctor<FunctorT>(
453 std::forward<FunctorT>(functor)));
454 }
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100455
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100456 // From TaskQueueBase
457 void PostTask(std::unique_ptr<webrtc::QueuedTask> task) override;
458 void PostDelayedTask(std::unique_ptr<webrtc::QueuedTask> task,
459 uint32_t milliseconds) override;
460 void Delete() override;
461
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200462 // ProcessMessages will process I/O and dispatch messages until:
463 // 1) cms milliseconds have elapsed (returns true)
464 // 2) Stop() is called (returns false)
465 bool ProcessMessages(int cms);
466
467 // Returns true if this is a thread that we created using the standard
468 // constructor, false if it was created by a call to
469 // ThreadManager::WrapCurrentThread(). The main thread of an application
470 // is generally not owned, since the OS representation of the thread
471 // obviously exists before we can get to it.
472 // You cannot call Start on non-owned threads.
473 bool IsOwned();
474
Tommi51492422017-12-04 15:18:23 +0100475 // Expose private method IsRunning() for tests.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200476 //
477 // DANGER: this is a terrible public API. Most callers that might want to
478 // call this likely do not have enough control/knowledge of the Thread in
479 // question to guarantee that the returned value remains true for the duration
480 // of whatever code is conditionally executing because of the return value!
Tommi51492422017-12-04 15:18:23 +0100481 bool RunningForTest() { return IsRunning(); }
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200482
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200483 // These functions are public to avoid injecting test hooks. Don't call them
484 // outside of tests.
485 // This method should be called when thread is created using non standard
486 // method, like derived implementation of rtc::Thread and it can not be
487 // started by calling Start(). This will set started flag to true and
488 // owned to false. This must be called from the current thread.
489 bool WrapCurrent();
490 void UnwrapCurrent();
491
Karl Wiberg32562252019-02-21 13:38:30 +0100492 // Sets the per-thread allow-blocking-calls flag to false; this is
493 // irrevocable. Must be called on this thread.
494 void DisallowBlockingCalls() { SetAllowBlockingCalls(false); }
495
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200496 protected:
Sebastian Jansson6ce033a2020-01-22 10:12:56 +0100497 class CurrentThreadSetter : CurrentTaskQueueSetter {
498 public:
499 explicit CurrentThreadSetter(Thread* thread)
500 : CurrentTaskQueueSetter(thread),
501 manager_(rtc::ThreadManager::Instance()),
502 previous_(manager_->CurrentThread()) {
503 manager_->ChangeCurrentThreadForTest(thread);
504 }
505 ~CurrentThreadSetter() { manager_->ChangeCurrentThreadForTest(previous_); }
506
507 private:
508 rtc::ThreadManager* const manager_;
509 rtc::Thread* const previous_;
510 };
511
Sebastian Jansson61380c02020-01-17 14:46:08 +0100512 // DelayedMessage goes into a priority queue, sorted by trigger time. Messages
513 // with the same trigger time are processed in num_ (FIFO) order.
514 class DelayedMessage {
515 public:
516 DelayedMessage(int64_t delay,
517 int64_t run_time_ms,
518 uint32_t num,
519 const Message& msg)
520 : delay_ms_(delay),
521 run_time_ms_(run_time_ms),
522 message_number_(num),
523 msg_(msg) {}
524
525 bool operator<(const DelayedMessage& dmsg) const {
526 return (dmsg.run_time_ms_ < run_time_ms_) ||
527 ((dmsg.run_time_ms_ == run_time_ms_) &&
528 (dmsg.message_number_ < message_number_));
529 }
530
531 int64_t delay_ms_; // for debugging
532 int64_t run_time_ms_;
533 // Monotonicaly incrementing number used for ordering of messages
534 // targeted to execute at the same time.
535 uint32_t message_number_;
536 Message msg_;
537 };
538
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100539 class PriorityQueue : public std::priority_queue<DelayedMessage> {
540 public:
541 container_type& container() { return c; }
542 void reheap() { make_heap(c.begin(), c.end(), comp); }
543 };
544
545 void DoDelayPost(const Location& posted_from,
546 int64_t cmsDelay,
547 int64_t tstamp,
548 MessageHandler* phandler,
549 uint32_t id,
550 MessageData* pdata);
551
552 // Perform initialization, subclasses must call this from their constructor
553 // if false was passed as init_queue to the Thread constructor.
554 void DoInit();
555
556 // Does not take any lock. Must be called either while holding crit_, or by
557 // the destructor (by definition, the latter has exclusive access).
558 void ClearInternal(MessageHandler* phandler,
559 uint32_t id,
560 MessageList* removed) RTC_EXCLUSIVE_LOCKS_REQUIRED(&crit_);
561
562 // Perform cleanup; subclasses must call this from the destructor,
563 // and are not expected to actually hold the lock.
564 void DoDestroy() RTC_EXCLUSIVE_LOCKS_REQUIRED(&crit_);
565
566 void WakeUpSocketServer();
567
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200568 // Same as WrapCurrent except that it never fails as it does not try to
569 // acquire the synchronization access of the thread. The caller should never
570 // call Stop() or Join() on this thread.
571 void SafeWrapCurrent();
572
573 // Blocks the calling thread until this thread has terminated.
574 void Join();
575
576 static void AssertBlockingIsAllowedOnCurrentThread();
577
578 friend class ScopedDisallowBlockingCalls;
579
Markus Handell3cb525b2020-07-16 16:16:09 +0200580 RecursiveCriticalSection* CritForTest() { return &crit_; }
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100581
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200582 private:
Harald Alvestrandba694422021-01-27 21:52:14 +0000583 static const int kSlowDispatchLoggingThreshold = 50; // 50 ms
584
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100585 class QueuedTaskHandler final : public MessageHandler {
586 public:
Tomas Gunnarsson77baeee2020-09-24 22:39:21 +0200587 QueuedTaskHandler() {}
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100588 void OnMessage(Message* msg) override;
589 };
Steve Antonbcc1a762019-12-11 11:21:53 -0800590
Karl Wiberg32562252019-02-21 13:38:30 +0100591 // Sets the per-thread allow-blocking-calls flag and returns the previous
592 // value. Must be called on this thread.
593 bool SetAllowBlockingCalls(bool allow);
594
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200595#if defined(WEBRTC_WIN)
596 static DWORD WINAPI PreRun(LPVOID context);
597#else
Yves Gerey665174f2018-06-19 15:03:05 +0200598 static void* PreRun(void* pv);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200599#endif
600
601 // ThreadManager calls this instead WrapCurrent() because
602 // ThreadManager::Instance() cannot be used while ThreadManager is
603 // being created.
604 // The method tries to get synchronization rights of the thread on Windows if
605 // |need_synchronize_access| is true.
606 bool WrapCurrentWithThreadManager(ThreadManager* thread_manager,
607 bool need_synchronize_access);
608
Tommi51492422017-12-04 15:18:23 +0100609 // Return true if the thread is currently running.
610 bool IsRunning();
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200611
Danil Chapovalov89313452019-11-29 12:56:43 +0100612 void InvokeInternal(const Location& posted_from,
613 rtc::FunctionView<void()> functor);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200614
Tommi6866dc72020-05-15 10:11:56 +0200615 // Called by the ThreadManager when being set as the current thread.
616 void EnsureIsCurrentTaskQueue();
617
618 // Called by the ThreadManager when being unset as the current thread.
619 void ClearCurrentTaskQueue();
620
Steve Antonbcc1a762019-12-11 11:21:53 -0800621 // Returns a static-lifetime MessageHandler which runs message with
622 // MessageLikeTask payload data.
623 static MessageHandler* GetPostTaskMessageHandler();
624
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100625 bool fPeekKeep_;
626 Message msgPeek_;
Sebastian Jansson61380c02020-01-17 14:46:08 +0100627 MessageList messages_ RTC_GUARDED_BY(crit_);
628 PriorityQueue delayed_messages_ RTC_GUARDED_BY(crit_);
629 uint32_t delayed_next_num_ RTC_GUARDED_BY(crit_);
Tommife041642021-04-07 10:08:28 +0200630#if RTC_DCHECK_IS_ON
631 uint32_t blocking_call_count_ RTC_GUARDED_BY(this) = 0;
632 uint32_t could_be_blocking_call_count_ RTC_GUARDED_BY(this) = 0;
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200633 std::vector<Thread*> allowed_threads_ RTC_GUARDED_BY(this);
634 bool invoke_policy_enabled_ RTC_GUARDED_BY(this) = false;
635#endif
Markus Handell3cb525b2020-07-16 16:16:09 +0200636 RecursiveCriticalSection crit_;
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100637 bool fInitialized_;
638 bool fDestroyed_;
639
640 volatile int stop_;
641
642 // The SocketServer might not be owned by Thread.
643 SocketServer* const ss_;
644 // Used if SocketServer ownership lies with |this|.
645 std::unique_ptr<SocketServer> own_ss_;
646
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200647 std::string name_;
Tommi51492422017-12-04 15:18:23 +0100648
Jonas Olssona4d87372019-07-05 19:08:33 +0200649 // TODO(tommi): Add thread checks for proper use of control methods.
650 // Ideally we should be able to just use PlatformThread.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200651
652#if defined(WEBRTC_POSIX)
Tommi6cea2b02017-12-04 18:51:16 +0100653 pthread_t thread_ = 0;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200654#endif
655
656#if defined(WEBRTC_WIN)
Tommi6cea2b02017-12-04 18:51:16 +0100657 HANDLE thread_ = nullptr;
658 DWORD thread_id_ = 0;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200659#endif
660
Tommi51492422017-12-04 15:18:23 +0100661 // Indicates whether or not ownership of the worker thread lies with
662 // this instance or not. (i.e. owned_ == !wrapped).
663 // Must only be modified when the worker thread is not running.
664 bool owned_ = true;
665
666 // Only touched from the worker thread itself.
667 bool blocking_calls_allowed_ = true;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200668
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100669 // Runs webrtc::QueuedTask posted to the Thread.
670 QueuedTaskHandler queued_task_handler_;
Tommi6866dc72020-05-15 10:11:56 +0200671 std::unique_ptr<TaskQueueBase::CurrentTaskQueueSetter>
672 task_queue_registration_;
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100673
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200674 friend class ThreadManager;
675
Harald Alvestrandba694422021-01-27 21:52:14 +0000676 int dispatch_warning_ms_ RTC_GUARDED_BY(this) = kSlowDispatchLoggingThreshold;
677
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200678 RTC_DISALLOW_COPY_AND_ASSIGN(Thread);
679};
680
681// AutoThread automatically installs itself at construction
682// uninstalls at destruction, if a Thread object is
683// _not already_ associated with the current OS thread.
Tomas Gunnarsson0fd4c4e2020-09-04 16:33:25 +0200684//
685// NOTE: *** This class should only be used by tests ***
686//
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200687class AutoThread : public Thread {
688 public:
689 AutoThread();
690 ~AutoThread() override;
691
692 private:
693 RTC_DISALLOW_COPY_AND_ASSIGN(AutoThread);
694};
695
696// AutoSocketServerThread automatically installs itself at
697// construction and uninstalls at destruction. If a Thread object is
698// already associated with the current OS thread, it is temporarily
699// disassociated and restored by the destructor.
700
701class AutoSocketServerThread : public Thread {
702 public:
703 explicit AutoSocketServerThread(SocketServer* ss);
704 ~AutoSocketServerThread() override;
705
706 private:
707 rtc::Thread* old_thread_;
708
709 RTC_DISALLOW_COPY_AND_ASSIGN(AutoSocketServerThread);
710};
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200711} // namespace rtc
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000712
Mirko Bonadei92ea95e2017-09-15 06:47:31 +0200713#endif // RTC_BASE_THREAD_H_