src/async_loop-posix.h

// Copyright 2023 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#ifndef NINJA_ASYNC_LOOP_POSIX_H
#define NINJA_ASYNC_LOOP_POSIX_H

// Posix implementation of the AsyncLoop class.
// This contains specialized code paths for Linux ppoll() and MacOS kqueue()

#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <poll.h>
#include <signal.h>
#include <stddef.h>
#include <string.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>

#include "async_loop.h"
#include "async_loop_timers.h"
#include "util.h"     // For Fatal() and Win32Fatal().

class AsyncHandleImpl final : public AsyncHandle {
 public:
  enum class Kind {
    None = 0,
    Read,
    Write,
  };

  // Constructor.
  AsyncHandleImpl(ScopedHandle handle, AsyncLoop& async_loop,
                  AsyncHandle::Callback&& callback)
      : fd_(std::move(handle)), callback_(std::move(callback)),
        async_loop_(async_loop) {
    if (is_valid())
      async_loop_.AttachHandle(this);
  }

  // Destructor. Note that this type is unmoveable due to Win32
  // requirements.
  virtual ~AsyncHandleImpl() {
    if (is_valid())
      async_loop_.DetachHandle(this);
  }

  // Return AsyncLoop instance this instance belongs to.
  AsyncLoop& async_loop() const override { return async_loop_; }

  // Return AsyncLoop implementation instance.
  AsyncLoop::Impl& loop() const { return *async_loop_.impl_; }

  // Return file descriptor.
  int fd() const { return fd_.get(); }
  int get() const override { return fd(); }

  // Return true if the handle for this instnace is valid.
  bool is_valid() const override { return !!fd_; }

  // Release the native handle for this instance, making it invalid.
  ScopedHandle Release() override {
    ScopedHandle result;
    Reset(Kind::None);
    if (is_valid()) {
      async_loop_.DetachHandle(this);
      result = std::move(fd_);
    }
    return result;
  }

  // Is an asynchronous operation running?
  bool IsRunning() const override { return kind_ != Kind::None && !completed_; }

  // Cancel current asynchronous operation, if any.
  void Cancel() override {
    if (kind_ != Kind::None) {
      async_loop_.CancelHandle(this);
      Reset(Kind::None);
    }
  }

  // Reset the callback.
  void ResetCallback(AsyncHandle::Callback&& cb) override {
    callback_ = std::move(cb);
  }

  void ResetHandle(ScopedHandle handle) override {
    Reset(Kind::None);
    if (is_valid()) {
      async_loop_.DetachHandle(this);
    }
    fd_ = std::move(handle);
    if (is_valid())
      async_loop_.AttachHandle(this);
  }

  bool NeedsEvent() const { return !completed_ && kind_ != Kind::None; }

  bool NeedsWriteEvent() const { return !completed_ && (kind_ == Kind::Write); }

  bool NeedsReadEvent() const { return !completed_ && (kind_ == Kind::Read); }

  short poll_events() const {
    if (completed_)
      return 0;
    if (kind_ == Kind::Write)
      return POLLOUT;
    if (kind_ == Kind::Read)
      return POLLIN | POLLPRI;
    return 0;
  }

  short poll_revents() const { return poll_events() | POLLERR | POLLHUP; }

  // Reset state to start a new asynchronous operation. Cancels current one
  // if any.
  void Reset(Kind kind = Kind::None) {
    kind_ = kind;
    completed_ = false;
    error_ = 0;
    buffer_ = nullptr;
    wanted_size_ = 0;
    actual_size_ = 0;
  }

  void StartRead(void* buffer, size_t size) override {
    Reset(Kind::Read);
    buffer_ = buffer;
    wanted_size_ = size;
    if (!size)
      completed_ = true;
    else
      async_loop().UpdateHandle(this);
  }

  void StartWrite(const void* buffer, size_t size) override {
    Reset(Kind::Write);
    buffer_ = const_cast<void*>(buffer);
    wanted_size_ = size;
    if (!size)
      completed_ = true;
    else
      async_loop().UpdateHandle(this);
  }

