blob: 38e9732fbbb07febb5b43d7a61b7f8db74bd0f44 [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"
Mirko Bonadei481e3452021-07-30 13:57:25 +020031#include "rtc_base/checks.h"
Steve Anton10542f22019-01-11 09:11:00 -080032#include "rtc_base/constructor_magic.h"
Markus Handell3cb525b2020-07-16 16:16:09 +020033#include "rtc_base/deprecated/recursive_critical_section.h"
Yves Gerey988cc082018-10-23 12:03:01 +020034#include "rtc_base/location.h"
Steve Anton10542f22019-01-11 09:11:00 -080035#include "rtc_base/message_handler.h"
Mirko Bonadei92ea95e2017-09-15 06:47:31 +020036#include "rtc_base/platform_thread_types.h"
Steve Anton10542f22019-01-11 09:11:00 -080037#include "rtc_base/socket_server.h"
Mirko Bonadei35214fc2019-09-23 14:54:28 +020038#include "rtc_base/system/rtc_export.h"
Yves Gerey988cc082018-10-23 12:03:01 +020039#include "rtc_base/thread_annotations.h"
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +010040#include "rtc_base/thread_message.h"
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020041
42#if defined(WEBRTC_WIN)
Mirko Bonadei92ea95e2017-09-15 06:47:31 +020043#include "rtc_base/win32.h"
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020044#endif
45
Tommife041642021-04-07 10:08:28 +020046#if RTC_DCHECK_IS_ON
47// Counts how many blocking Thread::Invoke or Thread::Send calls are made from
48// within a scope and logs the number of blocking calls at the end of the scope.
49#define RTC_LOG_THREAD_BLOCK_COUNT() \
50 rtc::Thread::ScopedCountBlockingCalls blocked_call_count_printer( \
51 [func = __func__](uint32_t actual_block, uint32_t could_block) { \
52 auto total = actual_block + could_block; \
53 if (total) { \
54 RTC_LOG(LS_WARNING) << "Blocking " << func << ": total=" << total \
55 << " (actual=" << actual_block \
56 << ", could=" << could_block << ")"; \
57 } \
58 })
59
60// Adds an RTC_DCHECK_LE that checks that the number of blocking calls are
61// less than or equal to a specific value. Use to avoid regressing in the
62// number of blocking thread calls.
63// Note: Use of this macro, requires RTC_LOG_THREAD_BLOCK_COUNT() to be called
64// first.
Tomas Gunnarsson89f3dd52021-04-14 12:54:10 +020065#define RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(x) \
66 do { \
67 blocked_call_count_printer.set_minimum_call_count_for_callback(x + 1); \
68 RTC_DCHECK_LE(blocked_call_count_printer.GetTotalBlockedCallCount(), x); \
69 } while (0)
Tommife041642021-04-07 10:08:28 +020070#else
71#define RTC_LOG_THREAD_BLOCK_COUNT()
72#define RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(x)
73#endif
74
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +020075namespace rtc {
76
77class Thread;
78
Henrik Boströmba4dcc32019-02-28 09:34:06 +010079namespace rtc_thread_internal {
80
Niels Möllerf13a0962019-05-17 10:15:06 +020081class MessageLikeTask : public MessageData {
Henrik Boströmba4dcc32019-02-28 09:34:06 +010082 public:
Niels Möllerf13a0962019-05-17 10:15:06 +020083 virtual void Run() = 0;
84};
85
86template <class FunctorT>
87class MessageWithFunctor final : public MessageLikeTask {
88 public:
89 explicit MessageWithFunctor(FunctorT&& functor)
Henrik Boströmba4dcc32019-02-28 09:34:06 +010090 : functor_(std::forward<FunctorT>(functor)) {}
91
Niels Möllerf13a0962019-05-17 10:15:06 +020092 void Run() override { functor_(); }
Henrik Boströmba4dcc32019-02-28 09:34:06 +010093
94 private:
Niels Möllerf13a0962019-05-17 10:15:06 +020095 ~MessageWithFunctor() override {}
Henrik Boströmba4dcc32019-02-28 09:34:06 +010096
97 typename std::remove_reference<FunctorT>::type functor_;
98
Niels Möllerf13a0962019-05-17 10:15:06 +020099 RTC_DISALLOW_COPY_AND_ASSIGN(MessageWithFunctor);
100};
101
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100102} // namespace rtc_thread_internal
103
Mirko Bonadei35214fc2019-09-23 14:54:28 +0200104class RTC_EXPORT ThreadManager {
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200105 public:
106 static const int kForever = -1;
107
108 // Singleton, constructor and destructor are private.
109 static ThreadManager* Instance();
110
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100111 static void Add(Thread* message_queue);
112 static void Remove(Thread* message_queue);
113 static void Clear(MessageHandler* handler);
114
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100115 // For testing purposes, for use with a simulated clock.
116 // Ensures that all message queues have processed delayed messages
117 // up until the current point in time.
118 static void ProcessAllMessageQueuesForTesting();
119
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200120 Thread* CurrentThread();
121 void SetCurrentThread(Thread* thread);
Sebastian Jansson178a6852020-01-14 11:12:26 +0100122 // Allows changing the current thread, this is intended for tests where we
123 // want to simulate multiple threads running on a single physical thread.
124 void ChangeCurrentThreadForTest(Thread* thread);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200125
126 // Returns a thread object with its thread_ ivar set
127 // to whatever the OS uses to represent the thread.
128 // If there already *is* a Thread object corresponding to this thread,
129 // this method will return that. Otherwise it creates a new Thread
130 // object whose wrapped() method will return true, and whose
131 // handle will, on Win32, be opened with only synchronization privileges -
132 // if you need more privilegs, rather than changing this method, please
133 // write additional code to adjust the privileges, or call a different
134 // factory method of your own devising, because this one gets used in
135 // unexpected contexts (like inside browser plugins) and it would be a
136 // shame to break it. It is also conceivable on Win32 that we won't even
137 // be able to get synchronization privileges, in which case the result
138 // will have a null handle.
Yves Gerey665174f2018-06-19 15:03:05 +0200139 Thread* WrapCurrentThread();
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200140 void UnwrapCurrentThread();
141
Niels Moller9d1840c2019-05-21 07:26:37 +0000142 bool IsMainThread();
143
Sebastian Janssonda7267a2020-03-03 10:48:05 +0100144#if RTC_DCHECK_IS_ON
Artem Titov96e3b992021-07-26 16:03:14 +0200145 // Registers that a Send operation is to be performed between `source` and
146 // `target`, while checking that this does not cause a send cycle that could
Sebastian Janssonda7267a2020-03-03 10:48:05 +0100147 // potentially cause a deadlock.
148 void RegisterSendAndCheckForCycles(Thread* source, Thread* target);
149#endif
150
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200151 private:
152 ThreadManager();
153 ~ThreadManager();
154
Sebastian Jansson178a6852020-01-14 11:12:26 +0100155 void SetCurrentThreadInternal(Thread* thread);
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100156 void AddInternal(Thread* message_queue);
157 void RemoveInternal(Thread* message_queue);
158 void ClearInternal(MessageHandler* handler);
159 void ProcessAllMessageQueuesInternal();
Sebastian Janssonda7267a2020-03-03 10:48:05 +0100160#if RTC_DCHECK_IS_ON
161 void RemoveFromSendGraph(Thread* thread) RTC_EXCLUSIVE_LOCKS_REQUIRED(crit_);
162#endif
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100163
164 // This list contains all live Threads.
165 std::vector<Thread*> message_queues_ RTC_GUARDED_BY(crit_);
166
167 // Methods that don't modify the list of message queues may be called in a
168 // re-entrant fashion. "processing_" keeps track of the depth of re-entrant
169 // calls.
Markus Handell3cb525b2020-07-16 16:16:09 +0200170 RecursiveCriticalSection crit_;
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100171 size_t processing_ RTC_GUARDED_BY(crit_) = 0;
Sebastian Janssonda7267a2020-03-03 10:48:05 +0100172#if RTC_DCHECK_IS_ON
173 // Represents all thread seand actions by storing all send targets per thread.
174 // This is used by RegisterSendAndCheckForCycles. This graph has no cycles
175 // since we will trigger a CHECK failure if a cycle is introduced.
176 std::map<Thread*, std::set<Thread*>> send_graph_ RTC_GUARDED_BY(crit_);
177#endif
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100178
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200179#if defined(WEBRTC_POSIX)
180 pthread_key_t key_;
181#endif
182
183#if defined(WEBRTC_WIN)
Tommi51492422017-12-04 15:18:23 +0100184 const DWORD key_;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200185#endif
186
Niels Moller9d1840c2019-05-21 07:26:37 +0000187 // The thread to potentially autowrap.
188 const PlatformThreadRef main_thread_ref_;
189
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200190 RTC_DISALLOW_COPY_AND_ASSIGN(ThreadManager);
191};
192
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200193// WARNING! SUBCLASSES MUST CALL Stop() IN THEIR DESTRUCTORS! See ~Thread().
194
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100195class RTC_LOCKABLE RTC_EXPORT Thread : public webrtc::TaskQueueBase {
tommia8a35152017-07-13 05:47:25 -0700196 public:
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100197 static const int kForever = -1;
198
199 // Create a new Thread and optionally assign it to the passed
200 // SocketServer. Subclasses that override Clear should pass false for
201 // init_queue and call DoInit() from their constructor to prevent races
202 // with the ThreadManager using the object while the vtable is still
203 // being created.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200204 explicit Thread(SocketServer* ss);
205 explicit Thread(std::unique_ptr<SocketServer> ss);
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100206
Taylor Brandstetter08672602018-03-02 15:20:33 -0800207 // Constructors meant for subclasses; they should call DoInit themselves and
Artem Titov96e3b992021-07-26 16:03:14 +0200208 // pass false for `do_init`, so that DoInit is called only on the fully
Taylor Brandstetter08672602018-03-02 15:20:33 -0800209 // instantiated class, which avoids a vptr data race.
210 Thread(SocketServer* ss, bool do_init);
211 Thread(std::unique_ptr<SocketServer> ss, bool do_init);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200212
213 // NOTE: ALL SUBCLASSES OF Thread MUST CALL Stop() IN THEIR DESTRUCTORS (or
214 // guarantee Stop() is explicitly called before the subclass is destroyed).
215 // This is required to avoid a data race between the destructor modifying the
216 // vtable, and the Thread::PreRun calling the virtual method Run().
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100217
218 // NOTE: SUBCLASSES OF Thread THAT OVERRIDE Clear MUST CALL
219 // DoDestroy() IN THEIR DESTRUCTORS! This is required to avoid a data race
220 // between the destructor modifying the vtable, and the ThreadManager
221 // calling Clear on the object from a different thread.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200222 ~Thread() override;
223
224 static std::unique_ptr<Thread> CreateWithSocketServer();
225 static std::unique_ptr<Thread> Create();
226 static Thread* Current();
227
228 // Used to catch performance regressions. Use this to disallow blocking calls
229 // (Invoke) for a given scope. If a synchronous call is made while this is in
230 // effect, an assert will be triggered.
231 // Note that this is a single threaded class.
232 class ScopedDisallowBlockingCalls {
233 public:
234 ScopedDisallowBlockingCalls();
Sebastian Jansson9debe5a2019-03-22 15:42:38 +0100235 ScopedDisallowBlockingCalls(const ScopedDisallowBlockingCalls&) = delete;
236 ScopedDisallowBlockingCalls& operator=(const ScopedDisallowBlockingCalls&) =
237 delete;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200238 ~ScopedDisallowBlockingCalls();
Yves Gerey665174f2018-06-19 15:03:05 +0200239
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200240 private:
241 Thread* const thread_;
242 const bool previous_state_;
243 };
244
Tommife041642021-04-07 10:08:28 +0200245#if RTC_DCHECK_IS_ON
246 class ScopedCountBlockingCalls {
247 public:
248 ScopedCountBlockingCalls(std::function<void(uint32_t, uint32_t)> callback);
249 ScopedCountBlockingCalls(const ScopedDisallowBlockingCalls&) = delete;
250 ScopedCountBlockingCalls& operator=(const ScopedDisallowBlockingCalls&) =
251 delete;
252 ~ScopedCountBlockingCalls();
253
254 uint32_t GetBlockingCallCount() const;
255 uint32_t GetCouldBeBlockingCallCount() const;
256 uint32_t GetTotalBlockedCallCount() const;
257
Tomas Gunnarsson89f3dd52021-04-14 12:54:10 +0200258 void set_minimum_call_count_for_callback(uint32_t minimum) {
259 min_blocking_calls_for_callback_ = minimum;
260 }
261
Tommife041642021-04-07 10:08:28 +0200262 private:
263 Thread* const thread_;
264 const uint32_t base_blocking_call_count_;
265 const uint32_t base_could_be_blocking_call_count_;
Tomas Gunnarsson89f3dd52021-04-14 12:54:10 +0200266 // The minimum number of blocking calls required in order to issue the
267 // result_callback_. This is used by RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN to
268 // tame log spam.
269 // By default we always issue the callback, regardless of callback count.
270 uint32_t min_blocking_calls_for_callback_ = 0;
Tommife041642021-04-07 10:08:28 +0200271 std::function<void(uint32_t, uint32_t)> result_callback_;
272 };
273
274 uint32_t GetBlockingCallCount() const;
275 uint32_t GetCouldBeBlockingCallCount() const;
276#endif
277
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100278 SocketServer* socketserver();
279
280 // Note: The behavior of Thread has changed. When a thread is stopped,
281 // futher Posts and Sends will fail. However, any pending Sends and *ready*
282 // Posts (as opposed to unexpired delayed Posts) will be delivered before
283 // Get (or Peek) returns false. By guaranteeing delivery of those messages,
284 // we eliminate the race condition when an MessageHandler and Thread
285 // may be destroyed independently of each other.
286 virtual void Quit();
287 virtual bool IsQuitting();
288 virtual void Restart();
289 // Not all message queues actually process messages (such as SignalThread).
290 // In those cases, it's important to know, before posting, that it won't be
291 // Processed. Normally, this would be true until IsQuitting() is true.
292 virtual bool IsProcessingMessagesForTesting();
293
294 // Get() will process I/O until:
295 // 1) A message is available (returns true)
296 // 2) cmsWait seconds have elapsed (returns false)
297 // 3) Stop() is called (returns false)
298 virtual bool Get(Message* pmsg,
299 int cmsWait = kForever,
300 bool process_io = true);
301 virtual bool Peek(Message* pmsg, int cmsWait = 0);
Artem Titov96e3b992021-07-26 16:03:14 +0200302 // `time_sensitive` is deprecated and should always be false.
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100303 virtual void Post(const Location& posted_from,
304 MessageHandler* phandler,
305 uint32_t id = 0,
306 MessageData* pdata = nullptr,
307 bool time_sensitive = false);
308 virtual void PostDelayed(const Location& posted_from,
Sebastian Jansson61380c02020-01-17 14:46:08 +0100309 int delay_ms,
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100310 MessageHandler* phandler,
311 uint32_t id = 0,
312 MessageData* pdata = nullptr);
313 virtual void PostAt(const Location& posted_from,
Sebastian Jansson61380c02020-01-17 14:46:08 +0100314 int64_t run_at_ms,
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100315 MessageHandler* phandler,
316 uint32_t id = 0,
317 MessageData* pdata = nullptr);
318 virtual void Clear(MessageHandler* phandler,
319 uint32_t id = MQID_ANY,
320 MessageList* removed = nullptr);
321 virtual void Dispatch(Message* pmsg);
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100322
323 // Amount of time until the next message can be retrieved
324 virtual int GetDelay();
325
326 bool empty() const { return size() == 0u; }
327 size_t size() const {
Sebastian Jansson61380c02020-01-17 14:46:08 +0100328 CritScope cs(&crit_);
329 return messages_.size() + delayed_messages_.size() + (fPeekKeep_ ? 1u : 0u);
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100330 }
331
332 // Internally posts a message which causes the doomed object to be deleted
333 template <class T>
334 void Dispose(T* doomed) {
335 if (doomed) {
336 Post(RTC_FROM_HERE, nullptr, MQID_DISPOSE, new DisposeData<T>(doomed));
337 }
338 }
339
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200340 bool IsCurrent() const;
341
342 // Sleeps the calling thread for the specified number of milliseconds, during
343 // which time no processing is performed. Returns false if sleeping was
344 // interrupted by a signal (POSIX only).
345 static bool SleepMs(int millis);
346
347 // Sets the thread's name, for debugging. Must be called before Start().
Artem Titov96e3b992021-07-26 16:03:14 +0200348 // If `obj` is non-null, its value is appended to `name`.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200349 const std::string& name() const { return name_; }
350 bool SetName(const std::string& name, const void* obj);
351
Harald Alvestrandba694422021-01-27 21:52:14 +0000352 // Sets the expected processing time in ms. The thread will write
353 // log messages when Invoke() takes more time than this.
354 // Default is 50 ms.
355 void SetDispatchWarningMs(int deadline);
356
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200357 // Starts the execution of the thread.
Niels Möllerd2e50132019-06-11 09:24:14 +0200358 bool Start();
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200359
360 // Tells the thread to stop and waits until it is joined.
361 // Never call Stop on the current thread. Instead use the inherited Quit
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100362 // function which will exit the base Thread without terminating the
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200363 // underlying OS thread.
364 virtual void Stop();
365
366 // By default, Thread::Run() calls ProcessMessages(kForever). To do other
367 // work, override Run(). To receive and dispatch messages, call
368 // ProcessMessages occasionally.
369 virtual void Run();
370
371 virtual void Send(const Location& posted_from,
372 MessageHandler* phandler,
373 uint32_t id = 0,
374 MessageData* pdata = nullptr);
375
376 // Convenience method to invoke a functor on another thread. Caller must
Artem Titov96e3b992021-07-26 16:03:14 +0200377 // provide the `ReturnT` template argument, which cannot (easily) be deduced.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200378 // Uses Send() internally, which blocks the current thread until execution
379 // is complete.
380 // Ex: bool result = thread.Invoke<bool>(RTC_FROM_HERE,
381 // &MyFunctionReturningBool);
382 // NOTE: This function can only be called when synchronous calls are allowed.
383 // See ScopedDisallowBlockingCalls for details.
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100384 // NOTE: Blocking invokes are DISCOURAGED, consider if what you're doing can
385 // be achieved with PostTask() and callbacks instead.
Danil Chapovalov89313452019-11-29 12:56:43 +0100386 template <
387 class ReturnT,
388 typename = typename std::enable_if<!std::is_void<ReturnT>::value>::type>
389 ReturnT Invoke(const Location& posted_from, FunctionView<ReturnT()> functor) {
390 ReturnT result;
391 InvokeInternal(posted_from, [functor, &result] { result = functor(); });
392 return result;
393 }
394
395 template <
396 class ReturnT,
397 typename = typename std::enable_if<std::is_void<ReturnT>::value>::type>
398 void Invoke(const Location& posted_from, FunctionView<void()> functor) {
399 InvokeInternal(posted_from, functor);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200400 }
401
Artem Titov96e3b992021-07-26 16:03:14 +0200402 // Allows invoke to specified `thread`. Thread never will be dereferenced and
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200403 // will be used only for reference-based comparison, so instance can be safely
Mirko Bonadei481e3452021-07-30 13:57:25 +0200404 // deleted. If NDEBUG is defined and RTC_DCHECK_IS_ON is undefined do
Mirko Bonadei8c185fc2021-07-21 13:12:38 +0200405 // nothing.
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200406 void AllowInvokesToThread(Thread* thread);
Tomas Gunnarssonabdb4702020-09-05 18:43:36 +0200407
Mirko Bonadei481e3452021-07-30 13:57:25 +0200408 // If NDEBUG is defined and RTC_DCHECK_IS_ON is undefined do nothing.
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200409 void DisallowAllInvokes();
Artem Titov96e3b992021-07-26 16:03:14 +0200410 // Returns true if `target` was allowed by AllowInvokesToThread() or if no
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200411 // calls were made to AllowInvokesToThread and DisallowAllInvokes. Otherwise
412 // returns false.
Mirko Bonadei481e3452021-07-30 13:57:25 +0200413 // If NDEBUG is defined and RTC_DCHECK_IS_ON is undefined always returns
Mirko Bonadei8c185fc2021-07-21 13:12:38 +0200414 // true.
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200415 bool IsInvokeToThreadAllowed(rtc::Thread* target);
416
Artem Titov96e3b992021-07-26 16:03:14 +0200417 // Posts a task to invoke the functor on `this` thread asynchronously, i.e.
418 // without blocking the thread that invoked PostTask(). Ownership of `functor`
419 // is passed and (usually, see below) destroyed on `this` thread after it is
Niels Möllerf13a0962019-05-17 10:15:06 +0200420 // invoked.
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100421 // Requirements of FunctorT:
422 // - FunctorT is movable.
423 // - FunctorT implements "T operator()()" or "T operator()() const" for some T
Artem Titov96e3b992021-07-26 16:03:14 +0200424 // (if T is not void, the return value is discarded on `this` thread).
425 // - FunctorT has a public destructor that can be invoked from `this` thread
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100426 // after operation() has been invoked.
427 // - The functor must not cause the thread to quit before PostTask() is done.
428 //
Niels Möllerf13a0962019-05-17 10:15:06 +0200429 // Destruction of the functor/task mimics what TaskQueue::PostTask does: If
Artem Titov96e3b992021-07-26 16:03:14 +0200430 // the task is run, it will be destroyed on `this` thread. However, if there
Niels Möllerf13a0962019-05-17 10:15:06 +0200431 // are pending tasks by the time the Thread is destroyed, or a task is posted
432 // to a thread that is quitting, the task is destroyed immediately, on the
433 // calling thread. Destroying the Thread only blocks for any currently running
434 // task to complete. Note that TQ abstraction is even vaguer on how
435 // destruction happens in these cases, allowing destruction to happen
436 // asynchronously at a later time and on some arbitrary thread. So to ease
437 // migration, don't depend on Thread::PostTask destroying un-run tasks
438 // immediately.
439 //
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100440 // Example - Calling a class method:
441 // class Foo {
442 // public:
443 // void DoTheThing();
444 // };
445 // Foo foo;
446 // thread->PostTask(RTC_FROM_HERE, Bind(&Foo::DoTheThing, &foo));
447 //
448 // Example - Calling a lambda function:
449 // thread->PostTask(RTC_FROM_HERE,
450 // [&x, &y] { x.TrackComputations(y.Compute()); });
451 template <class FunctorT>
452 void PostTask(const Location& posted_from, FunctorT&& functor) {
Steve Antonbcc1a762019-12-11 11:21:53 -0800453 Post(posted_from, GetPostTaskMessageHandler(), /*id=*/0,
Niels Möllerf13a0962019-05-17 10:15:06 +0200454 new rtc_thread_internal::MessageWithFunctor<FunctorT>(
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100455 std::forward<FunctorT>(functor)));
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100456 }
Steve Antonbcc1a762019-12-11 11:21:53 -0800457 template <class FunctorT>
458 void PostDelayedTask(const Location& posted_from,
459 FunctorT&& functor,
460 uint32_t milliseconds) {
461 PostDelayed(posted_from, milliseconds, GetPostTaskMessageHandler(),
462 /*id=*/0,
463 new rtc_thread_internal::MessageWithFunctor<FunctorT>(
464 std::forward<FunctorT>(functor)));
465 }
Henrik Boströmba4dcc32019-02-28 09:34:06 +0100466
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100467 // From TaskQueueBase
468 void PostTask(std::unique_ptr<webrtc::QueuedTask> task) override;
469 void PostDelayedTask(std::unique_ptr<webrtc::QueuedTask> task,
470 uint32_t milliseconds) override;
471 void Delete() override;
472
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200473 // ProcessMessages will process I/O and dispatch messages until:
474 // 1) cms milliseconds have elapsed (returns true)
475 // 2) Stop() is called (returns false)
476 bool ProcessMessages(int cms);
477
478 // Returns true if this is a thread that we created using the standard
479 // constructor, false if it was created by a call to
480 // ThreadManager::WrapCurrentThread(). The main thread of an application
481 // is generally not owned, since the OS representation of the thread
482 // obviously exists before we can get to it.
483 // You cannot call Start on non-owned threads.
484 bool IsOwned();
485
Tommi51492422017-12-04 15:18:23 +0100486 // Expose private method IsRunning() for tests.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200487 //
488 // DANGER: this is a terrible public API. Most callers that might want to
489 // call this likely do not have enough control/knowledge of the Thread in
490 // question to guarantee that the returned value remains true for the duration
491 // of whatever code is conditionally executing because of the return value!
Tommi51492422017-12-04 15:18:23 +0100492 bool RunningForTest() { return IsRunning(); }
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200493
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200494 // These functions are public to avoid injecting test hooks. Don't call them
495 // outside of tests.
496 // This method should be called when thread is created using non standard
497 // method, like derived implementation of rtc::Thread and it can not be
498 // started by calling Start(). This will set started flag to true and
499 // owned to false. This must be called from the current thread.
500 bool WrapCurrent();
501 void UnwrapCurrent();
502
Karl Wiberg32562252019-02-21 13:38:30 +0100503 // Sets the per-thread allow-blocking-calls flag to false; this is
504 // irrevocable. Must be called on this thread.
505 void DisallowBlockingCalls() { SetAllowBlockingCalls(false); }
506
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200507 protected:
Sebastian Jansson6ce033a2020-01-22 10:12:56 +0100508 class CurrentThreadSetter : CurrentTaskQueueSetter {
509 public:
510 explicit CurrentThreadSetter(Thread* thread)
511 : CurrentTaskQueueSetter(thread),
512 manager_(rtc::ThreadManager::Instance()),
513 previous_(manager_->CurrentThread()) {
514 manager_->ChangeCurrentThreadForTest(thread);
515 }
516 ~CurrentThreadSetter() { manager_->ChangeCurrentThreadForTest(previous_); }
517
518 private:
519 rtc::ThreadManager* const manager_;
520 rtc::Thread* const previous_;
521 };
522
Sebastian Jansson61380c02020-01-17 14:46:08 +0100523 // DelayedMessage goes into a priority queue, sorted by trigger time. Messages
524 // with the same trigger time are processed in num_ (FIFO) order.
525 class DelayedMessage {
526 public:
527 DelayedMessage(int64_t delay,
528 int64_t run_time_ms,
529 uint32_t num,
530 const Message& msg)
531 : delay_ms_(delay),
532 run_time_ms_(run_time_ms),
533 message_number_(num),
534 msg_(msg) {}
535
536 bool operator<(const DelayedMessage& dmsg) const {
537 return (dmsg.run_time_ms_ < run_time_ms_) ||
538 ((dmsg.run_time_ms_ == run_time_ms_) &&
539 (dmsg.message_number_ < message_number_));
540 }
541
542 int64_t delay_ms_; // for debugging
543 int64_t run_time_ms_;
544 // Monotonicaly incrementing number used for ordering of messages
545 // targeted to execute at the same time.
546 uint32_t message_number_;
547 Message msg_;
548 };
549
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100550 class PriorityQueue : public std::priority_queue<DelayedMessage> {
551 public:
552 container_type& container() { return c; }
553 void reheap() { make_heap(c.begin(), c.end(), comp); }
554 };
555
556 void DoDelayPost(const Location& posted_from,
557 int64_t cmsDelay,
558 int64_t tstamp,
559 MessageHandler* phandler,
560 uint32_t id,
561 MessageData* pdata);
562
563 // Perform initialization, subclasses must call this from their constructor
564 // if false was passed as init_queue to the Thread constructor.
565 void DoInit();
566
567 // Does not take any lock. Must be called either while holding crit_, or by
568 // the destructor (by definition, the latter has exclusive access).
569 void ClearInternal(MessageHandler* phandler,
570 uint32_t id,
571 MessageList* removed) RTC_EXCLUSIVE_LOCKS_REQUIRED(&crit_);
572
573 // Perform cleanup; subclasses must call this from the destructor,
574 // and are not expected to actually hold the lock.
575 void DoDestroy() RTC_EXCLUSIVE_LOCKS_REQUIRED(&crit_);
576
577 void WakeUpSocketServer();
578
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200579 // Same as WrapCurrent except that it never fails as it does not try to
580 // acquire the synchronization access of the thread. The caller should never
581 // call Stop() or Join() on this thread.
582 void SafeWrapCurrent();
583
584 // Blocks the calling thread until this thread has terminated.
585 void Join();
586
587 static void AssertBlockingIsAllowedOnCurrentThread();
588
589 friend class ScopedDisallowBlockingCalls;
590
Markus Handell3cb525b2020-07-16 16:16:09 +0200591 RecursiveCriticalSection* CritForTest() { return &crit_; }
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100592
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200593 private:
Harald Alvestrandba694422021-01-27 21:52:14 +0000594 static const int kSlowDispatchLoggingThreshold = 50; // 50 ms
595
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100596 class QueuedTaskHandler final : public MessageHandler {
597 public:
Tomas Gunnarsson77baeee2020-09-24 22:39:21 +0200598 QueuedTaskHandler() {}
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100599 void OnMessage(Message* msg) override;
600 };
Steve Antonbcc1a762019-12-11 11:21:53 -0800601
Karl Wiberg32562252019-02-21 13:38:30 +0100602 // Sets the per-thread allow-blocking-calls flag and returns the previous
603 // value. Must be called on this thread.
604 bool SetAllowBlockingCalls(bool allow);
605
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200606#if defined(WEBRTC_WIN)
607 static DWORD WINAPI PreRun(LPVOID context);
608#else
Yves Gerey665174f2018-06-19 15:03:05 +0200609 static void* PreRun(void* pv);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200610#endif
611
612 // ThreadManager calls this instead WrapCurrent() because
613 // ThreadManager::Instance() cannot be used while ThreadManager is
614 // being created.
615 // The method tries to get synchronization rights of the thread on Windows if
Artem Titov96e3b992021-07-26 16:03:14 +0200616 // `need_synchronize_access` is true.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200617 bool WrapCurrentWithThreadManager(ThreadManager* thread_manager,
618 bool need_synchronize_access);
619
Tommi51492422017-12-04 15:18:23 +0100620 // Return true if the thread is currently running.
621 bool IsRunning();
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200622
Danil Chapovalov89313452019-11-29 12:56:43 +0100623 void InvokeInternal(const Location& posted_from,
624 rtc::FunctionView<void()> functor);
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200625
Tommi6866dc72020-05-15 10:11:56 +0200626 // Called by the ThreadManager when being set as the current thread.
627 void EnsureIsCurrentTaskQueue();
628
629 // Called by the ThreadManager when being unset as the current thread.
630 void ClearCurrentTaskQueue();
631
Steve Antonbcc1a762019-12-11 11:21:53 -0800632 // Returns a static-lifetime MessageHandler which runs message with
633 // MessageLikeTask payload data.
634 static MessageHandler* GetPostTaskMessageHandler();
635
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100636 bool fPeekKeep_;
637 Message msgPeek_;
Sebastian Jansson61380c02020-01-17 14:46:08 +0100638 MessageList messages_ RTC_GUARDED_BY(crit_);
639 PriorityQueue delayed_messages_ RTC_GUARDED_BY(crit_);
640 uint32_t delayed_next_num_ RTC_GUARDED_BY(crit_);
Tommife041642021-04-07 10:08:28 +0200641#if RTC_DCHECK_IS_ON
642 uint32_t blocking_call_count_ RTC_GUARDED_BY(this) = 0;
643 uint32_t could_be_blocking_call_count_ RTC_GUARDED_BY(this) = 0;
Artem Titovdfc5f0d2020-07-03 12:09:26 +0200644 std::vector<Thread*> allowed_threads_ RTC_GUARDED_BY(this);
645 bool invoke_policy_enabled_ RTC_GUARDED_BY(this) = false;
646#endif
Markus Handell3cb525b2020-07-16 16:16:09 +0200647 RecursiveCriticalSection crit_;
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100648 bool fInitialized_;
649 bool fDestroyed_;
650
651 volatile int stop_;
652
653 // The SocketServer might not be owned by Thread.
654 SocketServer* const ss_;
Artem Titov96e3b992021-07-26 16:03:14 +0200655 // Used if SocketServer ownership lies with `this`.
Sebastian Jansson6ea2c6a2020-01-13 14:07:22 +0100656 std::unique_ptr<SocketServer> own_ss_;
657
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200658 std::string name_;
Tommi51492422017-12-04 15:18:23 +0100659
Jonas Olssona4d87372019-07-05 19:08:33 +0200660 // TODO(tommi): Add thread checks for proper use of control methods.
661 // Ideally we should be able to just use PlatformThread.
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200662
663#if defined(WEBRTC_POSIX)
Tommi6cea2b02017-12-04 18:51:16 +0100664 pthread_t thread_ = 0;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200665#endif
666
667#if defined(WEBRTC_WIN)
Tommi6cea2b02017-12-04 18:51:16 +0100668 HANDLE thread_ = nullptr;
669 DWORD thread_id_ = 0;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200670#endif
671
Tommi51492422017-12-04 15:18:23 +0100672 // Indicates whether or not ownership of the worker thread lies with
673 // this instance or not. (i.e. owned_ == !wrapped).
674 // Must only be modified when the worker thread is not running.
675 bool owned_ = true;
676
677 // Only touched from the worker thread itself.
678 bool blocking_calls_allowed_ = true;
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200679
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100680 // Runs webrtc::QueuedTask posted to the Thread.
681 QueuedTaskHandler queued_task_handler_;
Tommi6866dc72020-05-15 10:11:56 +0200682 std::unique_ptr<TaskQueueBase::CurrentTaskQueueSetter>
683 task_queue_registration_;
Danil Chapovalov912b3b82019-11-22 15:52:40 +0100684
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200685 friend class ThreadManager;
686
Harald Alvestrandba694422021-01-27 21:52:14 +0000687 int dispatch_warning_ms_ RTC_GUARDED_BY(this) = kSlowDispatchLoggingThreshold;
688
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200689 RTC_DISALLOW_COPY_AND_ASSIGN(Thread);
690};
691
692// AutoThread automatically installs itself at construction
693// uninstalls at destruction, if a Thread object is
694// _not already_ associated with the current OS thread.
Tomas Gunnarsson0fd4c4e2020-09-04 16:33:25 +0200695//
696// NOTE: *** This class should only be used by tests ***
697//
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200698class AutoThread : public Thread {
699 public:
700 AutoThread();
701 ~AutoThread() override;
702
703 private:
704 RTC_DISALLOW_COPY_AND_ASSIGN(AutoThread);
705};
706
707// AutoSocketServerThread automatically installs itself at
708// construction and uninstalls at destruction. If a Thread object is
709// already associated with the current OS thread, it is temporarily
710// disassociated and restored by the destructor.
711
712class AutoSocketServerThread : public Thread {
713 public:
714 explicit AutoSocketServerThread(SocketServer* ss);
715 ~AutoSocketServerThread() override;
716
717 private:
718 rtc::Thread* old_thread_;
719
720 RTC_DISALLOW_COPY_AND_ASSIGN(AutoSocketServerThread);
721};
Henrik Kjellanderec78f1c2017-06-29 07:52:50 +0200722} // namespace rtc
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000723
Mirko Bonadei92ea95e2017-09-15 06:47:31 +0200724#endif // RTC_BASE_THREAD_H_