src/async_loop.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_H
#define NINJA_ASYNC_LOOP_H

#include "ipc_handle.h"

#ifndef _WIN32
#include <signal.h>
#endif

#include <functional>
#include <memory>

/// AsyncLoop provides an abstraction to perform asynchronous i/o
/// operations on Posix and Win32. Its API is inspired by Win32
/// overlapped operations, since this model can be implemented
/// efficiently on Posix.
///
/// AsyncHandle wraps an i/o handle, and a user-provided callback.
/// It provides methods like StartRead() to initiate an asynchronous operation.
///
/// AsyncTimer wraps an expiration date and a user-provided callback.
/// It provides SetExpirationMs() and SetDurationMs() methods to indicate
/// when the timer's next expiration should occur.
///
/// Each AsyncHandle or AsyncTimer instance is scoped to a parent AsyncLoop
/// instance.
///
/// Example usage:
///
///  1) Obtain a reference to an AsyncLoop instance, e.g. by
///     calling AsyncLoop::Get() or AsyncLoop::CreateLocal().
///
///  2) Create as many AsyncHandle and AsyncTimer instances from
///     the parent loop as needed. passing a user-provided callback
///     when constructing each one of them.
///
///  4) Use AsyncHandle::StartXXX() methods to start an i/o asynchronous
///     operation, or AsyncTimer::Set{Expiration,Duration}Ms() to set a
///     timer's expiration time.
///
///  5) Call AsyncLoop::RunOnce() or AsyncLoop::RunUntil() to run the
///     the loop. This waits for the completion of async i/o events or
///     expiration of timers, and invokes their callbacks automatically.
///
///     Using AsyncLoop::RunOnce(-1) will wait until one of the following
///     happens:
///
///       - There are no more active handles or timers, in which case
///         the function returns immediately with AsyncLoop::ExitIdle.
///
///       - At least one AsyncHandle's i/o operation completes, in which
///         case their callbacks are invoked before the function returns
///         AsyncLoop::ExitSuccess.
///
///       - At least one AsyncTimer expires, in which case their callbacks
///         are invoked before the function returns AsyncLoop::ExitSuccess.
///
///       - If the AsyncLoop's interrupt catching feature is enabled (see
///         below), return immediately with AsyncLoop::ExitInterrupted.
///
///     It is also possible to pass a timeout in milliseconds, e.g.
///     calling AsyncLoop::RunOnce(1000) to wait for one second. If the
///     timeout expires without an i/o completion or timer expiration,
///     the function returns with AsyncLoop::ExitTimeout.
///
///     Calling AsyncLoop::RunUntil() is more convenient as it allows to
///     wait for specific conditions instead.
///
/// Some important points:
///
/// - There is a global AsyncLoop instance which can be
///   retrieved with AsyncLoop::Get(), this is meant to be
///   used from the main thread only.
///
/// - Use AsyncLoop::CreateLocal() to create additionnal
///   AsyncLoop instances. This can be useful to run them in
///   background threads.
///
/// - It is a runtime error to destroy an AsyncLoop before any of its
///   children AsyncHandle or AsyncTimer instances (assertion failure on
///   debug build).
///
/// - An AsyncLoop instance can catch interrupts (i.e. Ctrl-C
///   on Windows, and SIGINT/SIGUP/SIGTERM on Posix) automatically
///   and report them in AsyncLoop::ExitStatus.
///
///   IMPORTANT: At the moment, the global AsyncLoop always catches
///   interrupts by default. This may change in the future.
///   TODO(digit): Change this and make all uses explicit.
///
///   Only one AsyncLoop instance can catch interrupts at any
///   given time. Use AsyncLoop::ScopedInterruptCatcher if you
///   want to force a given AsyncLoop instance to catch
///   interrupts (see its documentation below for details).
///
/// - Once a timer expires, it must be re-armed explicitly, which can
///   be done by calling its Set{Expiration,Duration}Ms() method directly
///   from its callback.
///
/// - Similarly, it is possible (and actually the most efficient path)
///   to start another i/o operation right from the completion callback
///   of an AsyncHandle instance.
///
/// - AsyncLoop::RunOnce() and AsyncLoop::RunUntil() are not re-entrant,
///   meaning one should not call them from completion / expiration callbacks.
///
///   On the other hand, creating / deleting AsyncHandle / AsyncTimer instances
///   within callbacks is supported.
///
/// - There is no method to cancel all asynchronous operations
///   associated with a handle, or to detach all handles. This
///   is intentional.