  bool OnEvent() {
    error_ = 0;
    actual_size_ = 0;

    assert(!completed_);
    completed_ = true;

    switch (kind_) {
    case Kind::Read: {
      int ret;
      do {
        ret = ::read(fd_.get(), buffer_, wanted_size_);
      } while (ret < 0 && errno == EINTR);
      if (ret < 0) {
        error_ = errno;
        if (error_ == EAGAIN || error_ == EWOULDBLOCK) {
          // A spurious wakeup happened.
          return false;
        }
      } else {
        actual_size_ = static_cast<size_t>(ret);
      }
      break;
    }

    case Kind::Write: {
      int ret;
      do {
        ret = ::write(fd_.get(), buffer_, wanted_size_);
      } while (ret < 0 && errno == EINTR);
      if (ret < 0) {
        error_ = errno;
        if (error_ == EAGAIN || error_ == EWOULDBLOCK) {
          // A spurious wakeup happened.
          return false;
        }
      } else {
        actual_size_ = static_cast<size_t>(ret);
      }
      break;
    }

    default:
      assert(false && "Invalid runtime async op type!");
      return false;
    }
    async_loop_.UpdateHandle(this);
    callback_(error_, actual_size_);
    return true;
  }

  ScopedHandle fd_;
  Kind kind_ = Kind::None;
  bool completed_ = false;
  int error_ = 0;
  void* buffer_ = nullptr;
  size_t wanted_size_ = 0;
  size_t actual_size_ = 0;
  AsyncHandle::Callback callback_;
  AsyncLoop& async_loop_;
};

struct AsyncSigChildImpl final : public AsyncSigChild {
  AsyncSigChildImpl(AsyncLoop& async_loop, AsyncSigChild::Callback&& callback)
      : async_loop_(async_loop) {
    async_loop_.SetSigChildCallback(std::move(callback));
  }

  virtual ~AsyncSigChildImpl() { async_loop_.ClearSigChildCallback(); }

  AsyncLoop& async_loop() const override { return async_loop_; }

 private:
  AsyncLoop& async_loop_;
};

struct AsyncFdReadyImpl final : public AsyncFdReadyFlags {
  AsyncFdReadyImpl(AsyncLoop& async_loop, int fd)
      : async_loop_(async_loop), fd_(fd) {
    EnsureAttached();
  }

  virtual ~AsyncFdReadyImpl() { EnsureDetached(); }

  AsyncLoop& async_loop() const override { return async_loop_; }

  int fd() const { return fd_; }

  // Return true if file descriptor is valid.
  bool is_valid() const { return fd_ >= 0; }

  bool IsReadReady() const override { return read_ready_; }
  bool IsWriteReady() const override { return write_ready_; }

  void ResetFlags() override {
    read_ready_ = write_ready_ = false;
    if (fd_ >= 0)
      async_loop_.UpdateFdReady(this);
  }

  // Change the file descriptor. If |fd < 0| this will never
  // receive any event. Always resets the flag.
  void ResetFd(int fd) override {
    ResetFlags();
    if (fd != fd_) {
      EnsureDetached();
      fd_ = fd;
      EnsureAttached();
    }
  }

  void WantFlags(bool want_read, bool want_write) override {
    want_read_ = want_read;
    want_write_ = want_write;
    ResetFlags();
  }

  // Used by Watches classes.
  bool want_read() const { return want_read_; }

  bool want_write() const { return want_write_; }

  void SetFlags(bool read_ready, bool write_ready) {
    read_ready_ = read_ready;
    write_ready_ = write_ready;
  }

 private:
  void EnsureAttached() {
    if (fd_ >= 0)
      async_loop_.AttachFdReady(this);
  }

  void EnsureDetached() {
    if (fd_ >= 0)
      async_loop_.DetachFdReady(this);
  }

  AsyncLoop& async_loop_;
  bool want_read_ = false;
  bool want_write_ = false;

  bool read_ready_ = false;
  bool write_ready_ = false;
  int fd_ = -1;
};

template <int SIGNUM>
struct SignalHandler {
  /// Uninstall the signal handler. Assume the signal
  /// is already blocked.
  void InstallBlocked() {
    s_signaled_ = 0;
    struct sigaction action = {};
    action.sa_handler = [](int) { s_signaled_ = 1; };
    if (sigaction(SIGNUM, &action, &prev_action_) < 0)
      ErrnoFatal("sigaction");
  }

  /// Uninstall the signal handler. Assume the signal
  /// is already blocked.
  void UninstallBlocked() { sigaction(SIGNUM, &prev_action_, nullptr); }

