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