thread.cc 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436
  1. // Copyright (c) 2012 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "base/threading/thread.h"
  5. #include <memory>
  6. #include <type_traits>
  7. #include <utility>
  8. #include "base/bind.h"
  9. #include "base/callback_helpers.h"
  10. #include "base/lazy_instance.h"
  11. #include "base/location.h"
  12. #include "base/logging.h"
  13. #include "base/memory/ptr_util.h"
  14. #include "base/memory/scoped_refptr.h"
  15. #include "base/message_loop/message_pump.h"
  16. #include "base/run_loop.h"
  17. #include "base/synchronization/waitable_event.h"
  18. #include "base/task/current_thread.h"
  19. #include "base/task/sequence_manager/sequence_manager_impl.h"
  20. #include "base/task/sequence_manager/task_queue.h"
  21. #include "base/task/simple_task_executor.h"
  22. #include "base/third_party/dynamic_annotations/dynamic_annotations.h"
  23. #include "base/threading/thread_id_name_manager.h"
  24. #include "base/threading/thread_local.h"
  25. #include "base/threading/thread_restrictions.h"
  26. #include "base/threading/thread_task_runner_handle.h"
  27. #include "build/build_config.h"
  28. #if BUILDFLAG(IS_POSIX) && !BUILDFLAG(IS_NACL)
  29. #include "base/files/file_descriptor_watcher_posix.h"
  30. #include "third_party/abseil-cpp/absl/types/optional.h"
  31. #endif
  32. #if BUILDFLAG(IS_WIN)
  33. #include "base/win/scoped_com_initializer.h"
  34. #endif
  35. namespace base {
  36. namespace {
  37. // We use this thread-local variable to record whether or not a thread exited
  38. // because its Stop method was called. This allows us to catch cases where
  39. // MessageLoop::QuitWhenIdle() is called directly, which is unexpected when
  40. // using a Thread to setup and run a MessageLoop.
  41. base::LazyInstance<base::ThreadLocalBoolean>::Leaky lazy_tls_bool =
  42. LAZY_INSTANCE_INITIALIZER;
  43. class SequenceManagerThreadDelegate : public Thread::Delegate {
  44. public:
  45. explicit SequenceManagerThreadDelegate(
  46. MessagePumpType message_pump_type,
  47. OnceCallback<std::unique_ptr<MessagePump>()> message_pump_factory)
  48. : sequence_manager_(
  49. sequence_manager::internal::SequenceManagerImpl::CreateUnbound(
  50. sequence_manager::SequenceManager::Settings::Builder()
  51. .SetMessagePumpType(message_pump_type)
  52. .Build())),
  53. default_task_queue_(sequence_manager_->CreateTaskQueue(
  54. sequence_manager::TaskQueue::Spec("default_tq"))),
  55. message_pump_factory_(std::move(message_pump_factory)) {
  56. sequence_manager_->SetDefaultTaskRunner(default_task_queue_->task_runner());
  57. }
  58. ~SequenceManagerThreadDelegate() override = default;
  59. scoped_refptr<SingleThreadTaskRunner> GetDefaultTaskRunner() override {
  60. // Surprisingly this might not be default_task_queue_->task_runner() which
  61. // we set in the constructor. The Thread::Init() method could create a
  62. // SequenceManager on top of the current one and call
  63. // SequenceManager::SetDefaultTaskRunner which would propagate the new
  64. // TaskRunner down to our SequenceManager. Turns out, code actually relies
  65. // on this and somehow relies on
  66. // SequenceManagerThreadDelegate::GetDefaultTaskRunner returning this new
  67. // TaskRunner. So instead of returning default_task_queue_->task_runner() we
  68. // need to query the SequenceManager for it.
  69. // The underlying problem here is that Subclasses of Thread can do crazy
  70. // stuff in Init() but they are not really in control of what happens in the
  71. // Thread::Delegate, as this is passed in on calling StartWithOptions which
  72. // could happen far away from where the Thread is created. We should
  73. // consider getting rid of StartWithOptions, and pass them as a constructor
  74. // argument instead.
  75. return sequence_manager_->GetTaskRunner();
  76. }
  77. void BindToCurrentThread(TimerSlack timer_slack) override {
  78. sequence_manager_->BindToMessagePump(
  79. std::move(message_pump_factory_).Run());
  80. sequence_manager_->SetTimerSlack(timer_slack);
  81. simple_task_executor_.emplace(GetDefaultTaskRunner());
  82. }
  83. private:
  84. std::unique_ptr<sequence_manager::internal::SequenceManagerImpl>
  85. sequence_manager_;
  86. scoped_refptr<sequence_manager::TaskQueue> default_task_queue_;
  87. OnceCallback<std::unique_ptr<MessagePump>()> message_pump_factory_;
  88. absl::optional<SimpleTaskExecutor> simple_task_executor_;
  89. };
  90. } // namespace
  91. Thread::Options::Options() = default;
  92. Thread::Options::Options(MessagePumpType type, size_t size)
  93. : message_pump_type(type), stack_size(size) {}
  94. Thread::Options::Options(ThreadType thread_type) : thread_type(thread_type) {}
  95. Thread::Options::Options(Options&& other)
  96. : message_pump_type(std::move(other.message_pump_type)),
  97. delegate(std::move(other.delegate)),
  98. timer_slack(std::move(other.timer_slack)),
  99. message_pump_factory(std::move(other.message_pump_factory)),
  100. stack_size(std::move(other.stack_size)),
  101. thread_type(std::move(other.thread_type)),
  102. joinable(std::move(other.joinable)) {
  103. other.moved_from = true;
  104. }
  105. Thread::Options& Thread::Options::operator=(Thread::Options&& other) {
  106. DCHECK_NE(this, &other);
  107. message_pump_type = std::move(other.message_pump_type);
  108. delegate = std::move(other.delegate);
  109. timer_slack = std::move(other.timer_slack);
  110. message_pump_factory = std::move(other.message_pump_factory);
  111. stack_size = std::move(other.stack_size);
  112. thread_type = std::move(other.thread_type);
  113. joinable = std::move(other.joinable);
  114. other.moved_from = true;
  115. return *this;
  116. }
  117. Thread::Options::~Options() = default;
  118. Thread::Thread(const std::string& name)
  119. : id_event_(WaitableEvent::ResetPolicy::MANUAL,
  120. WaitableEvent::InitialState::NOT_SIGNALED),
  121. name_(name),
  122. start_event_(WaitableEvent::ResetPolicy::MANUAL,
  123. WaitableEvent::InitialState::NOT_SIGNALED) {
  124. // Only bind the sequence on Start(): the state is constant between
  125. // construction and Start() and it's thus valid for Start() to be called on
  126. // another sequence as long as every other operation is then performed on that
  127. // sequence.
  128. owning_sequence_checker_.DetachFromSequence();
  129. }
  130. Thread::~Thread() {
  131. Stop();
  132. }
  133. bool Thread::Start() {
  134. DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  135. Options options;
  136. #if BUILDFLAG(IS_WIN)
  137. if (com_status_ == STA)
  138. options.message_pump_type = MessagePumpType::UI;
  139. #endif
  140. return StartWithOptions(std::move(options));
  141. }
  142. bool Thread::StartWithOptions(Options options) {
  143. DCHECK(options.IsValid());
  144. DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  145. DCHECK(!delegate_);
  146. DCHECK(!IsRunning());
  147. DCHECK(!stopping_) << "Starting a non-joinable thread a second time? That's "
  148. << "not allowed!";
  149. #if BUILDFLAG(IS_WIN)
  150. DCHECK((com_status_ != STA) ||
  151. (options.message_pump_type == MessagePumpType::UI));
  152. #endif
  153. // Reset |id_| here to support restarting the thread.
  154. id_event_.Reset();
  155. id_ = kInvalidThreadId;
  156. SetThreadWasQuitProperly(false);
  157. timer_slack_ = options.timer_slack;
  158. if (options.delegate) {
  159. DCHECK(!options.message_pump_factory);
  160. delegate_ = std::move(options.delegate);
  161. } else if (options.message_pump_factory) {
  162. delegate_ = std::make_unique<SequenceManagerThreadDelegate>(
  163. MessagePumpType::CUSTOM, options.message_pump_factory);
  164. } else {
  165. delegate_ = std::make_unique<SequenceManagerThreadDelegate>(
  166. options.message_pump_type,
  167. BindOnce([](MessagePumpType type) { return MessagePump::Create(type); },
  168. options.message_pump_type));
  169. }
  170. start_event_.Reset();
  171. // Hold |thread_lock_| while starting the new thread to synchronize with
  172. // Stop() while it's not guaranteed to be sequenced (until crbug/629139 is
  173. // fixed).
  174. {
  175. AutoLock lock(thread_lock_);
  176. bool success = options.joinable
  177. ? PlatformThread::CreateWithType(
  178. options.stack_size, this, &thread_,
  179. options.thread_type, options.message_pump_type)
  180. : PlatformThread::CreateNonJoinableWithType(
  181. options.stack_size, this, options.thread_type,
  182. options.message_pump_type);
  183. if (!success) {
  184. DLOG(ERROR) << "failed to create thread";
  185. return false;
  186. }
  187. }
  188. joinable_ = options.joinable;
  189. return true;
  190. }
  191. bool Thread::StartAndWaitForTesting() {
  192. DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  193. bool result = Start();
  194. if (!result)
  195. return false;
  196. WaitUntilThreadStarted();
  197. return true;
  198. }
  199. bool Thread::WaitUntilThreadStarted() const {
  200. DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  201. if (!delegate_)
  202. return false;
  203. // https://crbug.com/918039
  204. base::ScopedAllowBaseSyncPrimitivesOutsideBlockingScope allow_wait;
  205. start_event_.Wait();
  206. return true;
  207. }
  208. void Thread::FlushForTesting() {
  209. DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  210. if (!delegate_)
  211. return;
  212. WaitableEvent done(WaitableEvent::ResetPolicy::AUTOMATIC,
  213. WaitableEvent::InitialState::NOT_SIGNALED);
  214. task_runner()->PostTask(FROM_HERE,
  215. BindOnce(&WaitableEvent::Signal, Unretained(&done)));
  216. done.Wait();
  217. }
  218. void Thread::Stop() {
  219. DCHECK(joinable_);
  220. // TODO(gab): Fix improper usage of this API (http://crbug.com/629139) and
  221. // enable this check, until then synchronization with Start() via
  222. // |thread_lock_| is required...
  223. // DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  224. AutoLock lock(thread_lock_);
  225. StopSoon();
  226. // Can't join if the |thread_| is either already gone or is non-joinable.
  227. if (thread_.is_null())
  228. return;
  229. // Wait for the thread to exit.
  230. //
  231. // TODO(darin): Unfortunately, we need to keep |delegate_| around
  232. // until the thread exits. Some consumers are abusing the API. Make them stop.
  233. PlatformThread::Join(thread_);
  234. thread_ = base::PlatformThreadHandle();
  235. // The thread should release |delegate_| on exit (note: Join() adds
  236. // an implicit memory barrier and no lock is thus required for this check).
  237. DCHECK(!delegate_);
  238. stopping_ = false;
  239. }
  240. void Thread::StopSoon() {
  241. // TODO(gab): Fix improper usage of this API (http://crbug.com/629139) and
  242. // enable this check.
  243. // DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  244. if (stopping_ || !delegate_)
  245. return;
  246. stopping_ = true;
  247. task_runner()->PostTask(
  248. FROM_HERE, base::BindOnce(&Thread::ThreadQuitHelper, Unretained(this)));
  249. }
  250. void Thread::DetachFromSequence() {
  251. DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  252. owning_sequence_checker_.DetachFromSequence();
  253. }
  254. PlatformThreadId Thread::GetThreadId() const {
  255. if (!id_event_.IsSignaled()) {
  256. // If the thread is created but not started yet, wait for |id_| being ready.
  257. base::ScopedAllowBaseSyncPrimitivesOutsideBlockingScope allow_wait;
  258. id_event_.Wait();
  259. }
  260. return id_;
  261. }
  262. bool Thread::IsRunning() const {
  263. // TODO(gab): Fix improper usage of this API (http://crbug.com/629139) and
  264. // enable this check.
  265. // DCHECK(owning_sequence_checker_.CalledOnValidSequence());
  266. // If the thread's already started (i.e. |delegate_| is non-null) and
  267. // not yet requested to stop (i.e. |stopping_| is false) we can just return
  268. // true. (Note that |stopping_| is touched only on the same sequence that
  269. // starts / started the new thread so we need no locking here.)
  270. if (delegate_ && !stopping_)
  271. return true;
  272. // Otherwise check the |running_| flag, which is set to true by the new thread
  273. // only while it is inside Run().
  274. AutoLock lock(running_lock_);
  275. return running_;
  276. }
  277. void Thread::Run(RunLoop* run_loop) {
  278. // Overridable protected method to be called from our |thread_| only.
  279. DCHECK(id_event_.IsSignaled());
  280. DCHECK_EQ(id_, PlatformThread::CurrentId());
  281. run_loop->Run();
  282. }
  283. // static
  284. void Thread::SetThreadWasQuitProperly(bool flag) {
  285. lazy_tls_bool.Pointer()->Set(flag);
  286. }
  287. // static
  288. bool Thread::GetThreadWasQuitProperly() {
  289. bool quit_properly = true;
  290. #if DCHECK_IS_ON()
  291. quit_properly = lazy_tls_bool.Pointer()->Get();
  292. #endif
  293. return quit_properly;
  294. }
  295. void Thread::ThreadMain() {
  296. // First, make GetThreadId() available to avoid deadlocks. It could be called
  297. // any place in the following thread initialization code.
  298. DCHECK(!id_event_.IsSignaled());
  299. // Note: this read of |id_| while |id_event_| isn't signaled is exceptionally
  300. // okay because ThreadMain has a happens-after relationship with the other
  301. // write in StartWithOptions().
  302. DCHECK_EQ(kInvalidThreadId, id_);
  303. id_ = PlatformThread::CurrentId();
  304. DCHECK_NE(kInvalidThreadId, id_);
  305. id_event_.Signal();
  306. // Complete the initialization of our Thread object.
  307. PlatformThread::SetName(name_.c_str());
  308. ANNOTATE_THREAD_NAME(name_.c_str()); // Tell the name to race detector.
  309. // Lazily initialize the |message_loop| so that it can run on this thread.
  310. DCHECK(delegate_);
  311. // This binds CurrentThread and ThreadTaskRunnerHandle.
  312. delegate_->BindToCurrentThread(timer_slack_);
  313. DCHECK(CurrentThread::Get());
  314. DCHECK(ThreadTaskRunnerHandle::IsSet());
  315. #if BUILDFLAG(IS_POSIX) && !BUILDFLAG(IS_NACL)
  316. // Allow threads running a MessageLoopForIO to use FileDescriptorWatcher API.
  317. std::unique_ptr<FileDescriptorWatcher> file_descriptor_watcher;
  318. if (CurrentIOThread::IsSet()) {
  319. file_descriptor_watcher = std::make_unique<FileDescriptorWatcher>(
  320. delegate_->GetDefaultTaskRunner());
  321. }
  322. #endif
  323. #if BUILDFLAG(IS_WIN)
  324. std::unique_ptr<win::ScopedCOMInitializer> com_initializer;
  325. if (com_status_ != NONE) {
  326. com_initializer.reset(
  327. (com_status_ == STA)
  328. ? new win::ScopedCOMInitializer()
  329. : new win::ScopedCOMInitializer(win::ScopedCOMInitializer::kMTA));
  330. }
  331. #endif
  332. // Let the thread do extra initialization.
  333. Init();
  334. {
  335. AutoLock lock(running_lock_);
  336. running_ = true;
  337. }
  338. start_event_.Signal();
  339. RunLoop run_loop;
  340. run_loop_ = &run_loop;
  341. Run(run_loop_);
  342. {
  343. AutoLock lock(running_lock_);
  344. running_ = false;
  345. }
  346. // Let the thread do extra cleanup.
  347. CleanUp();
  348. #if BUILDFLAG(IS_WIN)
  349. com_initializer.reset();
  350. #endif
  351. DCHECK(GetThreadWasQuitProperly());
  352. // We can't receive messages anymore.
  353. // (The message loop is destructed at the end of this block)
  354. delegate_.reset();
  355. run_loop_ = nullptr;
  356. }
  357. void Thread::ThreadQuitHelper() {
  358. DCHECK(run_loop_);
  359. run_loop_->QuitWhenIdle();
  360. SetThreadWasQuitProperly(true);
  361. }
  362. } // namespace base