/// AsyncError is an error code returned by a failed asynchronous operation.
/// On Win32, this is a GetLastError() value, while on Posix this is an errno
/// value. Always 0 to indicate that there is no error.
#ifdef _WIN32
using AsyncError = DWORD;
#else
using AsyncError = int;
#endif

/// Forward declaration.
class AsyncLoop;

/// Convert AsyncError to string.
std::string AsyncErrorToString(AsyncError error);

/// Scoped AsyncHandle type that only supports one async operation
/// at a time.
class AsyncHandle {
 public:
  /// The type of a callable object that will be invoked when an
  /// asynchronous operation completes. The first argument is an error
  /// code and will be 0 in case of success. The second argument is
  /// the transfer size, and will be 0 to connect and accept operations.
  using Callback = std::function<void(AsyncError, size_t)>;

  /// Default constructor creates an empty instance.
  AsyncHandle();

  /// Destructor, closes the handle automatically.
  ~AsyncHandle();

  /// Move operations.
  AsyncHandle(AsyncHandle&&) noexcept;
  AsyncHandle& operator=(AsyncHandle&&) noexcept;

  /// Create new instance that takes ownership of |native_handle|.
  static AsyncHandle Create(IpcHandle::HandleType native_handle,
                            AsyncLoop& async_loop, Callback&& callback);

  /// Create a new instance that duplicates |native_handle| and takes
  /// ownership of the resulting handle. Note that the duplicate is
  /// not inheritable and will be in non-blocking mode on Posix.
  static AsyncHandle CreateClone(IpcHandle::HandleType native_handle,
                                 AsyncLoop& async_loop, Callback&& callback);

  /// Create new instance that takes ownership of the native handle
  /// from |handle|. Note that when |handle| is an IpcServiceHandle instance,
  /// then it should only be destroyed _after_ this instance, in order to
  /// perform proper cleanups on MacOS (i.e. remove a Unix domain socket and
  /// pid file), even though ownership of the native handle itself was passed
  /// to this AsyncHandle instance.
  static AsyncHandle Create(IpcHandle handle, AsyncLoop& async_loop,
                            Callback&& callback);

  /// Close handle
  void Close();

  /// Release native handle ownership to caller.
  /// Note that this cancels any async operation for the handle.
  IpcHandle::HandleType Release();

  /// Return true if the native handle is valid, and was not released.
  bool is_valid() const;

  /// Return value of native handle, but does not transfer ownership.
  IpcHandle::HandleType native_handle() const;

  /// Conversion to bool, equivalent to is_valid().
  operator bool() const { return is_valid(); }

  /// Return true if an asynchronous operation was started and
  /// Did not complete yet.
  bool IsRunning() const;

  /// Cancel current asynchronous operation, if any.
  void Cancel();

  /// Change the callback for this instance.
  AsyncHandle& ResetCallback(Callback&& cb);

  /// Reset instance with a new native handle, keeping the same AsyncLoop
  /// reference, and same callback. Calls Cancel() implicitly.
  void ResetHandle(IpcHandle handle);

  /// Start an asynchronous read operation. Cancels any previous operation.
  void StartRead(void* buffer, size_t size);

  /// Start an asynchronous write operation. Cancels any previous operation.
  void StartWrite(const void* buffer, size_t size);

  /// Start an asynchronous connect operation. Cancels any previous operation.
  /// Note that this instance's handle should be the result of an
  /// IpcServiceHandle::AsyncConnectTo() operation.
  void StartConnect();

  /// Start an asynchronous accept operation. Note that in case of success,
  /// the client handle should be retrieved with TakeAcceptedHandle();
  /// Cancels any previous operation.
  /// Note that this instance's handle must be the result of an
  /// IpcServiceHandle::BindTo() call.
  void StartAccept();