  /// Clear signal flag.
  void Clear() { s_signaled_ = 0; }

  /// Return true if the signal was signalled at least once.
  bool IsSignaled() const { return s_signaled_ != 0; }

  /// Set the flag if the signal is pending.
  void HandlePending() {
    sigset_t pending;
    if (!sigpending(&pending) && sigismember(&pending, SIGNUM))
      s_signaled_ = 1;
  }

 private:
  struct sigaction prev_action_;
  static volatile sig_atomic_t s_signaled_;
};

template <int SIGNUM>
volatile sig_atomic_t SignalHandler<SIGNUM>::s_signaled_ = 0;

struct SignalsCatcher {
  // Constructor installs the signal handlers, and optionally
  // leave the signals blocked.
  SignalsCatcher(bool block_signals) {
    // Block all signals before changing their handler.
    sigset_t mask = GetSignalMask();
    sigprocmask(SIG_BLOCK, &mask, &old_signal_mask_);

    sigint_.InstallBlocked();
    sighup_.InstallBlocked();
    sigterm_.InstallBlocked();
    sigchld_.InstallBlocked();

    // Unblock signals if necessary.
    if (!block_signals)
      sigprocmask(SIG_UNBLOCK, &mask, nullptr);

    // Compute wait mask to unblocks all signals for pselect() / ppoll().
    wait_signal_mask_ = old_signal_mask_;
    sigdelset(&wait_signal_mask_, SIGINT);
    sigdelset(&wait_signal_mask_, SIGHUP);
    sigdelset(&wait_signal_mask_, SIGTERM);
    sigdelset(&wait_signal_mask_, SIGCHLD);
  }

  // Destructor restores the previous signal handlers and
  // the global signal mask that were recorded in the
  // constructor.
  ~SignalsCatcher() {
    sigset_t mask = GetSignalMask();
    sigprocmask(SIG_BLOCK, &mask, nullptr);

    sigint_.UninstallBlocked();
    sighup_.UninstallBlocked();
    sigterm_.UninstallBlocked();
    sigchld_.UninstallBlocked();

    sigprocmask(SIG_SETMASK, &old_signal_mask_, nullptr);
  }

  // Return previous global signal mask.
  sigset_t GetOldSignalMask() const { return old_signal_mask_; }

  // Return reference to signal mask that can be used in pselect() / ppoll()
  // to wait while unblocking the signals atomically.
  const sigset_t& GetWaitSignalMask() const { return wait_signal_mask_; }

  // Clear all signal flags.
  void Clear() {
    sigint_.Clear();
    sighup_.Clear();
    sigterm_.Clear();
    sigchld_.Clear();
  }

  // Return true if either one of SIGINT/SIGHUP/SIGTERM was signaled.
  bool IsInterrupted() const {
    return sigint_.IsSignaled() || sighup_.IsSignaled() ||
           sigterm_.IsSignaled();
  }

  // Check pending signals, updating their corresponding flags.
  void HandlePendingInterrupt() {
    sigint_.HandlePending();
    sighup_.HandlePending();
    sigterm_.HandlePending();
  }

  // Return the interruption signal, or 0.
  int GetInterruptSignal() const {
    if (sigint_.IsSignaled())
      return SIGINT;
    if (sighup_.IsSignaled())
      return SIGHUP;
    if (sigterm_.IsSignaled())
      return SIGTERM;
    else
      return 0;
  }

  // Return true if SIGCHLD was signaled, meaning that at least
  // one child has stopped.
  bool HasChildStopped() const { return sigchld_.IsSignaled(); }

 private:
  SignalHandler<SIGINT> sigint_;
  SignalHandler<SIGHUP> sighup_;
  SignalHandler<SIGTERM> sigterm_;
  SignalHandler<SIGCHLD> sigchld_;

  static sigset_t GetSignalMask() {
    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGINT);
    sigaddset(&mask, SIGHUP);
    sigaddset(&mask, SIGTERM);
    sigaddset(&mask, SIGCHLD);
    return mask;
  }

  sigset_t old_signal_mask_;
  sigset_t wait_signal_mask_;
};

class AsyncLoop::Impl {
  template <typename T>
  struct UnorderedPointerList : public std::vector<T*> {
    /// Indicates no position found.
    static constexpr size_t npos = ~size_t(0);

