poll_multiplexer.hpp

Go to the documentation of this file.

/* Copyright 2025 Kevin Exton
 *
 * 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.
 */
 
#pragma once
#ifndef IO_POLL_MULTIPLEXER_HPP
#define IO_POLL_MULTIPLEXER_HPP
#include "detail/execution_trigger.hpp"
#include "multiplexer.hpp"
 
#include <stdexec/execution.hpp>
 
#include <deque>
#include <memory>
 
#include <poll.h>
// Forward declarations.
namespace io::socket {
class socket_handle;
} // namespace io::socket
 
namespace io::execution {
struct poll_t {
  using event_type = struct pollfd;
  using interval_type = std::chrono::milliseconds;
  using size_type = std::size_t;
  template <typename Op> struct is_eager_t : public std::false_type {};
  template <typename Op>
  static constexpr bool is_eager_v = is_eager_t<Op>::value;
};
 
template <AllocatorLike Allocator = std::allocator<char>>
class basic_poll_multiplexer : public basic_multiplexer<poll_t> {
public:
  using Base = basic_multiplexer<poll_t>;
  using mutex = std::mutex;
  using socket_handle = ::io::socket::socket_handle;
  using task = Base::intrusive_task_queue::task;
  using native_socket_type = ::io::socket::native_socket_type;
  using vector_allocator =
      std::allocator_traits<Allocator>::template rebind_alloc<event_type>;
  using vector_type = std::vector<event_type, vector_allocator>;
 
  struct demultiplexer {
    intrusive_task_queue read_queue;
    intrusive_task_queue write_queue;
 
    socket_handle *socket = nullptr;
  };
 
  using map_allocator =
      std::allocator_traits<Allocator>::template rebind_alloc<demultiplexer>;
  using map_type = std::deque<demultiplexer, map_allocator>;
 
  template <Completion Fn> struct sender {
    using sender_concept = stdexec::sender_t;
    using completion_signatures = stdexec::completion_signatures<
        stdexec::set_value_t(typename std::invoke_result_t<Fn>::value_type),
        stdexec::set_error_t(std::error_code)>;
 
    template <typename Receiver> struct state : public task {
      static auto complete(task *task_ptr) noexcept -> void;
      auto start() noexcept -> void;
 
      Fn func;
      std::shared_ptr<socket_handle> socket;
      Receiver receiver{};
      demultiplexer *demux = nullptr;
      mutex *mtx = nullptr;
      execution_trigger trigger{};
    };
 
    template <typename Receiver>
    auto connect(Receiver &&receiver) -> state<std::decay_t<Receiver>>;
 
    Fn func;
    std::shared_ptr<socket_handle> socket;
    map_type *demux = nullptr;
    vector_type *list = nullptr;
    mutex *mtx = nullptr;
    execution_trigger trigger{};
  };
 
  auto wait_for(interval_type interval) -> size_type;
 
  template <Completion Fn>
  auto set(std::shared_ptr<socket_handle> socket, execution_trigger trigger,
           Fn &&func) -> sender<std::decay_t<Fn>>;
 
  constexpr basic_poll_multiplexer(
      const Allocator &alloc = Allocator()) noexcept(noexcept(Allocator()));
 
private:
  map_type demux_;
  vector_type list_;
  mutable mutex mtx_;
};
 
using poll_multiplexer = basic_poll_multiplexer<>;
 
} // namespace io::execution
 
#include "io/execution/impl/poll_multiplexer_impl.hpp" // IWYU pragma: export
 
#endif // IO_POLL_MULTIPLEXER_HPP