std::execution是如何工作的

int main() {
    // pool is a context. A context manages some resources, like CPU, GPU, or something else.
    exec::static_thread_pool pool;

    // Acquire a scheduler from thread pool. Scheduler is like a handle to the context.
    stdexec::scheduler auto sched = pool.get_scheduler();

    // Create a sender from scheduler.
    stdexec::sender auto begin = stdexec::schedule(sched);
    stdexec::sender auto lazy_task = stdexec::then(begin, [] { std::println("Hello, world!"); });

    // Busy waiting for the lazy task to be completed.
    //
    // stdexec::sync_wait is a sender consumer. It means that, this is the end of the lazy task. Usually, such sender
    // consumers create a new receiver and simply pass them to executor::connect() to combine them with senders. We will
    // talk about it in the future.
    stdexec::sync_wait(lazy_task);
    return 0;
}

template <sender_in<__env> _Sender>
auto apply_sender(_Sender &&sndr) const -> std::optional<...> {
    // Prepare for something. Let's ignore it for now.

    // Create operation state from sender and receiver.
    auto __op = stdexec::connect(sndr, sync_wait_receiver);
    // Start up the lazy task!
    stdexec::start(__op);
    // Busy wait.
    __local_state.__loop_.run();

    // Error handling and return value. Let's ignore it for now.
}

struct custom_sender {
    using sender_concept = stdexec::sender_t;

    template <std::receiver Receiver>
    auto connect(Receiver &&rcvr) -> operation_state {
        ... // Construct operation state with receiver, and return it.
    }
};

struct custom_operation_state {
    using operation_state_concept = stdexec::operation_state_t;

    auto start() & noexcept -> void {
        ...
    }
};

struct custom_receiver {
    using receiver_concept = stdexec::receiver_t;

    // Called if async task completed successfully.
    auto set_value(return_value) && noexcept -> void {
        ...
    }

    // Called if async task completed with error.
    auto set_error(error_value) && noexcept -> void {
        ...
    }

    // Called if async task is cancelled.
    auto set_stopped() && noexcept -> void {
        ...
    }
};

template <stdexec::receiver R>
class hello_operation_state {
public:
    using operation_state_concept = stdexec::operation_state_t;
    hello_operation_state(R &&next) : m_rcvr{std::move(next)} {}

    auto start() & noexcept -> void {
        std::println("Hello, world!");
        stdexec::set_value(std::move(m_rcvr));
    }

private:
    R m_rcvr;
};

struct hello_sender {
    using sender_concept        = stdexec::sender_t;
    using completion_signatures = stdexec::completion_signatures<stdexec::set_value_t()>;

    template <stdexec::receiver R>
        requires(stdexec::receiver_of<R, completion_signatures>)
    auto connect(R &&rcvr) const -> hello_operation_state<R> {
        return hello_operation_state<R>{std::move(rcvr)};
    }
};

struct my_receiver {
    using receiver_concept = stdexec::receiver_t;

    auto set_value() && noexcept -> void {
        std::println("my_receiver set_value() is called!");
    }
};

auto main() -> int {
    stdexec::sender auto          hello = hello_sender{};
    stdexec::operation_state auto op    = stdexec::connect(hello, my_receiver{});
    stdexec::start(op);
    return 0;
}

stdexec::then(begin, [] { std::println("Hello, world!"); });

template <stdexec::receiver Receiver, typename Func>
class then_receiver : public stdexec::receiver_adaptor<then_receiver<Receiver, Func>, Receiver> {
public:
    template <typename... Args>
    auto set_value(Args &&...args) && noexcept -> void {
        // The previous sender would invoke `set_value` once it is done. Therefore, we could change the receiver's
        // behavior by simply wrapping it.
        try {
            // auto result = m_func(args...);
            // set_value(base(), result);
            stdexec::set_value(std::move(*this).base(), std::invoke(std::move(m_func), std::forward<Args>(args)...));
        } catch (...) {
            stdexec::set_error(std::move(*this).base(), std::current_exception());
        }
    }

private:
    Func m_func;
};