  /// Return the client handle corresponding to the last successful
  /// asynchronous accept operation.
  IpcHandle TakeAcceptedHandle();

  /// Return AsyncLoop pointer from this handle. Will be nullptr if
  /// handle is invalid (i.e. after an explicit Close()).
  AsyncLoop& async_loop() const;

  /// Opaque type for platform-dependent asynchronous operation.
  class State;

 protected:
  /// Private constructor, use AsyncLoop::CreateHandle() to
  /// create a new non-default instance.
  explicit AsyncHandle(std::unique_ptr<State> state);

  std::unique_ptr<State> state_;
};

class AsyncTimer {
 public:
  using Callback = std::function<void(void)>;

  // Default constructor.
  AsyncTimer();

  // Constructor.
  AsyncTimer(AsyncLoop& async_loop, Callback&& callback);

  /// Destructor. Cancels the timer automatically.
  ~AsyncTimer();

  /// Move operations are allowed.
  AsyncTimer(AsyncTimer&&) noexcept;
  AsyncTimer& operator=(AsyncTimer&&) noexcept;

  /// Return true if this instance is valid (i.e. not default constructed).
  bool is_valid() const { return !!state_; }
  operator bool() const { return is_valid(); }

  /// Set the expiration time for this timer. After this the callback
  /// will be invoked once.
  void SetExpirationMs(int64_t expiration_ms);

  /// Set a duration after which the timer will expire.
  void SetDurationMs(int64_t duration_ms);

  /// Cancel this timer (remove any expiration).
  void Cancel();

  /// Close the timer before destruction (makes it invalid).
  void Close();

  /// Reset the callback for this instance.
  void ResetCallback(Callback&& callback);

  /// Construct a new timer and set its expiration time directly.
  static AsyncTimer CreateWithExpiration(int64_t expiration_ms,
                                         AsyncLoop& async_loop,
                                         Callback&& callback);

  /// Construct a new timer and set its duration period directly.
  static AsyncTimer CreateWithDuration(int64_t duration_ms,
                                       AsyncLoop& async_loop,
                                       Callback&& callback);

  /// Retrieve reference to AsyncLoop this timer belongs to.
  /// NOTE: It is a runtime error to call this on an invalid instance.
  AsyncLoop& async_loop() const;

  class State;

 private:
  std::unique_ptr<State> state_;
};

/// A class used to create asynchronous operations and wait for their
/// completion with a possible timeout. This also detect user interruptions
/// through Ctrl-C / SIGINT / SIGTERM / SIGHUP.
///
class AsyncLoop {
 public:
  /// Destructor
  ~AsyncLoop();

  /// Retrieve reference to global instance. Created on demand.
  /// This instance always catches interrupts by default.
  static AsyncLoop& Get();

  /// Create a new instance independent from the global one. It does _not_
  /// catches interrupts by default. This instance will have its own set of
  /// timer and async IDs.
  static std::unique_ptr<AsyncLoop> CreateLocal();

  /// Destroy current global instance. Only use this for testing.
  static void ResetForTesting();

#ifdef _WIN32
  using NativeHandle = HANDLE;
#else
  using NativeHandle = int;
#endif

  /// Each asynchonous operation is identified by a unique, non-zero, ID.
  using AsyncId = uint32_t;

  /// Return current time in milliseconds. Epoch is undetermined
  /// but all values are guaranteed to be non-negative.
  static int64_t NowMs();

  /// Possible return values for RunOnce() method.
  enum ExitStatus {
    ExitIdle = 0,     // no more async ops or timers to wait for.
    ExitSuccess,      // at least one async op completed or timer expired.
    ExitTimeout,      // timeout expired.
    ExitInterrupted,  // user interruption detected.
  };

  /// Run the loop once with a timeout. This function only returns after
  /// a least one async operation / timer has completed, or on expiration,
  /// or a user interrupt. A negative |timeout_ms| value corresponds to no
  /// timeout at all, but the function may return with ExitIdle if there
  /// are no more async operations registered. Note that ExitIdle
  /// _cannot_ be returned if the value of |timeout_ms| is not negative.
  /// Also the function cannot return ExitInterrupted if interrupt catching
  /// is not enabled for this AsyncLoop instance.
  ExitStatus RunOnce(int64_t timeout_ms);