    /// Find the index of a given value, or return npos.
    size_t IndexOf(T* aptr) const {
      size_t result = 0;
      for (auto* s : *this) {
        if (s == aptr)
          return result;
        ++result;
      }
      return npos;
    }

    bool Contains(T* aptr) const {
      for (auto* ptr : *this) {
        if (ptr == aptr)
          return true;
      }
      return false;
    }

    /// Remove one item from the list. Does not preserve order for performance.
    /// Returns true if the item was found and removed, false otherwise.
    bool Remove(T* aptr) {
      for (auto it = this->begin(); it != this->end(); ++it) {
        if (*it == aptr) {
          // Order does not matter for this list, so just move last item
          // to current location to avoid O(n) removal cost.
          auto it_last = this->end() - 1;
          if (it != it_last)
            *it = *it_last;
          this->pop_back();
          return true;
        }
      }
      return false;
    }
  };

  /// The list of pending AsyncHandleImpls that have completed but whose
  /// Invoke() method wasn't called yet. Identified by their address.
  using PendingList = UnorderedPointerList<AsyncHandleImpl>;

  /// The list of AsyncFdReadyImpls corresponding to file descriptors
  /// with readiness reported by ppoll/pselect().
  using ReadyList = UnorderedPointerList<AsyncFdReadyImpl>;

  PendingList pending_ops_;
  ReadyList ready_ops_;

#if defined(USE_PPOLL)
  class Watches {
    // To avoid the overhead of a map, use two vectors to store
    // pointers to async objects. Order does not matter.
    UnorderedPointerList<AsyncHandleImpl> states_;
    UnorderedPointerList<AsyncFdReadyImpl> readys_;

    // An array of |states_.size() + readys_.size()| pollfd struct, since
    // ppoll() expects a contiguous array of pollfd items in memory.
    //
    // This is initially empty, and populated on demand by WaitForEvents().
    // Adding or removing handles invalidates / clears the array.
    std::vector<pollfd> polls_;

    void Invalidate() { polls_.clear(); }

    void EnsureUpToDate() {
      if (polls_.empty()) {
        // Since handles have been added or removed since the last call,
        // recompute the array.
        polls_.reserve(states_.size() + readys_.size());
        for (const auto* state : states_) {
          polls_.push_back({ state->fd(), state->poll_events(), 0 });
        }
        for (const auto* ready : readys_) {
          int fd = ready->fd();
          short events = 0;
          if (ready->want_read())
            events |= POLLIN | POLLPRI;
          if (ready->want_write())
            events |= POLLOUT;
          polls_.push_back({ ready->fd(), events, 0 });
        }
      }
    }

   public:
    static constexpr size_t npos = ~size_t(0);

    ~Watches() {
      assert(states_.empty() &&
             "Destroying AsyncLoop before child AsyncHandle instances!");
    }

    bool HasWaiters() {
      EnsureUpToDate();
      for (const auto& poll : polls_) {
        if (poll.events != 0)
          return true;
      }
      return false;
    }

    void Insert(AsyncHandleImpl* state) {
      assert(!states_.Contains(state) && "Async state already in the set!");
      states_.push_back(state);
      Invalidate();
    }

    void Remove(AsyncHandleImpl* state) {
      bool removed = states_.Remove(state);
      assert(removed && "Async state not in the set!");
      Invalidate();
    }

    void Update(AsyncHandleImpl* state) {
      assert(states_.Contains(state) && "Updating unknown async state!");
      Invalidate();
    }

    void Insert(AsyncFdReadyImpl* ready) {
      assert(!readys_.Contains(ready) && "Async ready already in the set!");
      readys_.push_back(ready);
      Invalidate();
    }

    void Remove(AsyncFdReadyImpl* ready) {
      bool removed = readys_.Remove(ready);
      assert(removed && "Async ready not in the set!");
      Invalidate();
    }

    void Update(AsyncFdReadyImpl* ready) {
      assert(readys_.Contains(ready) && "Async ready is not in the set!");
      Invalidate();
    }

