io_uring.cpp

/*
 * Copyright (c) 2023 Maikel Nadolski
 * Copyright (c) 2023 NVIDIA Corporation
 *
 * Licensed under the Apache License Version 2.0 with LLVM Exceptions
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *   https://llvm.org/LICENSE.txt
 *
 * 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.
 */

#include "exec/linux/io_uring_context.hpp"

#include "exec/when_any.hpp"
#include "exec/finally.hpp"

#include "stdexec/execution.hpp"

#include <chrono>
#include <thread>

#include <iostream>

auto main() -> int {
  exec::io_uring_context context;
  exec::io_uring_context context2;
  std::thread io_thread{[&] {
    context.run_until_stopped();
  }};
  std::thread io_thread2{[&] {
    context2.run_until_stopped();
  }};
  auto scheduler = context.get_scheduler();
  auto scheduler2 = context2.get_scheduler();
  using namespace std::chrono_literals;

  stdexec::sync_wait(
    exec::when_any(
      exec::schedule_after(scheduler, 1s) | stdexec::then([] { std::cout << "Hello, 1!\n"; }),
      exec::schedule_after(scheduler2, 2s) | stdexec::then([] { std::cout << "Hello, 2!\n"; })
        | stdexec::upon_stopped([] { std::cout << "Hello, 2, stopped.\n"; })));

  stdexec::sync_wait(
    exec::when_any(
      exec::schedule_after(scheduler, 1s) | stdexec::then([] { std::cout << "Hello, 1!\n"; })
        | stdexec::upon_stopped([] { std::cout << "Hello, 1, stopped.\n"; }),
      exec::schedule_after(scheduler2, 500ms) | stdexec::then([] { std::cout << "Hello, 2!\n"; })
        | stdexec::upon_stopped([] { std::cout << "Hello, 2, stopped.\n"; })));

  stdexec::sync_wait(
    stdexec::when_all(
      stdexec::schedule(scheduler) | stdexec::then([] { std::cout << "Hello, 0!\n"; }),
      exec::schedule_after(scheduler, 1s) | stdexec::then([] { std::cout << "Hello, 1!\n"; }),
      exec::schedule_after(scheduler2, 2s) | stdexec::then([] { std::cout << "Hello, 2!\n"; }),
      exec::schedule_after(scheduler, 3s) | stdexec::then([] { std::cout << "Stop it!\n"; }),
      exec::finally(exec::schedule_after(scheduler2, 4s), stdexec::just() | stdexec::then([&] {
                                                            context.request_stop();
                                                          })),
      exec::finally(exec::schedule_after(scheduler, 4s), stdexec::just() | stdexec::then([&] {
                                                           context2.request_stop();
                                                         })),
      exec::schedule_after(scheduler, 10s)    //
        | stdexec::then([] {                  //
            std::cout << "Hello, world!\n";   //
          })                                  //
        | stdexec::upon_stopped([] {          //
            std::cout << "Hello, stopped.\n"; //
          }),                                 //
      exec::schedule_after(scheduler2, 10s)   //
        | stdexec::then([] {                  //
            std::cout << "Hello, world!\n";   //
          })                                  //
        | stdexec::upon_stopped([] {          //
            std::cout << "Hello, stopped.\n"; //
          })));                               //
  io_thread.join();
  io_thread2.join();

  stdexec::sync_wait(
    stdexec::schedule(scheduler)
    | stdexec::then([] { std::cout << "This should not print, because the context is stopped.\n"; })
    | stdexec::upon_stopped([] { std::cout << "The context is stopped!\n"; }));

  stdexec::sync_wait(
    stdexec::schedule(scheduler2)
    | stdexec::then([] { std::cout << "This should not print, because the context is stopped.\n"; })
    | stdexec::upon_stopped([] { std::cout << "The context is stopped!\n"; }));

  context.reset();
  io_thread = std::thread{[&] {
    context.run_until_stopped();
  }};

  while (!context.is_running())
    ;
  stdexec::sync_wait(
    exec::when_any(
      exec::schedule_after(scheduler, 1s) | stdexec::then([] { std::cout << "Hello, 1!\n"; }),
      exec::schedule_after(scheduler, 500ms) | stdexec::then([] { std::cout << "Hello, 2!\n"; })));

  auto time_point = std::chrono::steady_clock::now() + 1s;
  stdexec::sync_wait(exec::schedule_at(scheduler, time_point) | stdexec::then([] {
                       std::cout << "Hello, schedule_at!\n";
                     }));

  static_assert(exec::timed_scheduler<exec::io_uring_scheduler>);

  context.request_stop();
  io_thread.join();
}