API reference¶

#include <pika/execution.hpp>
#include <pika/init.hpp>

#include <fmt/printf.h>

#include <utility>

int main(int argc, char* argv[])
{
    namespace ex = pika::execution::experimental;
    namespace tt = pika::this_thread::experimental;

    pika::start(argc, argv);

    // The pika runtime is now active and we can schedule work on the default
    // thread pool
    auto s = ex::schedule(ex::thread_pool_scheduler{}) |
        ex::then([]() { fmt::print("Hello from the pika runtime\n"); });
    tt::sync_wait(std::move(s));

    pika::finalize();
    pika::stop();

    return 0;
}

#include <pika/execution.hpp>
#include <pika/init.hpp>

#include <fmt/printf.h>

#include <tuple>
#include <utility>

struct custom_type
{
};

int main(int argc, char* argv[])
{
    namespace ex = pika::execution::experimental;
    namespace tt = pika::this_thread::experimental;

    pika::start(argc, argv);
    ex::thread_pool_scheduler sched{};

    auto s = ex::just(42, custom_type{}, std::tuple("hello")) |
        ex::drop_value() |
        // No matter what is sent to drop_value, it won't be sent from
        // drop_value
        ex::then([] { fmt::print("I got nothing...\n"); });
    tt::sync_wait(std::move(s));

    pika::finalize();
    pika::stop();

    return 0;
}

#include <pika/execution.hpp>
#include <pika/init.hpp>

#include <cassert>
#include <memory>
#include <utility>

int main(int argc, char* argv[])
{
    namespace ex = pika::execution::experimental;
    namespace tt = pika::this_thread::experimental;

    pika::start(argc, argv);
    ex::thread_pool_scheduler sched{};

    auto sp = std::make_shared<int>(42);
    std::weak_ptr<int> sp_weak = sp;

    auto s = ex::just(std::move(sp)) |
        ex::then([&](auto&&) { assert(sp_weak.use_count() == 1); }) |
        // Even though the shared_ptr is no longer in use, it may be kept alive
        // by the operation state
        ex::then([&]() {
            assert(sp_weak.use_count() == 1);
            return 42;
        }) |
        ex::drop_operation_state() |
        // Once drop_operation_state has been used, the shared_ptr is guaranteed
        // to be released.  Values are passed through the adaptor.
        ex::then([&]([[maybe_unused]] int x) {
            assert(sp_weak.use_count() == 0);
            assert(x == 42);
        });
    tt::sync_wait(std::move(s));

    pika::finalize();
    pika::stop();

    return 0;
}

#include <pika/execution.hpp>
#include <pika/init.hpp>

#include <fmt/printf.h>

#include <cassert>
#include <cstdlib>
#include <exception>
#include <utility>

int main(int argc, char* argv[])
{
    namespace ex = pika::execution::experimental;
    namespace tt = pika::this_thread::experimental;

    pika::start(argc, argv);
    ex::thread_pool_scheduler sched{};

    {
        // require_started forwards values received from the predecessor sender
        auto s = ex::just(42) | ex::require_started() |
            ex::then([]([[maybe_unused]] auto&& i) { assert(i == 42); });
        tt::sync_wait(std::move(s));
    }

    {
        // The termination is ignored with discard, the sender is from the
        // user's perspective rightfully not used
        auto s = ex::just() | ex::require_started();
        s.discard();
    }

    {
        // The require_started sender terminates on destruction if it has not
        // been used
        auto s = ex::just() | ex::require_started();
    }
    assert(false);

    pika::finalize();
    pika::stop();
    return 0;
}

#include <pika/execution.hpp>
#include <pika/init.hpp>

#include <fmt/printf.h>

#include <chrono>
#include <thread>
#include <tuple>
#include <utility>

