select.hpp

Go to the documentation of this file.

/*
 
Copyright 2025 Matthew Tolman
 
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 MTCORE_THREAD_SELECT_HPP
#define MTCORE_THREAD_SELECT_HPP
 
#include "mtcore_thread/core.hpp"
#include "mtcore_thread/traits.hpp"
#include <chrono>
 
namespace mtcore::thread {
  enum class SelectErrors { SELECT_TIMEOUT, DEADLOCK_DETECTED, NOT_READY };

  enum class SelectOptionErrors {
    ATTEMPT_FAILED,
    SKIP,
    TIMEOUT,
  };

  template<SelectOption... Options>
  struct Select;

  template<SelectOption... Options>
  Result<void, SelectErrors> select(Options... opts);

  template<Sender S>
  struct SendOptionNoCallback {
    using Message = typename S::Message;
    S &channel;
    Message m;
    bool valid = true;
 
    Result<void, SelectOptionErrors> attempt() {
      auto res = channel.try_send(m);
      if (res.is_success()) {
        return success();
      }
      auto err = res.error().code;
      if constexpr (has_closed_error<std::remove_const_t<decltype(err)>>) {
        if (err == decltype(err)::CHANNEL_CLOSED) {
          valid = false;
          return error(SelectOptionErrors::SKIP);
        }
      }
      return error(SelectOptionErrors::ATTEMPT_FAILED);
    }

    [[nodiscard]] bool prevents_deadlock() const noexcept { return valid; }
  };
  static_assert(is_select_option<SendOptionNoCallback<Channel<int>>>, "SendOptionNoCallback is invalid!");

  template<Sender S, typename Callback>
  struct SendOption {
    using Message = typename S::Message;
    S &channel;
    Message m;
    Callback callback;
    bool valid = true;
 
    Result<void, SelectOptionErrors> attempt() {
      auto res = channel.try_send(m);
      if (res.is_success()) {
        callback();
        return success();
      }
      auto err = res.error().code;
      if constexpr (has_closed_error<std::remove_const_t<decltype(err)>>) {
        if (err == decltype(err)::CHANNEL_CLOSED) {
          valid = false;
          return error(SelectOptionErrors::SKIP);
        }
      }
      return error(SelectOptionErrors::ATTEMPT_FAILED);
    }

    [[nodiscard]] bool prevents_deadlock() const noexcept { return valid; }
  };
  static_assert(is_select_option<SendOption<Channel<int>, void (*)(void)>>, "SendOption is invalid!");

  template<Sender S, typename Callback>
  SendOption<S, Callback> send_to(S &to, typename S::Message msg, const Callback &callback) {
    return SendOption<S, Callback>{to, msg, callback};
  }

  template<Sender S>
  SendOptionNoCallback<S> send_to(S &to, typename S::Message msg) {
    return SendOptionNoCallback<S>{to, msg};
  }

  template<typename Clock, typename Duration, typename Callback>
  struct BeforeOption {
    std::chrono::time_point<Clock, Duration> time_point;
    Callback callback;
 
    Result<void, SelectOptionErrors> attempt() {
      if (time_point <= Clock::now()) {
        callback();
        return error(SelectOptionErrors::TIMEOUT);
      }
      return error(SelectOptionErrors::ATTEMPT_FAILED);
    }
 
    // No need to wait for us if everything else is no longer valid
    [[nodiscard]] bool prevents_deadlock() const noexcept { return false; }
  };
  static_assert(is_select_option<BeforeOption<std::chrono::system_clock, std::chrono::nanoseconds, void (*)(void)>>,
                "BeforeOption is invalid!");

  template<typename Clock, typename Duration>
  struct BeforeOptionNoCallback {
    std::chrono::time_point<Clock, Duration> time_point;
 
    Result<void, SelectOptionErrors> attempt() {
      if (time_point <= Clock::now()) {
        return error(SelectOptionErrors::TIMEOUT);
      }
      return error(SelectOptionErrors::ATTEMPT_FAILED);
    }
 
    // No need to wait for us if everything else is no longer valid
    [[nodiscard]] bool prevents_deadlock() const noexcept { return false; }
  };
  static_assert(is_select_option<BeforeOptionNoCallback<std::chrono::system_clock, std::chrono::nanoseconds>>,
                "BeforeOptionNoCallback is invalid!");

  template<typename Clock, typename Duration, typename Callback>
  BeforeOption<Clock, Duration, Callback> before(const std::chrono::time_point<Clock, Duration> &tp,
                                                 const Callback &callback) {
    return BeforeOption<Clock, Duration, Callback>{tp, callback};
  }

  template<typename Clock, typename Duration>
  BeforeOptionNoCallback<Clock, Duration> before(const std::chrono::time_point<Clock, Duration> &tp) {
    return BeforeOptionNoCallback<Clock, Duration>{tp};
  }

  template<typename Rep, typename Per, typename Callback>
  auto timeout(const std::chrono::duration<Rep, Per> &tp, const Callback &callback) {
    return before(std::chrono::steady_clock::now() + tp, callback);
  }

  template<typename Rep, typename Per>
  auto timeout(const std::chrono::duration<Rep, Per> &tp) {
    return before(std::chrono::steady_clock::now() + tp);
  }

  template<Receiver R, typename Callback>
  struct ReceiveOption {
    using Message = typename R::Message;
    R &channel;
    Callback callback;
 
    Result<void, SelectOptionErrors> attempt() {
      auto res = channel.try_receive();
      if (res.is_success()) {
        callback(res.value());
        return success();
      }
      return error(SelectOptionErrors::ATTEMPT_FAILED);
    }
 
    [[nodiscard]] bool prevents_deadlock() const noexcept { return true; }
  };
  static_assert(is_select_option<ReceiveOption<Channel<int>, void (*)(int)>>, "ReceiveOption is invalid!");

  template<Receiver R, typename Callback>
  ReceiveOption<R, Callback> receive_from(R &from, const Callback &callback) {
    return ReceiveOption<R, Callback>{from, callback};
  }
 
  namespace impl {
    template<SelectOption... Options>
    Result<void, SelectErrors> exec(Select<Options...> &options);
 
    template<typename T>
    concept SelectImpl = requires(T &t, const T &ct) {
      { t.run_step() } -> std::same_as<Result<void, SelectErrors>>;
      { ct.prevents_deadlock() } -> std::same_as<bool>;
      { ct.skip() } -> std::same_as<bool>;
      { t() } -> std::same_as<Result<void, SelectErrors>>;
    };
 
    template<typename T>
    struct IsSelectImpl;
 
    template<SelectImpl T>
    struct IsSelectImpl<T> : std::true_type {};
 
    template<typename T>
    struct IsSelectImpl : std::false_type {};
 
    template<typename T>
    constexpr bool is_select_impl = IsSelectImpl<T>::value;
  }  // namespace impl

  template<>
  struct Select<> {
    Result<void, SelectErrors> run_step() { return error(SelectErrors::NOT_READY); }
    [[nodiscard]] Result<void, SelectErrors> operator()() { return impl::exec(*this); }
 
    [[nodiscard]] bool prevents_deadlock() const noexcept { return false; }
    [[nodiscard]] bool skip() const noexcept { return false; }
  };
  static_assert(impl::is_select_impl<Select<>>, "Invalid select specialization");

  template<SelectOption CurOption, SelectOption... Options>
  struct Select<CurOption, Options...> {
    CurOption val;
    bool shouldSkip = false;
    Select<Options...> next;
 
    Select(CurOption val, Options... vals) : val(val), next({vals...}) {}
 
    Result<void, SelectErrors> run_step() {
      if (!skip()) {
        auto res = val.attempt();
        if (res.is_success()) {
          return success();
        }
        SelectOptionErrors err = res.error().code;
        if (err == SelectOptionErrors::TIMEOUT) {
          return error(SelectErrors::SELECT_TIMEOUT);
        }
        if (err == SelectOptionErrors::SKIP) {
          shouldSkip = true;
        }
      }
      return next.run_step();
    }
 
    [[nodiscard]] bool prevents_deadlock() const noexcept {
      return (val.prevents_deadlock() && !skip()) || next.prevents_deadlock();
    }
 
    [[nodiscard]] bool skip() const noexcept { return shouldSkip; }
 
    [[nodiscard]] Result<void, SelectErrors> operator()() { return impl::exec(*this); }
  };
  static_assert(impl::is_select_impl<Select<SendOptionNoCallback<Channel<int>>>>, "Invalid select specialization");
 
  static_assert(impl::is_select_impl<Select<SendOption<Channel<int>, void (*)(void)>>>,
                "Invalid select specialization");
 
  static_assert(impl::is_select_impl<Select<ReceiveOption<Channel<int>, void (*)(int)>>>,
                "Invalid select specialization");
 
  static_assert(
    impl::is_select_impl<Select<BeforeOption<std::chrono::system_clock, std::chrono::nanoseconds, void (*)(void)>>>,
    "Invalid select specialization");
 
  static_assert(
    impl::is_select_impl<Select<BeforeOptionNoCallback<std::chrono::system_clock, std::chrono::nanoseconds>>>,
    "Invalid select specialization");
 
  template<SelectOption... Options>
  [[nodiscard]] Result<void, SelectErrors> select(Options... opts) {
    return Select<Options...>{opts...}();
  }
 
  namespace impl {
    template<SelectOption... Options>
    Result<void, SelectErrors> exec(Select<Options...> &sel) {
      while (sel.prevents_deadlock()) {
        auto res = sel.run_step();
        if (res.is_error()) {
          if (res.error().code == SelectErrors::SELECT_TIMEOUT) {
            return error(SelectErrors::SELECT_TIMEOUT);
          }
          if (res.error().code == SelectErrors::DEADLOCK_DETECTED) {
            return error(SelectErrors::DEADLOCK_DETECTED);
          }
          // let something else run while we wait
          std::this_thread::yield();
          continue;
        }
        else {
          return success();
        }
      }
      return error(SelectErrors::DEADLOCK_DETECTED);
    }
  }  // namespace impl
}  // namespace mtcore::thread
 
#endif  // MTCORE_THREAD_SELECT_HPP