Add a new file descriptor watch API.

base/files/file_descriptor_watcher_posix.cc

Index: base/files/file_descriptor_watcher_posix.cc
diff --git a/base/files/file_descriptor_watcher_posix.cc b/base/files/file_descriptor_watcher_posix.cc
new file mode 100644
index 0000000000000000000000000000000000000000..7e7c600726903e3eeeb7de51cc2dff279fa82011
--- /dev/null
+++ b/base/files/file_descriptor_watcher_posix.cc
@@ -0,0 +1,208 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// 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/bind.h"
+#include "base/lazy_instance.h"
+#include "base/logging.h"
+#include "base/memory/ptr_util.h"
+#include "base/sequenced_task_runner.h"
+#include "base/single_thread_task_runner.h"
+#include "base/threading/sequenced_task_runner_handle.h"
+#include "base/threading/thread_checker.h"
+#include "base/threading/thread_local.h"
+
+namespace base {
+
+namespace {
+
+// MessageLoopForIO used to watch file descriptors for which callbacks are
+// registered from a given thread.
+LazyInstance<ThreadLocalPointer<MessageLoopForIO>>::Leaky
+    tls_message_loop_for_io = LAZY_INSTANCE_INITIALIZER;
+
+}  // namespace
+
+FileDescriptorWatcher::Controller::~Controller() {
+  DCHECK(sequence_checker_.CalledOnValidSequence());
+  if (was_deleted_) {
+    DCHECK(!*was_deleted_);
+    *was_deleted_ = true;
+  }
+  message_loop_for_io_task_runner_->DeleteSoon(FROM_HERE, watcher_.release());
+  // Since WeakPtrs are invalidated by the destructor, RunCallback() won't be
+  // invoked after this returns.
+}
+
+class FileDescriptorWatcher::Controller::Watcher
+    : public MessageLoopForIO::Watcher,
+      public MessageLoop::DestructionObserver {
+ public:
+  Watcher(WeakPtr<Controller> controller, MessageLoopForIO::Mode mode, int fd);
+  ~Watcher() override;
+
+ private:
+  friend class FileDescriptorWatcher;
+
+  // MessageLoopForIO::Watcher:
+  void OnFileCanReadWithoutBlocking(int fd) override;
+  void OnFileCanWriteWithoutBlocking(int fd) override;
+
+  // MessageLoop::DestructionObserver:
+  void WillDestroyCurrentMessageLoop() override;
+
+  // Used to instruct the MessageLoopForIO to stop watching the file descriptor.
+  MessageLoopForIO::FileDescriptorWatcher file_descriptor_watcher_;
+
+  // 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_ =
+      SequencedTaskRunnerHandle::Get();
+
+  // The Controller that created this Watcher.
+  WeakPtr<Controller> controller_;
+
+  // Whether this Watcher is notified when |fd_| becomes readable or writable
+  // without blocking.
+  const MessageLoopForIO::Mode mode_;
+
+  // The watched file descriptor.
+  const int fd_;
+
+  // Except for the constructor, every method of this class must run on the same
+  // MessageLoopForIO thread.
+  ThreadChecker thread_checker_;
+
+  DISALLOW_COPY_AND_ASSIGN(Watcher);
+};
+
+FileDescriptorWatcher::Controller::Watcher::Watcher(
+    WeakPtr<Controller> controller,
+    MessageLoopForIO::Mode mode,
+    int fd)
+    : controller_(controller), mode_(mode), fd_(fd) {
+  DCHECK(callback_task_runner_);
+  thread_checker_.DetachFromThread();
+}
+
+FileDescriptorWatcher::Controller::Watcher::~Watcher() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  MessageLoopForIO::current()->RemoveDestructionObserver(this);
+}
+
+void FileDescriptorWatcher::Controller::Watcher::OnFileCanReadWithoutBlocking(
+    int fd) {
+  DCHECK_EQ(fd_, fd);
+  DCHECK_EQ(MessageLoopForIO::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,
+                                  Bind(&Controller::RunCallback, controller_));
+}
+
+void FileDescriptorWatcher::Controller::Watcher::OnFileCanWriteWithoutBlocking(
+    int fd) {
+  DCHECK_EQ(fd_, fd);
+  DCHECK_EQ(MessageLoopForIO::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,
+                                  Bind(&Controller::RunCallback, controller_));
+}
+
+void FileDescriptorWatcher::Controller::Watcher::
+    WillDestroyCurrentMessageLoop() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // A Watcher is owned by a Controller. When the Controller is deleted, it
+  // transfers ownership of the Watcher to a delete task posted to the
+  // MessageLoopForIO. If the MessageLoopForIO is deleted before the delete task
+  // runs, the following line takes care of deleting the Watcher.
+  delete this;
+}
+
+FileDescriptorWatcher::Controller::Controller(MessageLoopForIO::Mode mode,
+                                              int fd,
+                                              const Closure& callback)
+    : callback_(new Closure(callback)),
+      message_loop_for_io_task_runner_(
+          tls_message_loop_for_io.Get().Get()->task_runner()),
+      weak_factory_(this) {
+  DCHECK(!callback_->is_null());
+  DCHECK(message_loop_for_io_task_runner_);
+  watcher_ = MakeUnique<Watcher>(weak_factory_.GetWeakPtr(), mode, fd);
+  StartWatching(true);
+}
+
+void FileDescriptorWatcher::Controller::StartWatching(
+    bool called_from_constructor) {
+  DCHECK(sequence_checker_.CalledOnValidSequence());
+  message_loop_for_io_task_runner_->PostTask(
+      FROM_HERE,
+      Bind(
+          [](Watcher* watcher, bool called_from_constructor) {
+            // |watcher| is guaranteed to be alive when this task
+            // runs because it can only be deleted by a task posted
+            // to the MessageLoopForIO by the Controller's
+            // destructor.
+            MessageLoopForIO::current()->WatchFileDescriptor(
+                watcher->fd_, false, watcher->mode_,
+                &watcher->file_descriptor_watcher_, watcher);
+
+            if (called_from_constructor) {
+              MessageLoopForIO::current()->AddDestructionObserver(watcher);

[dcheng, 2016/09/14 06:48:19]

Is it possible to just do this in the ctor?

[fdoray, 2016/09/14 17:06:14]

No. AddDestructionObserver has to be called on the MessageLoopForIO thread.
Added a comment to clarify that.

+            }
+          },
+          Unretained(watcher_.get()), called_from_constructor));
+}
+
+void FileDescriptorWatcher::Controller::RunCallback() {
+  DCHECK(sequence_checker_.CalledOnValidSequence());
+
+  // |was_deleted| will be set to true if the callback deletes |this|.
+  bool was_deleted = false;
+  was_deleted_ = &was_deleted;

[dcheng, 2016/09/14 06:48:19]

Isn't this what WeakPtr is for? WeakPtr<Controller> weak_this;

And then check weak_this below?

[fdoray, 2016/09/14 17:06:14]

Done. You're right.

+
+  // Take ownership of |callback_| in case it deletes |this|.
+  auto callback = std::move(callback_);

[dcheng, 2016/09/14 06:48:19]

Why do we have to transfer ownership at all? Why can't we just leave it?

[fdoray, 2016/09/14 17:06:14]

|callback_| can delete |this| and hence delete itself. I wasn't sure whether
that was valid, but since FileDescriptorWatcherTest.DeleteControllerFromCallback
succeeds, it must be.

[dcheng, 2016/09/14 17:13:17]

Sure, but that's OK: we can just check if |this| was deleted and skip the rest
of the work. We don't need to actually reference |callback_| at all after it
runs, so why bother saving and restoring it?

[fdoray, 2016/09/14 18:15:40]

In the latest patch set, I no longer save/restore callback_.

+  callback->Run();
+
+  // If |this| wasn't deleted, return ownership of |callback| and re-enable the
+  // watch.
+  if (!was_deleted) {
+    callback_ = std::move(callback);
+    was_deleted_ = nullptr;
+    StartWatching(false);
+  }
+}
+
+FileDescriptorWatcher::FileDescriptorWatcher(
+    MessageLoopForIO* message_loop_for_io) {
+  DCHECK(message_loop_for_io);
+  DCHECK(!tls_message_loop_for_io.Get().Get());
+  tls_message_loop_for_io.Get().Set(message_loop_for_io);
+}
+
+FileDescriptorWatcher::~FileDescriptorWatcher() {
+  tls_message_loop_for_io.Get().Set(nullptr);
+}
+
+std::unique_ptr<FileDescriptorWatcher::Controller>
+FileDescriptorWatcher::WatchReadable(int fd, const Closure& callback) {
+  return std::unique_ptr<Controller>(

[dcheng, 2016/09/14 06:48:19]

WrapUnique() here and below

[fdoray, 2016/09/14 17:06:14]

Done.

+      new Controller(MessageLoopForIO::WATCH_READ, fd, callback));
+}
+
+std::unique_ptr<FileDescriptorWatcher::Controller>
+FileDescriptorWatcher::WatchWritable(int fd, const Closure& callback) {
+  return std::unique_ptr<Controller>(
+      new Controller(MessageLoopForIO::WATCH_WRITE, fd, callback));
+}
+
+}  // namespace base