int main(int argc, char* argv[])
{
    namespace ex = pika::execution::experimental;
    namespace tt = pika::this_thread::experimental;

    pika::start(argc, argv);
    ex::thread_pool_scheduler sched{};

    // split_tuple can be used to process the result and its square through
    // senders, without having to pass both around together
    auto [snd, snd_squared] = ex::schedule(sched) |
        ex::then([]() { return 42; }) |
        ex::then([](int x) { return std::tuple(x, x * x); }) |
        ex::split_tuple();

    // snd and snd_squared will be ready at the same time, but can be used
    // independently
    auto snd_print = std::move(snd) | ex::continues_on(sched) |
        ex::then([](int x) { fmt::print("x is {}\n", x); });
    auto snd_process = std::move(snd_squared) | ex::continues_on(sched) |
        ex::then([](int x_squared) {
            fmt::print("Performing expensive operations on x * x\n");
            std::this_thread::sleep_for(std::chrono::milliseconds(300));
            return x_squared / 2;
        });

    auto x_squared_processed = tt::sync_wait(
        ex::when_all(std::move(snd_print), std::move(snd_process)));
    fmt::print("The final result is {}\n", x_squared_processed);

    pika::finalize();
    pika::stop();

    return 0;
}

#include <pika/execution.hpp>
#include <pika/init.hpp>

#include <fmt/printf.h>

#include <string>
#include <tuple>
#include <utility>

int main(int argc, char* argv[])
{
    namespace ex = pika::execution::experimental;
    namespace tt = pika::this_thread::experimental;

    pika::start(argc, argv);
    ex::thread_pool_scheduler sched{};

    auto tuple_sender = ex::just(std::tuple(std::string("hello!"), 42)) |
        ex::continues_on(sched);
    auto process_data = [](auto message, auto answer) {
        fmt::print("{}\nthe answer is: {}\n", message, answer);
    };

    // With the unpack adaptor, process_data does not have to know that the data
    // was originally sent as a tuple
    auto unpack_sender = tuple_sender | ex::unpack() | ex::then(process_data);

    // We can manually recreate the behaviour of the unpack adaptor by using
    // std::apply. This is equivalent to the above.
    auto apply_sender = tuple_sender | ex::then([&](auto tuple_of_data) {
        return std::apply(process_data, std::move(tuple_of_data));
    });

    tt::sync_wait(
        ex::when_all(std::move(unpack_sender), std::move(apply_sender)));

    pika::finalize();
    pika::stop();

    return 0;
}

#include <pika/execution.hpp>
#include <pika/init.hpp>

#include <fmt/printf.h>
#include <fmt/ranges.h>

#include <cstddef>
#include <random>
#include <utility>
#include <vector>

std::size_t get_n() { return 13; }
std::size_t calculate(std::size_t i) { return (std::rand() % 4) * i * i; }

int main(int argc, char* argv[])
{
    namespace ex = pika::execution::experimental;
    namespace tt = pika::this_thread::experimental;

    pika::start(argc, argv);
    ex::thread_pool_scheduler sched{};

    // when_all_vector is like when_all, but for a dynamic number of senders
    // through a vector of senders
    auto const n = get_n();
    std::vector<ex::unique_any_sender<std::size_t>> snds;
    snds.reserve(n);
    for (std::size_t i = 0; i < n; ++i)
    {
        snds.push_back(
            ex::just(i) | ex::continues_on(sched) | ex::then(calculate));
    }
    auto snds_print = ex::when_all_vector(std::move(snds)) |
        ex::then([](std::vector<std::size_t> results) {
            fmt::print("Results are: {}\n", fmt::join(results, ", "));
        });
    tt::sync_wait(std::move(snds_print));

    // when_all_vector will send no value on completion if the input vector
    // contains senders sending no value
    std::vector<ex::unique_any_sender<>> snds_nothing;
    snds_nothing.reserve(n);
    for (std::size_t i = 0; i < n; ++i)
    {
        snds_nothing.push_back(ex::just(i) | ex::continues_on(sched) |
            ex::then([](auto i) { fmt::print("{}: {}\n", i, calculate(i)); }));
    }
    auto snds_nothing_done = ex::when_all_vector(std::move(snds_nothing)) |
        ex::then([]() { fmt::print("Done printing all results\n"); });
    tt::sync_wait(std::move(snds_nothing_done));

    pika::finalize();
    pika::stop();

    return 0;
}