| 535 | } |
| 536 | |
| 537 | void run_until_stopped() |
| 538 | { |
| 539 | bool expected_running = false; |
| 540 | // Only one thread of execution is allowed to drive the io context. |
| 541 | if (!__is_running_.compare_exchange_strong(expected_running, |
| 542 | true, |
| 543 | STDEXEC::__std::memory_order_relaxed)) |
| 544 | { |
| 545 | STDEXEC_THROW(std::runtime_error("exec::io_uring_context::run() called on a running " |
| 546 | "context")); |
| 547 | } |
| 548 | else |
| 549 | { |
| 550 | // Check whether we restart the context after a context-wide stop. |
| 551 | // We have to reset the stop source in this case. |
| 552 | int __in_flight = __n_submissions_in_flight_.load(STDEXEC::__std::memory_order_relaxed); |
| 553 | if (__in_flight == __no_new_submissions) |
| 554 | { |
| 555 | __stop_source_.emplace(); |
| 556 | // Make emplacement of stop source visible to other threads and open the door for new submissions. |
| 557 | __n_submissions_in_flight_.store(0, STDEXEC::__std::memory_order_release); |
| 558 | } |
| 559 | else |
| 560 | { |
| 561 | // This can only happen for the very first pass of run_until_stopped() |
| 562 | __wakeup_operation_.start(); |
| 563 | } |
| 564 | } |
| 565 | scope_guard __not_running{ |
| 566 | [&]() noexcept { __is_running_.store(false, STDEXEC::__std::memory_order_relaxed); }}; |
| 567 | __pending_.append(__requests_.pop_all_reversed()); |
| 568 | while (__n_total_submitted_ > 0 || !__pending_.empty()) |
| 569 | { |
| 570 | run_some(); |
| 571 | if (__n_total_submitted_ == 0 |
| 572 | || (__n_total_submitted_ == 1 |
| 573 | && __break_loop_.load(STDEXEC::__std::memory_order_acquire))) |
| 574 | { |
| 575 | __break_loop_.store(false, STDEXEC::__std::memory_order_relaxed); |
| 576 | break; |
| 577 | } |
| 578 | constexpr int __min_complete = 1; |
| 579 | STDEXEC_ASSERT(0 <= __n_total_submitted_ |
| 580 | && std::cmp_less_equal(__n_total_submitted_, __params_.cq_entries)); |
| 581 | int rc = __io_uring_enter(__ring_fd_, |
| 582 | static_cast<unsigned>(__n_newly_submitted_), |
| 583 | __min_complete, |
| 584 | IORING_ENTER_GETEVENTS); |
| 585 | __throw_error_code_if(rc < 0 && rc != -EINTR, -rc); |
| 586 | if (rc != -EINTR) |
| 587 | { |
| 588 | STDEXEC_ASSERT(rc <= __n_newly_submitted_); |
| 589 | __n_newly_submitted_ -= rc; |
| 590 | } |
| 591 | __n_total_submitted_ -= __completion_queue_.complete(); |
| 592 | STDEXEC_ASSERT(0 <= __n_total_submitted_); |
| 593 | __pending_.append(__requests_.pop_all_reversed()); |
| 594 | } |
no test coverage detected