    int WaitForEvents(const struct timespec* ts, PendingList& pending_list,
                      size_t* ready_count, const sigset_t* signal_mask) {
      EnsureUpToDate();

      *ready_count = 0;

      int ret = ppoll(polls_.size() > 0 ? &polls_.front() : nullptr,
                      static_cast<nfds_t>(polls_.size()), ts, signal_mask);
      if (ret < 0) {
        if (errno != EINTR)
          ErrnoFatal("ppoll");
      } else if (ret > 0) {
        auto cur_poll = polls_.begin();
        for (AsyncHandleImpl* state : states_) {
          assert(state->fd() == cur_poll->fd);
          if (cur_poll->revents & state->poll_revents())
            pending_list.push_back(state);
          cur_poll++;
        }
        for (AsyncFdReadyImpl* ready : readys_) {
          int fd = ready->fd();
          if (fd < 0)
            continue;
          assert(ready->fd() == cur_poll->fd);
          short events = cur_poll->revents;
          if (events) {
            bool read_ready = events & (POLLIN | POLLPRI | POLLERR | POLLHUP);
            bool write_ready = events & (POLLOUT | POLLERR | POLLHUP);
            ready->SetFlags(read_ready, write_ready);
            *ready_count += 1;
          }
          cur_poll++;
        }
      }
      return ret;
    }
  };

#else   // !USE_PPOLL
  class Watches {
    static constexpr int kInvalid = -2;
    mutable int max_fds_ = kInvalid;

    UnorderedPointerList<AsyncHandleImpl> states_;
    UnorderedPointerList<AsyncFdReadyImpl> readys_;

    fd_set read_fds_;
    fd_set write_fds_;
    fd_set event_read_fds_;
    fd_set event_write_fds_;

    void Invalidate() { max_fds_ = kInvalid; }

    int num_fds() const {
      // Recompute max file descriptor if needed.
      if (max_fds_ == kInvalid) {
        max_fds_ = -1;
        for (AsyncHandleImpl* state : states_) {
          if (state->NeedsReadEvent() || state->NeedsWriteEvent()) {
            if (state->fd() > max_fds_)
              max_fds_ = state->fd();
          }
        }
        for (AsyncFdReadyImpl* ready : readys_) {
          int fd = ready->fd();
          if (fd >= 0 && fd > max_fds_)
            max_fds_ = fd;
        }
      }
      return max_fds_ + 1;
    }

   public:
    Watches() {
      FD_ZERO(&read_fds_);
      FD_ZERO(&write_fds_);
    }

    void Insert(AsyncHandleImpl* state) {
      if (state->fd() >= static_cast<int>(FD_SETSIZE))
        Fatal("File descriptor too large for pselect(): %d\n", state->fd());

      bool contained = states_.Contains(state);
      assert(!contained && "Async state already in the set!");
      if (contained)
        return;
      states_.push_back(state);
      if (state->NeedsReadEvent())
        FD_SET(state->fd(), &read_fds_);
      if (state->NeedsWriteEvent())
        FD_SET(state->fd(), &write_fds_);
      Invalidate();
    }

    void Remove(AsyncHandleImpl* state) {
      bool removed = states_.Remove(state);
      assert(removed && "Removing unknown Async state!");
      if (!removed)
        return;
      FD_CLR(state->fd(), &read_fds_);
      FD_CLR(state->fd(), &write_fds_);
      Invalidate();
    }

    void Update(AsyncHandleImpl* state) {
      int fd = state->fd();
      if (state->NeedsReadEvent()) {
        FD_SET(fd, &read_fds_);
      } else {
        FD_CLR(fd, &read_fds_);
      }
      if (state->NeedsWriteEvent()) {
        FD_SET(fd, &write_fds_);
      } else {
        FD_CLR(fd, &write_fds_);
      }
    }

    void Insert(AsyncFdReadyImpl* ready) {
      bool contained = readys_.Contains(ready);
      assert(!contained && "Async FdReady already in the set!");
      if (contained)
        return;
      readys_.push_back(ready);
      int fd = ready->fd();
      assert(fd >= 0 && "Invalid file descriptor");
      if (ready->want_read())
        FD_SET(fd, &read_fds_);
      if (ready->want_write())
        FD_SET(fd, &write_fds_);
      Invalidate();
    }

    void Remove(AsyncFdReadyImpl* ready) {
      bool removed = readys_.Remove(ready);
      assert(removed && "Async FdReady was not in the set!");
      if (!removed)
        return;
      int fd = ready->fd();
      assert(fd >= 0);
      FD_CLR(fd, &read_fds_);
      FD_CLR(fd, &write_fds_);
      Invalidate();
    }