  /// Run the loop until a given condition is met, or an interrupt is
  /// detected, or a timeout expires. This will return ExitSuccess,
  /// ExitInterrupted and ExitTimeout respectively.
  ///
  /// Also, if no timeout is specified (i.e. the value of |timeout_ms|
  /// is negative), this will return ExitIdle if there are no more
  /// registered async operations or active timers.
  ///
  /// |condition| is a predicate to check for the condition, whose
  /// value only depends on variables modified by async handlers or
  /// timers.
  ///
  template <typename PREDICATE>
  ExitStatus RunUntil(PREDICATE condition, int64_t timeout_ms = -1) {
    RunUntilState state(timeout_ms);
    do {
      if (condition())
        return ExitSuccess;
    } while (state.LoopAgain(*this));
    return state.status();
  }

  /// Clear the interrupt flag after it has been acknowledged by the caller.
  void ClearInterrupt();

#ifndef _WIN32
  /// Return the signal number after RunOnce() returns with ExitInterrupted.
  int GetInterruptSignal() const;

  /// Return the process signal mask that was set before this instance was
  /// created (since the constructor blocks all signals).
  sigset_t GetOldSignalMask() const;
#endif

  /// Completely reset the instance, forgetting about all registered
  /// async operations and timers, only the interrupt catcher count is
  /// preserved.
  void Reset();

  /// Convenience struct that allows to temporarily redirect interrupts
  /// (i.e. Ctrl-C on Windows, and SIGINT/SIGHUP/SIGTERM on Posix) to
  /// a given AsyncLoop instance. Note that only one AsyncLoop instance can
  /// catch them at any given point in the process.
  ///
  /// Usage is simply:
  ///
  ///   {
  ///     AsyncLoop::ScopedInterruptCatcher catcher(async_loop);
  ///
  ///     ... all interrupts redirect to |async_loop| until the
  ///     ... destructor is called.
  ///   }
  ///
  /// It is safe to nest multiple catchers, which can even refer to the
  /// same async_loop. Only the most-nested instance will get the interrupts.
  ///
  struct ScopedInterruptCatcher {
    explicit ScopedInterruptCatcher(AsyncLoop& async_loop)
        : async_loop_(async_loop) {
      async_loop_.ChangeInterruptCatcher(true);
    }

    ~ScopedInterruptCatcher() { async_loop_.ChangeInterruptCatcher(false); }

    AsyncLoop& async_loop_;
  };

 private:
  friend class AsyncHandle::State;
  friend class AsyncTimer::State;
  friend class AsyncLoopTimers;

  AsyncLoop();

  // Used internally to implement AsyncHandles.
  void AttachHandle(AsyncHandle::State* state);
  void DetachHandle(AsyncHandle::State* state);
  void UpdateHandle(AsyncHandle::State* state);
  void CancelHandle(AsyncHandle::State* state);

  void AttachTimer(AsyncTimer::State* state);
  void DetachTimer(AsyncTimer::State* state);
  void UpdateTimer(AsyncTimer::State* state);

  /// Internal struct used by RunUntil() template instantiations.
  struct RunUntilState {
    /// Constructor sets up the state.
    RunUntilState(int64_t timeout_ms);

    /// Call this method to call RunOnce() to drain events or
    /// timers. Returns true to indicate that the function
    /// must be called again, and false in case of exit
    /// condition (reported by status()).
    bool LoopAgain(AsyncLoop& async_loop);

    /// Return final status, only valid after LoopAgain()
    /// returns false. Cannot be ExitSuccess.
    ExitStatus status() const { return status_; }

    int64_t timeout_ms_;
    int64_t expiration_ms_ = -1;
    ExitStatus status_ = ExitIdle;
  };

  /// Used internally by ScopedInterruptCatcher class.
  void ChangeInterruptCatcher(bool increment);

  /// Implementation details hidden from this header intentionally.
  class Impl;
  std::unique_ptr<Impl> impl_;
  int interrupt_catcher_count_ = 0;
};

#endif  // NINJA_ASYNC_LOOP_H