file_descriptor_watcher_posix.cc 10.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283
  1. // Copyright 2016 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/files/file_descriptor_watcher_posix.h"
  5. #include <utility>
  6. #include "base/bind.h"
  7. #include "base/callback_helpers.h"
  8. #include "base/memory/ptr_util.h"
  9. #include "base/message_loop/message_pump_for_io.h"
  10. #include "base/no_destructor.h"
  11. #include "base/synchronization/waitable_event.h"
  12. #include "base/task/current_thread.h"
  13. #include "base/task/sequenced_task_runner.h"
  14. #include "base/task/single_thread_task_runner.h"
  15. #include "base/threading/sequenced_task_runner_handle.h"
  16. #include "base/threading/thread_checker.h"
  17. #include "base/threading/thread_local.h"
  18. #include "base/threading/thread_restrictions.h"
  19. namespace base {
  20. namespace {
  21. // Per-thread FileDescriptorWatcher registration.
  22. ThreadLocalPointer<FileDescriptorWatcher>& GetTlsFdWatcher() {
  23. static NoDestructor<ThreadLocalPointer<FileDescriptorWatcher>> tls_fd_watcher;
  24. return *tls_fd_watcher;
  25. }
  26. } // namespace
  27. class FileDescriptorWatcher::Controller::Watcher
  28. : public MessagePumpForIO::FdWatcher,
  29. public CurrentThread::DestructionObserver {
  30. public:
  31. Watcher(WeakPtr<Controller> controller,
  32. base::WaitableEvent& on_destroyed,
  33. MessagePumpForIO::Mode mode,
  34. int fd);
  35. Watcher(const Watcher&) = delete;
  36. Watcher& operator=(const Watcher&) = delete;
  37. ~Watcher() override;
  38. void StartWatching();
  39. private:
  40. friend class FileDescriptorWatcher;
  41. // MessagePumpForIO::FdWatcher:
  42. void OnFileCanReadWithoutBlocking(int fd) override;
  43. void OnFileCanWriteWithoutBlocking(int fd) override;
  44. // CurrentThread::DestructionObserver:
  45. void WillDestroyCurrentMessageLoop() override;
  46. // The MessagePumpForIO's watch handle (stops the watch when destroyed).
  47. MessagePumpForIO::FdWatchController fd_watch_controller_;
  48. // Runs tasks on the sequence on which this was instantiated (i.e. the
  49. // sequence on which the callback must run).
  50. const scoped_refptr<SequencedTaskRunner> callback_task_runner_ =
  51. SequencedTaskRunnerHandle::Get();
  52. // The Controller that created this Watcher. This WeakPtr is bound to the
  53. // |controller_| thread and can only be used by this Watcher to post back to
  54. // |callback_task_runner_|.
  55. WeakPtr<Controller> controller_;
  56. // WaitableEvent to signal to ensure that the Watcher is always destroyed
  57. // before the Controller.
  58. base::WaitableEvent& on_destroyed_;
  59. // Whether this Watcher is notified when |fd_| becomes readable or writable
  60. // without blocking.
  61. const MessagePumpForIO::Mode mode_;
  62. // The watched file descriptor.
  63. const int fd_;
  64. // Except for the constructor, every method of this class must run on the same
  65. // MessagePumpForIO thread.
  66. ThreadChecker thread_checker_;
  67. // Whether this Watcher was registered as a DestructionObserver on the
  68. // MessagePumpForIO thread.
  69. bool registered_as_destruction_observer_ = false;
  70. };
  71. FileDescriptorWatcher::Controller::Watcher::Watcher(
  72. WeakPtr<Controller> controller,
  73. base::WaitableEvent& on_destroyed,
  74. MessagePumpForIO::Mode mode,
  75. int fd)
  76. : fd_watch_controller_(FROM_HERE),
  77. controller_(controller),
  78. on_destroyed_(on_destroyed),
  79. mode_(mode),
  80. fd_(fd) {
  81. DCHECK(callback_task_runner_);
  82. thread_checker_.DetachFromThread();
  83. }
  84. FileDescriptorWatcher::Controller::Watcher::~Watcher() {
  85. DCHECK(thread_checker_.CalledOnValidThread());
  86. CurrentIOThread::Get()->RemoveDestructionObserver(this);
  87. // Stop watching the descriptor before signalling |on_destroyed_|.
  88. CHECK(fd_watch_controller_.StopWatchingFileDescriptor());
  89. on_destroyed_.Signal();
  90. }
  91. void FileDescriptorWatcher::Controller::Watcher::StartWatching() {
  92. DCHECK(thread_checker_.CalledOnValidThread());
  93. DCHECK(CurrentIOThread::IsSet());
  94. const bool watch_success = CurrentIOThread::Get()->WatchFileDescriptor(
  95. fd_, false, mode_, &fd_watch_controller_, this);
  96. DCHECK(watch_success) << "Failed to watch fd=" << fd_;
  97. if (!registered_as_destruction_observer_) {
  98. CurrentIOThread::Get()->AddDestructionObserver(this);
  99. registered_as_destruction_observer_ = true;
  100. }
  101. }
  102. void FileDescriptorWatcher::Controller::Watcher::OnFileCanReadWithoutBlocking(
  103. int fd) {
  104. DCHECK_EQ(fd_, fd);
  105. DCHECK_EQ(MessagePumpForIO::WATCH_READ, mode_);
  106. DCHECK(thread_checker_.CalledOnValidThread());
  107. // Run the callback on the sequence on which the watch was initiated.
  108. callback_task_runner_->PostTask(
  109. FROM_HERE, BindOnce(&Controller::RunCallback, controller_));
  110. }
  111. void FileDescriptorWatcher::Controller::Watcher::OnFileCanWriteWithoutBlocking(
  112. int fd) {
  113. DCHECK_EQ(fd_, fd);
  114. DCHECK_EQ(MessagePumpForIO::WATCH_WRITE, mode_);
  115. DCHECK(thread_checker_.CalledOnValidThread());
  116. // Run the callback on the sequence on which the watch was initiated.
  117. callback_task_runner_->PostTask(
  118. FROM_HERE, BindOnce(&Controller::RunCallback, controller_));
  119. }
  120. void FileDescriptorWatcher::Controller::Watcher::
  121. WillDestroyCurrentMessageLoop() {
  122. DCHECK(thread_checker_.CalledOnValidThread());
  123. if (callback_task_runner_->RunsTasksInCurrentSequence()) {
  124. // |controller_| can be accessed directly when Watcher runs on the same
  125. // thread.
  126. Watcher* watcher = controller_->watcher_;
  127. controller_->watcher_ = nullptr;
  128. delete watcher;
  129. } else {
  130. // If the Watcher and the Controller live on different threads, delete
  131. // |this| synchronously. Pending tasks bound to an unretained Watcher* will
  132. // not run since this loop is dead. The associated Controller will not know
  133. // whether the Watcher has been destroyed but it never uses it directly and
  134. // will ultimately send it to this thread for deletion (and that also won't
  135. // run since the loop being dead).
  136. delete this;
  137. }
  138. }
  139. FileDescriptorWatcher::Controller::Controller(MessagePumpForIO::Mode mode,
  140. int fd,
  141. const RepeatingClosure& callback)
  142. : callback_(callback),
  143. io_thread_task_runner_(GetTlsFdWatcher().Get()->io_thread_task_runner()) {
  144. DCHECK(!callback_.is_null());
  145. DCHECK(io_thread_task_runner_);
  146. watcher_ =
  147. new Watcher(weak_factory_.GetWeakPtr(), on_watcher_destroyed_, mode, fd);
  148. StartWatching();
  149. }
  150. FileDescriptorWatcher::Controller::~Controller() {
  151. DCHECK(sequence_checker_.CalledOnValidSequence());
  152. if (io_thread_task_runner_->BelongsToCurrentThread()) {
  153. // If the MessagePumpForIO and the Controller live on the same thread.
  154. if (watcher_)
  155. delete watcher_;
  156. } else {
  157. // Synchronously wait until |watcher_| is deleted on the MessagePumpForIO
  158. // thread. This ensures that the file descriptor is never accessed after
  159. // this destructor returns.
  160. //
  161. // We considered associating "generations" to file descriptors to avoid the
  162. // synchronous wait. For example, if the IO thread gets a "cancel" for fd=6,
  163. // generation=1 after getting a "start watching" for fd=6, generation=2, it
  164. // can ignore the "Cancel". However, "generations" didn't solve this race:
  165. //
  166. // T1 (client) Start watching fd = 6 with WatchReadable()
  167. // Stop watching fd = 6
  168. // Close fd = 6
  169. // Open a new file, fd = 6 gets reused.
  170. // T2 (io) Watcher::StartWatching()
  171. // Incorrectly starts watching fd = 6 which now refers to a
  172. // different file than when WatchReadable() was called.
  173. auto delete_task = BindOnce(
  174. [](Watcher* watcher) {
  175. // Since |watcher| is a raw pointer, it isn't deleted if this callback
  176. // is deleted before it gets to run.
  177. delete watcher;
  178. },
  179. Unretained(watcher_));
  180. io_thread_task_runner_->PostTask(FROM_HERE, std::move(delete_task));
  181. ScopedAllowBaseSyncPrimitivesOutsideBlockingScope allow;
  182. on_watcher_destroyed_.Wait();
  183. }
  184. // Since WeakPtrs are invalidated by the destructor, any pending RunCallback()
  185. // won't be invoked after this returns.
  186. }
  187. void FileDescriptorWatcher::Controller::StartWatching() {
  188. DCHECK(sequence_checker_.CalledOnValidSequence());
  189. if (io_thread_task_runner_->BelongsToCurrentThread()) {
  190. // If the MessagePumpForIO and the Controller live on the same thread.
  191. watcher_->StartWatching();
  192. } else {
  193. // It is safe to use Unretained() below because |watcher_| can only be
  194. // deleted by a delete task posted to |io_thread_task_runner_| by this
  195. // Controller's destructor. Since this delete task hasn't been posted yet,
  196. // it can't run before the task posted below.
  197. io_thread_task_runner_->PostTask(
  198. FROM_HERE, BindOnce(&Watcher::StartWatching, Unretained(watcher_)));
  199. }
  200. }
  201. void FileDescriptorWatcher::Controller::RunCallback() {
  202. DCHECK(sequence_checker_.CalledOnValidSequence());
  203. WeakPtr<Controller> weak_this = weak_factory_.GetWeakPtr();
  204. // Run a copy of the callback in case this Controller is deleted by the
  205. // callback. This would cause the callback itself to be deleted while it is
  206. // being run.
  207. RepeatingClosure callback_copy = callback_;
  208. callback_copy.Run();
  209. // If |this| wasn't deleted, re-enable the watch.
  210. if (weak_this)
  211. StartWatching();
  212. }
  213. FileDescriptorWatcher::FileDescriptorWatcher(
  214. scoped_refptr<SingleThreadTaskRunner> io_thread_task_runner)
  215. : io_thread_task_runner_(std::move(io_thread_task_runner)) {
  216. DCHECK(!GetTlsFdWatcher().Get());
  217. GetTlsFdWatcher().Set(this);
  218. }
  219. FileDescriptorWatcher::~FileDescriptorWatcher() {
  220. GetTlsFdWatcher().Set(nullptr);
  221. }
  222. std::unique_ptr<FileDescriptorWatcher::Controller>
  223. FileDescriptorWatcher::WatchReadable(int fd, const RepeatingClosure& callback) {
  224. return WrapUnique(new Controller(MessagePumpForIO::WATCH_READ, fd, callback));
  225. }
  226. std::unique_ptr<FileDescriptorWatcher::Controller>
  227. FileDescriptorWatcher::WatchWritable(int fd, const RepeatingClosure& callback) {
  228. return WrapUnique(
  229. new Controller(MessagePumpForIO::WATCH_WRITE, fd, callback));
  230. }
  231. #if DCHECK_IS_ON()
  232. void FileDescriptorWatcher::AssertAllowed() {
  233. DCHECK(GetTlsFdWatcher().Get());
  234. }
  235. #endif
  236. } // namespace base