    void Update(AsyncFdReadyImpl* ready) {
      assert(readys_.Contains(ready) && "Async ready is not in the set!");
      int fd = ready->fd();
      assert(fd >= 0);
      if (ready->want_read()) {
        FD_SET(fd, &read_fds_);
      } else {
        FD_CLR(fd, &read_fds_);
      }
      if (ready->want_write()) {
        FD_SET(fd, &write_fds_);
      } else {
        FD_CLR(fd, &write_fds_);
      }
      Invalidate();
    }

    bool HasWaiters() const {
      for (const auto* state : states_) {
        if (state->IsRunning())
          return true;
      }
      for (const auto* ready : readys_) {
        if (ready->fd() >= 0)
          return true;
      }
      return false;
    }

    int WaitForEvents(const struct timespec* ts, PendingList& pending_list,
                      size_t* ready_count, const sigset_t* signal_mask) {
      event_read_fds_ = read_fds_;
      event_write_fds_ = write_fds_;

      *ready_count = 0;

      int count = num_fds();
      int ret = pselect(count, count > 0 ? &event_read_fds_ : nullptr,
                        count > 0 ? &event_write_fds_ : nullptr, nullptr, ts,
                        signal_mask);
      if (ret < 0) {
        if (errno != EINTR)
          ErrnoFatal("pselect");
      } else if (ret > 0) {
        for (auto* state : states_) {
          int fd = state->fd();
          bool has_event = false;
          has_event |=
              state->NeedsWriteEvent() && FD_ISSET(fd, &event_write_fds_);
          has_event |=
              state->NeedsReadEvent() && FD_ISSET(fd, &event_read_fds_);
          if (has_event)
            pending_list.emplace_back(state);
        }
        for (auto* ready : readys_) {
          int fd = ready->fd();
          if (fd < 0)
            continue;
          bool read_ready =
              ready->want_read() && FD_ISSET(fd, &event_read_fds_);
          bool write_ready =
              ready->want_write() && FD_ISSET(fd, &event_write_fds_);
          ready->SetFlags(read_ready, write_ready);
          if (read_ready || write_ready)
            *ready_count += 1;
        }
      }
      return ret;
    }
  };
#endif  // !USE_PPOLL

 public:
  ~Impl() {
    assert(!watches_.HasWaiters() &&
           "Destroying AsyncLoop before children AsyncHandle instances!");
    assert(timers_.empty() &&
           "Destroying AsyncLoop before children AsyncTimer instances!");
  }

  AsyncLoopTimers& timers() { return timers_; }

  void AttachHandle(AsyncHandleImpl* state) {
    assert(state->is_valid() && "Trying to attach invalid handle");
    watches_.Insert(state);
  }

  void DetachHandle(AsyncHandleImpl* state) {
    assert(state->is_valid() && "Trying to detach invalid handle");
    watches_.Remove(state);
  }

  void UpdateHandle(AsyncHandleImpl* state) { watches_.Update(state); }

  void CancelHandle(AsyncHandleImpl* state) {
    pending_ops_.Remove(state);
    watches_.Update(state);
  }

  void AttachFdReady(AsyncFdReadyImpl* ready) {
    assert(ready->is_valid() && "Trying to attach invalid FdReady object");
    watches_.Insert(ready);
  }

  void DetachFdReady(AsyncFdReadyImpl* ready) {
    assert(ready->is_valid() && "Trying to detach invalid FdReady object");
    watches_.Remove(ready);
  }

  void UpdateFdReady(AsyncFdReadyImpl* ready) {
    assert(ready->is_valid() && "Trying to update invalid FdReady object");
    watches_.Update(ready);
  }