template <stdexec::sender PrevSender, typename Func>
class then_sender {
public:
    using sender_concept = stdexec::sender_t;

    then_sender(PrevSender s, Func f) : m_prev{std::move(s)}, m_func{std::move(f)} {}

    template <stdexec::receiver Receiver>
    auto connect(Receiver r) && -> stdexec::operation_state auto {
        // Simply wrap the receiver and pass it to previous sender.
        return stdexec::connect(std::move(m_prev), then_receiver<Receiver, Func>{std::move(r), std::move(m_func)});
    }

private:
    PrevSender m_prev;
    Func       m_func;
};

template <stdexec::sender S, typename F>
auto then(S s, F f) -> stdexec::sender auto {
    return then_sender<S, F>{std::move(s), std::move(f)};
}

struct overlapped : OVERLAPPED {
    void (*on_complete)(overlapped *self, DWORD bytes, int error) noexcept;
};

while (!should_stop) {
    if (GetQueuedCompletionStatus(iocp, &bytes, &key, &ovlp, INFINITE)) {
        auto *object = static_cast<overlapped *>(ovlp);
        object->on_complete(object, bytes, GetLastError());
    }
}

class listener {
public:
    virtual void on_complete(io_uring_cqe *cqe) noexcept = 0;
};

while (!should_stop) {
    if (io_uring_wait_cqe(&ring, &cqe) == 0) {
        auto *object = static_cast<listener *>(io_uring_cqe_get_data(&cqe));
        object->on_complete(&cqe);
    }
}

struct operation_base : WSAOVERLAPPED {
    auto (*on_complete)(operation_base *self, DWORD bytes, int error) noexcept -> void;
};

template <stdexec::receiver R>
class recv_operation : public operation_base {
public:
    using operation_state_concept = stdexec::operation_state_t;

    recv_operation(R &&receiver, SOCKET socket, void *buffer, std::size_t size)
        : operation_base{},
          m_receiver{std::move(receiver), socket, this},
          m_socket{socket},
          m_buffer{static_cast<ULONG>(size), static_cast<CHAR *>(buffer)} {
        this->on_complete = &recv_operation::callback;
    }

    auto start() & noexcept -> void {
        // Prepare for asynchronous recv operation. We assume that FILE_SKIP_COMPLETION_PORT_ON_SUCCESS is already set.
        DWORD bytes  = 0;
        DWORD flags  = 0;
        int   result = WSARecv(m_socket, &m_buffer, 1, &bytes, &flags, static_cast<WSAOVERLAPPED *>(this), nullptr);
        if (result == SOCKET_ERROR) [[likely]] {
            int error = WSAGetLastError();
            if (error != WSA_IO_PENDING) { // Actual error.
                stdexec::set_error(std::move(m_receiver), std::error_code{error, std::system_category()});
            }
        } else { // Completed immediately.
            stdexec::set_value(static_cast<R &&>(m_receiver), bytes);
        }
    }

    static auto callback(operation_base *op, DWORD bytes, int error) noexcept -> void {
        recv_operation *self = static_cast<recv_operation *>(op);
        // IO operation is cancelled. We should forward the stop request.
        if (error == ERROR_OPERATION_ABORTED) {
            stdexec::set_stopped(static_cast<R &&>(self->m_receiver));
            return;
        }

        if (error == 0) {
            stdexec::set_value(static_cast<R &&>(self->m_receiver), bytes);
        } else {
            stdexec::set_error(static_cast<R &&>(self->m_receiver), std::error_code{error, std::system_category()});
        }
    }

private:
    R      m_receiver;
    SOCKET m_socket;
    WSABUF m_buffer;
};

class recv_sender {
public:
    using sender_concept        = stdexec::sender_t;
    using completion_signatures = stdexec::completion_signatures<stdexec::set_value_t(std::size_t),
                                                                 stdexec::set_error_t(std::error_code),
                                                                 stdexec::set_stopped_t()>;

    recv_sender(SOCKET socket, void *buffer, std::size_t size) : m_socket{socket}, m_buffer{buffer}, m_size{size} {}