  AsyncLoop::ExitStatus RunOnce(int64_t timeout_ms, AsyncLoop& loop) {
    assert(pending_ops_.empty());

    /// An interrupt occured outside of this loop, return immediately.
    if (signals_catcher_ && signals_catcher_->IsInterrupted())
      return ExitInterrupted;

    /// Handle timeout.
    bool has_timers = false;
    int64_t timer_expiration_ms = timers_.ComputeNextExpiration();
    if (timer_expiration_ms >= 0) {
      has_timers = true;
      int64_t now_ms = loop.NowMs();
      int64_t timer_timeout_ms = timer_expiration_ms - now_ms;
      if (timer_timeout_ms < 0) {
        timer_timeout_ms = 0;
      }

      if (timeout_ms < 0)
        timeout_ms = timer_timeout_ms;
      else if (timeout_ms > timer_timeout_ms)
        timeout_ms = timer_timeout_ms;
    }

    const struct timespec* ts = NULL;
    struct timespec timeout_ts = {};
    if (timeout_ms >= 0) {
      timeout_ts.tv_sec = static_cast<time_t>(timeout_ms / 1000);
      timeout_ts.tv_nsec = static_cast<long>((timeout_ms % 1000) * 1000000LL);
      ts = &timeout_ts;
    }

    /// Exit immediately if there is nothing to do and no timeout.
    if (!watches_.HasWaiters() && !sigchild_cb_ && pending_ops_.empty() &&
        timeout_ms < 0)
      return ExitIdle;

    if (signals_catcher_) {
      signals_catcher_->Clear();
    }
    size_t ready_count = 0;
    int ret = watches_.WaitForEvents(
        ts, pending_ops_, &ready_count,
        signals_catcher_ ? &signals_catcher_->GetWaitSignalMask() : nullptr);

    ExitStatus result = ExitTimeout;

    if (ret == -1) {
      // This corresponds to EINTR, which could be an interruption or SIGCHLD.
      if (!signals_catcher_ || signals_catcher_->IsInterrupted()) {
        return ExitInterrupted;
      }
      if (signals_catcher_->HasChildStopped()) {
        if (sigchild_cb_)
          sigchild_cb_();
        result = ExitSuccess;
      }
    }

    // Handle all pending operations.
    // Note that OnEvent() may invoke a callback that changes the
    // state of pending_ops_.
    while (!pending_ops_.empty()) {
      AsyncHandleImpl* state = pending_ops_.back();
      pending_ops_.pop_back();
      if (state->OnEvent())
        result = ExitSuccess;
    }

    if (ready_count > 0) {
      // Ready flags have already been reported.
      result = ExitSuccess;
    }

    if (has_timers && timers_.ProcessExpiration(loop.NowMs()))
      result = ExitSuccess;

    return result;
  }

  void ClearInterrupt() {
    if (signals_catcher_)
      signals_catcher_->Clear();
  }

  int GetInterruptSignal() const {
    return signals_catcher_ ? signals_catcher_->GetInterruptSignal() : 0;
  }

  sigset_t GetOldSignalMask() const {
    if (signals_catcher_)
      return signals_catcher_->GetOldSignalMask();

    sigset_t old_mask;
    sigset_t empty;
    sigemptyset(&empty);
    sigprocmask(SIG_BLOCK, &empty, &old_mask);
    return old_mask;
  }

  void EnableInterruptCatcher() {
    signals_catcher_.reset(new SignalsCatcher(kShouldBlockSignals));
  }

  void DisableInterruptCatcher() { signals_catcher_.reset(); }

  void SetSigChildCallback(AsyncSigChild::Callback&& cb) {
    if (sigchild_cb_)
      Fatal("Cannot register more than one SIGCHLD callback at a time!");
    sigchild_cb_ = std::move(cb);
  }

  void ClearSigChildCallback() { sigchild_cb_ = nullptr; }

 private:
  std::unique_ptr<SignalsCatcher> signals_catcher_;

  // Optional callback invoked when SIGCHLD is signaled.
  // This is treated as an asynchronous operation as it
  // should not stop the AsyncLoop. This also keeps the
  // AsyncLoop::Run() interface simple.
  AsyncSigChild::Callback sigchild_cb_;

  // A flag indicating whether SIGINT/SIGHUP/SIGTERM/SIGCHLD signals
  // should be blocked, except when waiting in pselect()  or ppoll().
  // When using kqueue(), signals should not be blocked as kevent()
  // lacks the ability to atomically change the signal mask while
  // waiting.
#ifdef USE_KQUEUE
  static constexpr bool kShouldBlockSignals = false;
#else   // !USE_KQUEUE
  static constexpr bool kShouldBlockSignals = true;
#endif  // !USE_KQUEUE

  AsyncLoopTimers timers_;
  Watches watches_;
};

#endif  // NINJA_ASYNC_LOOP_POSIX_H