    template <stdexec::receiver R>
        requires(stdexec::receiver_of<R, void(std::size_t)>)
    auto connect(R &&receiver) const -> recv_operation<R> {
        return recv_operation<R>{std::move(receiver), m_socket, m_buffer, m_size};
    }

private:
    SOCKET      m_socket;
    void       *m_buffer;
    std::size_t m_size;
};

stdexec::sender auto begin = stdexec::schedule(sched);

template <class ReceiverId>
class _static_thread_pool::_opstate<ReceiverId>::__t : public task_base {
    // This is constructor.
    explicit __t(...) {
        this->execute_ = [](...) {
            stdexec::set_value(static_cast<Receiver&&>(op.rcvr_));
        };
    }

    void enqueue_(task_base* op) const {
        pool_.enqueue(*queue_, op, thread_index_);
    }

public:
    void start() & noexcept {
        enqueue_(this);
    }
};

template <stdexec::scheduler Scheduler>
auto some_func(Scheduler sched) -> exec::task<void> {
    co_await stdexec::schedule(sched);
    std::println("This is coroutine function.");
    co_return;
}

auto main() -> int {
    exec::static_thread_pool pool;
    stdexec::scheduler auto sched = pool.get_scheduler();
    stdexec::sync_wait(some_func(sched));
    return 0;
}

template <stdexec::scheduler Scheduler>
auto some_func(Scheduler sched) -> exec::task<int> {
    co_await stdexec::schedule(sched);
    std::println("This is coroutine function.");
    co_return 1;
}

auto some_func1(int value) -> std::string {
    std::println("Received {} from some_func.", value);
    return "some_func1 string";
}

auto some_func2(std::string value) -> void {
    std::println("Received \"{}\" from some_func1.", value);
}

auto main() -> int {
    exec::static_thread_pool pool;
    stdexec::scheduler auto  sched = pool.get_scheduler();
    stdexec::sender auto     task  = some_func(sched) | stdexec::then(some_func1) | stdexec::then(some_func2);
    stdexec::sync_wait(std::move(task));
    return 0;
}

// This is a coroutine:
unifex::task<int> read_socket_async(socket, span<char, 1024>);

unifex::task<void> concurrent_read_async(socket s1, socket s2) {
  char buff1[1024];
  char buff2[1024];

  auto [cbytes1, cbytes2] =
      co_await std::execution::when_all( 
          read_socket_async(s1, buff1),
          read_socket_async(s2, buff2)
      ) | into_tuple();

  /*...*/
}

template <stdexec::scheduler Scheduler>
auto listener(Scheduler sched, exec::async_scope &scope, tcp_listener srv) -> exec::task<void> {
    co_await stdexec::schedule(sched);

    auto stop_token = scope.get_stop_token();
    while (!stop_token.stop_requested()) {
        tcp_stream stream = co_await srv.accept();
        scope.spawn(stdexec::schedule(sched) | handle_connection(std::move(stream)));
    }
}

auto main() -> int {
    io_context context;
    exec::async_scope scope;

    // Create tcp_listener...
    stdexec::sync_wait(listener(stdexec::schedule(sched), scope, std::move(srv)));
    return 0;
}

auto result = co_await stream.recv(buffer);
if (result.has_value()) {
    ...
}

include(cmake/CPM.cmake)
CPMAddPackage(
    NAME stdexec
    GIT_REPOSITORY https://github.com/NVIDIA/stdexec.git
    GIT_TAG main
    OPTIONS
        "STDEXEC_BUILD_TESTS OFF"
        "STDEXEC_ENABLE_EXTRA_TYPE_CHECKING OFF"
        "STDEXEC_ENABLE_CUDA OFF"
        "STDEXEC_ENABLE_TBB OFF"
        "STDEXEC_ENABLE_TASKFLOW OFF"
        "STDEXEC_ENABLE_ASIO OFF"
        "STDEXEC_ENABLE_IO_URING OFF"
        "STDEXEC_BUILD_EXAMPLES OFF"
        "STDEXEC_BUILD_TESTS OFF"
)

target_link_libraries(your-project PRIVATE stdexec)