diff --git a/CMakeLists.txt b/CMakeLists.txt index 40e0d5a..4d58656 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -71,6 +71,7 @@ file(GLOB_RECURSE BYCORF_FORMAT_FILES CONFIGURE_DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/src/*.cpp ${CMAKE_CURRENT_SOURCE_DIR}/src/*.h ${CMAKE_CURRENT_SOURCE_DIR}/tests/*.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/tests/*.h ) list(FILTER BYCORF_FORMAT_FILES EXCLUDE REGEX "/runtime/concurrentqueue\\.h$") if (BYCORF_CLANG_FORMAT) @@ -406,35 +407,9 @@ if (BYCORF_BUILD_EXAMPLES) target_link_libraries(rpc_bench PRIVATE bycorf::rpc) endif() +# Command-line probes keep their explicit environment/argument contracts. They +# are separate from automatic software regressions and never discovered by GTest. if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS) - add_executable(bycorf_received_buffer_queue_check tests/received_buffer_queue_check.cpp) - target_link_libraries(bycorf_received_buffer_queue_check PRIVATE bycorf::core) - - add_executable(bycorf_task_transfer_check tests/task_transfer_check.cpp) - target_include_directories(bycorf_task_transfer_check PRIVATE include) - # Exercise the public compile policy with this build's actual optimization - # level. Avoid linking Worker because this isolated check supplies its own - # scheduling hooks to observe Task ownership and background classification. - target_compile_options(bycorf_task_transfer_check PRIVATE - $) - - add_executable(bycorf_connect_timer_check - example/connect_timer_check.cpp - ) - target_link_libraries(bycorf_connect_timer_check PRIVATE bycorf::core) - - add_executable(bycorf_connection_storage_check example/connection_storage_check.cpp) - target_link_libraries(bycorf_connection_storage_check PRIVATE bycorf::core) - - add_executable(bycorf_tcp_buffered_read_check tests/tcp_buffered_read_check.cpp) - target_link_libraries(bycorf_tcp_buffered_read_check PRIVATE bycorf::core) - - add_executable(bycorf_io_uring_backpressure_check tests/io_uring_backpressure_check.cpp) - target_link_libraries(bycorf_io_uring_backpressure_check PRIVATE bycorf::core) - - add_executable(bycorf_backend_selection_check example/backend_selection_check.cpp) - target_link_libraries(bycorf_backend_selection_check PRIVATE bycorf::core) - add_executable(bycorf_backend_startup_check example/backend_startup_check.cpp) target_link_libraries(bycorf_backend_startup_check PRIVATE bycorf::core) @@ -452,19 +427,74 @@ if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS) endif() if (BYCORF_BUILD_TESTS) + # Reuse a parent's GoogleTest targets when available. The pinned fallback + # matches Lavik and is fetched only for test builds, never for library users. + if (NOT TARGET GTest::gtest_main) + include(FetchContent) + FetchContent_Declare(googletest + GIT_REPOSITORY https://github.com/google/googletest.git + GIT_TAG 52eb8108c5bdec04579160ae17225d66034bd723 # v1.17.0 + ) + set(BUILD_GMOCK OFF CACHE BOOL "" FORCE) + set(INSTALL_GTEST OFF CACHE BOOL "" FORCE) + FetchContent_MakeAvailable(googletest) + endif() + include(GoogleTest) + enable_testing() + + function(bycorf_discover_tests target timeout labels) + gtest_discover_tests(${target} + TEST_PREFIX "bycorf_" NO_PRETTY_TYPES NO_PRETTY_VALUES + PROPERTIES TIMEOUT ${timeout} ${ARGN}) + # GoogleTest's discovery helper flattens list-valued properties. Apply + # labels after discovery, retaining every label and keeping the generated + # CTest metadata self-contained for copied CI build bundles. + set(labels_file "${CMAKE_CURRENT_BINARY_DIR}/${target}_labels.cmake") + file(WRITE "${labels_file}" + "if(DEFINED ${target}_TESTS)\n" + " set_tests_properties(\${${target}_TESTS} PROPERTIES LABELS \"${labels}\")\n" + "endif()\n") + set_property(DIRECTORY APPEND PROPERTY TEST_INCLUDE_FILES "${labels_file}") + endfunction() + + add_executable(bycorf_unit_tests + tests/received_buffer_queue_test.cpp + tests/backend_selection_test.cpp) + target_link_libraries(bycorf_unit_tests PRIVATE bycorf::core GTest::gtest_main) + bycorf_discover_tests(bycorf_unit_tests 30 "runtime;unit") + + add_executable(bycorf_task_tests tests/task_transfer_test.cpp) + target_link_libraries(bycorf_task_tests PRIVATE GTest::gtest) + target_include_directories(bycorf_task_tests PRIVATE include) + # Use this build's real optimization level and public coroutine compile + # policy. The test supplies scheduler hooks, so it must not link Worker. + target_compile_options(bycorf_task_tests PRIVATE + $) + bycorf_discover_tests(bycorf_task_tests 30 "runtime;task") + + add_executable(bycorf_runtime_tests + tests/connect_timer_test.cpp + tests/connection_storage_test.cpp + tests/tcp_buffered_read_test.cpp + tests/io_uring_backpressure_test.cpp) + target_link_libraries(bycorf_runtime_tests PRIVATE bycorf::core GTest::gtest_main) + bycorf_discover_tests(bycorf_runtime_tests 90 "runtime;net;io_uring") + if (BYCORF_KERNEL_BYPASS) - add_executable(bycorf_dpdk_adaptive_check tests/dpdk_adaptive_check.cpp) - target_link_libraries(bycorf_dpdk_adaptive_check PRIVATE bycorf::core) - target_include_directories(bycorf_dpdk_adaptive_check PRIVATE + add_executable(bycorf_dpdk_adaptive_tests tests/dpdk_adaptive_test.cpp) + target_link_libraries(bycorf_dpdk_adaptive_tests PRIVATE bycorf::core GTest::gtest_main) + target_include_directories(bycorf_dpdk_adaptive_tests PRIVATE "${BYCORF_DPDK_PREFIX}/include") - target_compile_options(bycorf_dpdk_adaptive_check PRIVATE + target_compile_options(bycorf_dpdk_adaptive_tests PRIVATE ${BYCORF_DPDK_COMPILE_FLAGS}) foreach(symbol IN ITEMS rte_eth_dev_configure rte_eth_dev_info_get rte_eth_dev_rx_intr_ctl_q_get_fd rte_eth_dev_rx_intr_enable rte_eth_dev_rx_intr_disable) - target_link_options(bycorf_dpdk_adaptive_check PRIVATE + target_link_options(bycorf_dpdk_adaptive_tests PRIVATE "LINKER:--wrap=${symbol}") endforeach() + bycorf_discover_tests(bycorf_dpdk_adaptive_tests 30 "runtime;net;dpdk;stub" + RUN_SERIAL TRUE) # Explicit hardware paths/allowlists are required; do not run in ctest. add_executable(bycorf_spdk_backpressure_check tests/spdk_backpressure_check.cpp) @@ -473,10 +503,10 @@ if (BYCORF_BUILD_TESTS) # Run the real backend against link-time NVMe stubs in hosted CI. Keep # device emulation out of the production request path and use SPDK's own # headers so API changes also type-check the stub signatures. - add_executable(bycorf_spdk_backpressure_stub_check - tests/spdk_backpressure_stub_check.cpp) - target_link_libraries(bycorf_spdk_backpressure_stub_check PRIVATE bycorf::core) - target_include_directories(bycorf_spdk_backpressure_stub_check PRIVATE + add_executable(bycorf_spdk_backpressure_tests + tests/spdk_backpressure_stub_test.cpp) + target_link_libraries(bycorf_spdk_backpressure_tests PRIVATE bycorf::core GTest::gtest_main) + target_include_directories(bycorf_spdk_backpressure_tests PRIVATE "${BYCORF_SPDK_SOURCE_DIR}/include") foreach(symbol IN ITEMS spdk_env_init spdk_nvme_probe spdk_nvme_ctrlr_get_ns @@ -485,46 +515,14 @@ if (BYCORF_BUILD_TESTS) spdk_nvme_ctrlr_free_io_qpair spdk_nvme_qpair_get_num_outstanding_reqs spdk_nvme_qpair_process_completions spdk_nvme_ns_cmd_read spdk_nvme_ns_cmd_write spdk_nvme_ns_cmd_flush) - target_link_options(bycorf_spdk_backpressure_stub_check PRIVATE + target_link_options(bycorf_spdk_backpressure_tests PRIVATE "LINKER:--wrap=${symbol}") endforeach() + bycorf_discover_tests(bycorf_spdk_backpressure_tests 30 "runtime;storage;spdk;stub") + + add_executable(bycorf_freebsd_local_connect_tests + tests/freebsd_local_connect_test.cpp) + target_link_libraries(bycorf_freebsd_local_connect_tests PRIVATE bycorf_freebsd GTest::gtest_main) + bycorf_discover_tests(bycorf_freebsd_local_connect_tests 15 "runtime;net;freebsd") endif() - enable_testing() - add_test(NAME bycorf_received_buffer_queue_check COMMAND bycorf_received_buffer_queue_check) - set_tests_properties(bycorf_received_buffer_queue_check PROPERTIES - TIMEOUT 30 LABELS "runtime;net") - add_test(NAME bycorf_task_transfer_check COMMAND bycorf_task_transfer_check) - set_tests_properties(bycorf_task_transfer_check PROPERTIES - TIMEOUT 30 LABELS "runtime;task") - if (BYCORF_KERNEL_BYPASS) - foreach(scenario IN ITEMS configure descriptor enable disable poll - runtime-enable runtime-disable mlx5-empty) - add_test(NAME bycorf_dpdk_adaptive_${scenario} - COMMAND bycorf_dpdk_adaptive_check ${scenario}) - set_tests_properties(bycorf_dpdk_adaptive_${scenario} PROPERTIES - TIMEOUT 30 LABELS "runtime;net;dpdk;stub" RUN_SERIAL TRUE) - endforeach() - add_test(NAME bycorf_spdk_backpressure_stub_check - COMMAND bycorf_spdk_backpressure_stub_check) - set_tests_properties(bycorf_spdk_backpressure_stub_check PROPERTIES - TIMEOUT 30 LABELS "runtime;storage;spdk;stub") - add_executable(bycorf_freebsd_local_connect_check - tests/freebsd_local_connect_check.cpp) - target_link_libraries(bycorf_freebsd_local_connect_check PRIVATE bycorf_freebsd) - add_test(NAME bycorf_freebsd_local_connect_check - COMMAND bycorf_freebsd_local_connect_check) - set_tests_properties(bycorf_freebsd_local_connect_check PROPERTIES - TIMEOUT 15 LABELS "runtime;net;freebsd") - endif() - add_test(NAME bycorf_connect_timer_check COMMAND bycorf_connect_timer_check) - add_test(NAME bycorf_connection_storage_check COMMAND bycorf_connection_storage_check) - add_test(NAME bycorf_tcp_buffered_read_check COMMAND bycorf_tcp_buffered_read_check) - add_test(NAME bycorf_io_uring_backpressure_check COMMAND bycorf_io_uring_backpressure_check) - add_test(NAME bycorf_backend_selection_check COMMAND bycorf_backend_selection_check) - set_tests_properties(bycorf_connection_storage_check bycorf_backend_selection_check - bycorf_tcp_buffered_read_check bycorf_io_uring_backpressure_check - PROPERTIES TIMEOUT 90 LABELS "runtime;net;io_uring") - set_tests_properties(bycorf_connect_timer_check PROPERTIES - TIMEOUT 90 LABELS "runtime;net;io_uring" - ) endif() diff --git a/README.md b/README.md index 00f10b1..d41b38a 100644 --- a/README.md +++ b/README.md @@ -91,17 +91,36 @@ container policy, and enough locked-memory allowance for registered buffers. If startup fails, check the host's `kernel.io_uring_disabled` setting and the shell's `ulimit -l`; compilation alone does not verify these runtime conditions. -CTest always registers the following software regressions. The Task check has -a 30-second timeout; the other checks below have 90-second timeouts: - -| Test | Coverage | +Software regressions use GoogleTest and are built only with +`BYCORF_BUILD_TESTS=ON`. Test builds fetch the pinned GoogleTest v1.17.0 +sources unless an embedding project already supplies its targets; library and +example-only builds do not need GoogleTest. CTest discovers each case separately +with a `bycorf_` prefix, preserving per-case failures, timeouts, labels, and JUnit +results. + +| Test suite | Coverage | |---|---| -| `bycorf_task_transfer_check` | Bounded native stack for sequential and nested awaits, asynchronous completion, cancellation ownership and reentrant callbacks | -| `bycorf_connect_timer_check` | Early shutdown, timers and cancellation, loopback TCP, refused connections, deadlines, address validation, and retirement of losing deadlines | -| `bycorf_connection_storage_check` | Connection storage lifetime while suspended coroutines still borrow it | -| `bycorf_tcp_buffered_read_check` | Bounded reads across queued receive buffers, partial buffers, unchanged ordinary reads, waiting, errors, and EOF | -| `bycorf_io_uring_backpressure_check` | Full submission queues, deferred sends and storage reads, queued cancellation, descriptor reuse, peer wakes, and shutdown | -| `bycorf_backend_selection_check` | Unsupported backend requests, inactive optional backends, and immutable allocator/backend selection | +| `InlineRingQueue`, `ReceivedBufferQueue` | Queue types and inline capacities, moves, FIFO order, wrapping, growth, and shrinking | +| `TaskTransfer` | Bounded native stack for sequential and nested awaits, asynchronous completion, cancellation ownership and reentrant callbacks | +| `RuntimeStartup`, `TcpServiceStop`, `Timer`, `Connect` | Early shutdown, timers and cancellation, loopback TCP, refused connections, deadlines, address validation, and retirement of losing deadlines | +| `ConnectionStorage` | Connection storage lifetime while suspended coroutines still borrow it | +| `TcpBufferedRead` | Bounded reads across queued receive buffers, partial buffers, unchanged ordinary reads, waiting, errors, EOF, and close without shrinking | +| `IoUringBackpressure` | Full submission queues, deferred sends and storage reads, queued cancellation, descriptor reuse, peer wakes, and shutdown | +| `BackendSelectionDeathTest` | Unsupported backend requests, inactive optional backends, and immutable allocator/backend selection | + +Queue, backend-selection, and Task cases have 30-second timeouts; worker/I/O +cases have 90-second timeouts. Task cases run after re-exec with a 512 KiB native +stack. Backend-selection cases use subprocesses because allocator/backend +selection has process lifetime. Other cases own fresh worker/backend fixtures. +Listing tests does not initialize I/O or access devices. + +List all cases, run one case through CTest, or filter the runtime binary directly: + +```bash +ctest --test-dir build -N +ctest --test-dir build -R '^bycorf_Timer.CancelOnWorker$' --output-on-failure +./build/bycorf_runtime_tests --gtest_filter='Timer.*' +``` Network checks use one unpinned worker and ephemeral loopback ports or Unix socket pairs. The connection-timeout probe warns if the host rejects its @@ -110,16 +129,22 @@ instead of the deadline. These software tests also run in a bypass-capable build with the default kernel/io_uring selection, without activating EAL or accessing devices. -With `BYCORF_KERNEL_BYPASS=ON`, CTest also runs the FreeBSD local-connect -regression and `bycorf_spdk_backpressure_stub_check`. The SPDK check uses the -real backend and Worker with link-time stubs for EAL initialization and NVMe -device calls. It covers queue exhaustion, FIFO retries, independent qpairs, -large backlogs, submission/completion errors, completion-tag reuse, and shutdown -draining without NVMe hardware, hugepages, or device binding. It still requires -working io_uring and the normal runtime memlock allowance. Run it alone with: +With `BYCORF_KERNEL_BYPASS=ON`, CTest also discovers the FreeBSD local-connect, +DPDK adaptive-mode, and SPDK backpressure suites. The DPDK and SPDK suites use +link-time device stubs; DPDK additionally uses an in-memory ring PMD. Their +cases run in subprocesses to isolate process-global backend initialization. +DPDK cases remain serial under CTest. The FreeBSD timeout is 15 seconds and +the DPDK/SPDK timeouts are 30 seconds. + +The SPDK suite uses the real backend and Worker with stubs for EAL +initialization and NVMe device calls. It covers queue exhaustion, FIFO retries, +independent qpairs, large backlogs, submission/completion errors, completion-tag +reuse, and shutdown draining without NVMe hardware, hugepages, or device +binding. It still requires working io_uring and the normal runtime memlock +allowance. Run it alone with: ```bash -ctest --test-dir build -R '^bycorf_spdk_backpressure_stub_check$' --output-on-failure +ctest --test-dir build -R '^bycorf_SpdkBackpressureDeathTest\.' --output-on-failure ``` `bycorf_spdk_backpressure_check` is a separate read-only hardware check. It is diff --git a/docs/dpdk-prototype.md b/docs/dpdk-prototype.md index c22a529..08d7c8e 100644 --- a/docs/dpdk-prototype.md +++ b/docs/dpdk-prototype.md @@ -285,8 +285,9 @@ The complete device allowlist must be set before that first initialization. `bycorf_echo` accepts a final positional network selector after the idle timeout: `bycorf_echo ADDRESS PORT WORKERS IDLE_TIMEOUT_MS kernel|dpdk`. -`dpdk_smoke.py` selects DPDK explicitly. `bycorf_backend_selection_check` checks -ordinary allocation and immutable selection without touching physical devices. +`dpdk_smoke.py` selects DPDK explicitly. The `BackendSelectionDeathTest` +GoogleTest cases check ordinary allocation and immutable selection without +touching physical devices. ## Application integration diff --git a/example/backend_selection_check.cpp b/example/backend_selection_check.cpp deleted file mode 100644 index 1932b21..0000000 --- a/example/backend_selection_check.cpp +++ /dev/null @@ -1,57 +0,0 @@ -/* - * Copyright (C) 2026 EloqData Inc. - * - * 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 - * - * https://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. - */ - -#include -#include -#include - -#include "bycorf/io/backend_options.h" -#include "bycorf/io/dpdk_environment.h" -#include "bycorf/io/spdk_storage.h" - -// No device access: even an all-capability binary must allocate ordinary memory -// with the default selection, and may never change allocator identity later. -int main() { - if (bycorf::DpdkNetworkEnabled() || bycorf::SpdkStorageEnabled()) return 1; -#if BYCORF_KERNEL_BYPASS - // Compiling both capabilities must still allow independent startup choices. - // Selection alone must not initialize EAL or access a device. - const bycorf::IoBackends supported[] = { - {.dpdk_network = true}, - {.spdk_storage = true}, - {.dpdk_network = true, .spdk_storage = true}, - }; - for (const auto options : supported) { - if (!bycorf::ConfigureIoBackends(options).ok() || - bycorf::DpdkNetworkEnabled() != options.dpdk_network || - bycorf::SpdkStorageEnabled() != options.spdk_storage) - return 9; - } -#else - if (bycorf::ConfigureIoBackends({.dpdk_network = true}).ok()) return 2; - if (bycorf::ConfigureIoBackends({.spdk_storage = true}).ok()) return 3; -#endif - if (!bycorf::ConfigureIoBackends({}).ok()) return 4; - void* memory = bycorf::AllocateStorageBuffer(4096, 4096); - if (!memory) return 5; - std::memset(memory, 0xa5, 4096); - if (bycorf::ConfigureIoBackends({.dpdk_network = true}).ok()) return 6; - if (bycorf::ConfigureIoBackends({.spdk_storage = true}).ok()) return 7; - if (!bycorf::ConfigureIoBackends({}).ok()) return 8; - bycorf::FreeStorageBuffer(memory, 4096); - std::cout - << "default kernel/uring; inactive DMA; immutable selection: PASS\n"; -} diff --git a/tests/backend_selection_test.cpp b/tests/backend_selection_test.cpp new file mode 100644 index 0000000..77634af --- /dev/null +++ b/tests/backend_selection_test.cpp @@ -0,0 +1,79 @@ +/* + * Copyright (C) 2026 EloqData Inc. + * + * 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 + * + * https://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. + */ + +#include + +#include +#include + +#include "bycorf/io/backend_options.h" +#include "bycorf/io/dpdk_environment.h" +#include "bycorf/io/spdk_storage.h" + +namespace { +// Backend selection is process-global and becomes immutable on allocation. +// Subprocesses keep these cases independent even under --gtest_shuffle/repeat; +// no production reset hook or device initialization is needed. +TEST(BackendSelectionDeathTest, DefaultsToKernelAndIoUring) { + ASSERT_EXIT(([] { + EXPECT_FALSE(bycorf::DpdkNetworkEnabled()); + EXPECT_FALSE(bycorf::SpdkStorageEnabled()); + std::exit(testing::Test::HasFailure() ? 1 : 0); + }()), + testing::ExitedWithCode(0), ""); +} + +TEST(BackendSelectionDeathTest, CompiledCapabilitiesSelectIndependently) { + ASSERT_EXIT(([] { + const bycorf::IoBackends supported[] = { + {.dpdk_network = true}, + {.spdk_storage = true}, + {.dpdk_network = true, .spdk_storage = true}, + }; + for (const auto options : supported) { + const auto status = bycorf::ConfigureIoBackends(options); +#if BYCORF_KERNEL_BYPASS + EXPECT_TRUE(status.ok()) << status; + EXPECT_EQ(bycorf::DpdkNetworkEnabled(), options.dpdk_network); + EXPECT_EQ(bycorf::SpdkStorageEnabled(), options.spdk_storage); +#else + EXPECT_TRUE(absl::IsUnimplemented(status)) << status; + EXPECT_FALSE(bycorf::DpdkNetworkEnabled()); + EXPECT_FALSE(bycorf::SpdkStorageEnabled()); +#endif + } + EXPECT_TRUE(bycorf::ConfigureIoBackends({}).ok()); + std::exit(testing::Test::HasFailure() ? 1 : 0); + }()), + testing::ExitedWithCode(0), ""); +} + +TEST(BackendSelectionDeathTest, AllocationFreezesBackendAndAllocator) { + ASSERT_EXIT( + ([] { + EXPECT_TRUE(bycorf::ConfigureIoBackends({}).ok()); + void* memory = bycorf::AllocateStorageBuffer(4096, 4096); + EXPECT_NE(memory, nullptr); + if (memory) std::memset(memory, 0xa5, 4096); + EXPECT_FALSE(bycorf::ConfigureIoBackends({.dpdk_network = true}).ok()); + EXPECT_FALSE(bycorf::ConfigureIoBackends({.spdk_storage = true}).ok()); + EXPECT_TRUE(bycorf::ConfigureIoBackends({}).ok()); + bycorf::FreeStorageBuffer(memory, 4096); + std::exit(testing::Test::HasFailure() ? 1 : 0); + }()), + testing::ExitedWithCode(0), ""); +} +} // namespace diff --git a/example/connect_timer_check.cpp b/tests/connect_timer_test.cpp similarity index 73% rename from example/connect_timer_check.cpp rename to tests/connect_timer_test.cpp index 8d72c6b..c1b71c6 100644 --- a/example/connect_timer_check.cpp +++ b/tests/connect_timer_test.cpp @@ -19,7 +19,7 @@ // - bycorf::CancellableSleepFor (one-shot timer with a thread-safe cancel) // - bycorf::Runtime (stop requests racing worker startup) // - bycorf::TcpService (foreign stop before/after listener startup) -// Runs every check on one bycorf worker and exits non-zero on any failure. +// Each asynchronous case owns one worker and reports assertions to GoogleTest. #include #include @@ -41,25 +41,14 @@ #include "bycorf/net/tcp_listener.h" #include "bycorf/net/tcp_service.h" #include "bycorf/net/tcp_stream.h" -#include "bycorf/runtime/runtime.h" #include "bycorf/runtime/worker.h" +#include "runtime_test.h" #include "spdlog/spdlog.h" namespace { using namespace std::chrono_literals; -int g_failures = 0; - -void Check(bool condition, std::string_view name) { - if (condition) { - spdlog::info("[pass] {}", name); - } else { - spdlog::error("[FAIL] {}", name); - ++g_failures; - } -} - void CheckStopDuringStartup(bool before_init) { bycorf::Runtime runtime; std::promise parked; @@ -93,9 +82,9 @@ void CheckStopDuringStartup(bool before_init) { runtime.RequestStop(); release.set_value(); runtime.WaitUntilStopped(); - Check(initialized.ok() && runtime.exit_code() == 0, - before_init ? "startup stop: before Worker::Init" - : "startup stop: after Init, before Run"); + EXPECT_TRUE(initialized.ok() && runtime.exit_code() == 0) + << (before_init ? "startup stop: before Worker::Init" + : "startup stop: after Init, before Run"); } class StopCheckService final : public bycorf::TcpService { @@ -134,7 +123,7 @@ void CheckTcpServiceStop(bool before_run) { getsockname(reservation, reinterpret_cast(&address), &length) == 0; if (reservation >= 0) close(reservation); - Check(reserved, "TCP service stop: reserve loopback port"); + EXPECT_TRUE(reserved) << "TCP service stop: reserve loopback port"; if (!reserved) return; StopCheckService service(ntohs(address.sin_port)); @@ -175,7 +164,7 @@ void CheckTcpServiceStop(bool before_run) { /*pin_workers=*/false); const bool initialized = ready.wait_for(5s) == std::future_status::ready && ready.get().ok(); - Check(initialized, "TCP service stop: startup reached gate"); + EXPECT_TRUE(initialized) << "TCP service stop: startup reached gate"; service.Stop(); service.Stop(); if (before_run) release.set_value(); @@ -186,19 +175,21 @@ void CheckTcpServiceStop(bool before_run) { // runtime teardown must not hide a lost stop-before-publication race. const bool completed = stopped.wait_for(5s) == std::future_status::ready && stopped.get().ok(); - Check(completed, "TCP service stop: before Run publishes owner"); + EXPECT_TRUE(completed) << "TCP service stop: before Run publishes owner"; } else { // Stop posts listener closure before this marker on the same ingress. // A consumed accept may still retain the socket until ring teardown, so // require owner-side dispatch here rather than completion of that accept. const bool posted = runtime.GetForeignExecutor(0).Notify( [&owner_marker]() noexcept { owner_marker.set_value(); }); - Check(posted && owner_reached.wait_for(5s) == std::future_status::ready, - "TCP service stop: foreign close reaches owning worker"); + EXPECT_TRUE(posted && + owner_reached.wait_for(5s) == std::future_status::ready) + << "TCP service stop: foreign close reaches owning worker"; } runtime.RequestStop(); runtime.WaitUntilStopped(); - Check(runtime.exit_code() == 0, "TCP service stop: clean worker teardown"); + EXPECT_TRUE(runtime.exit_code() == 0) + << "TCP service stop: clean worker teardown"; } std::int64_t ElapsedMs(std::chrono::steady_clock::time_point start) { @@ -221,8 +212,8 @@ bycorf::Task CheckTimerFire(bycorf::Worker& worker) { auto timer = bycorf::CancellableSleepFor(worker, 100ms); const auto started = std::chrono::steady_clock::now(); const absl::Status fired = co_await timer; - Check(fired.ok(), "timer fires with OkStatus"); - Check(ElapsedMs(started) >= 90, "timer fire respects the deadline"); + EXPECT_TRUE(fired.ok()) << "timer fires with OkStatus"; + EXPECT_TRUE(ElapsedMs(started) >= 90) << "timer fire respects the deadline"; co_return absl::OkStatus(); } @@ -232,11 +223,11 @@ bycorf::Task CheckTimerCancelOnWorker(bycorf::Worker& worker) { worker.Spawn(CancelAfterOnWorker(worker, handle, 50ms)); const auto started = std::chrono::steady_clock::now(); const absl::Status fired = co_await timer; - Check(fired.code() == absl::StatusCode::kCancelled, - "on-worker cancel resolves kCancelled"); + EXPECT_TRUE(fired.code() == absl::StatusCode::kCancelled) + << "on-worker cancel resolves kCancelled"; // The pending timeout is pulled off the ring, so the resume is prompt // (~50ms), not at the 5s deadline. - Check(ElapsedMs(started) < 2000, "on-worker cancel resumes promptly"); + EXPECT_TRUE(ElapsedMs(started) < 2000) << "on-worker cancel resumes promptly"; co_return absl::OkStatus(); } @@ -254,10 +245,10 @@ bycorf::Task CheckTimerCancelForeignThread( const auto started = std::chrono::steady_clock::now(); const absl::Status fired = co_await timer; canceller.join(); - Check(fired.code() == absl::StatusCode::kCancelled, - "foreign-thread cancel resolves kCancelled"); - Check(ElapsedMs(started) >= 700, - "foreign-thread cancel: late fire is a no-op at the deadline"); + EXPECT_TRUE(fired.code() == absl::StatusCode::kCancelled) + << "foreign-thread cancel resolves kCancelled"; + EXPECT_TRUE(ElapsedMs(started) >= 700) + << "foreign-thread cancel: late fire is a no-op at the deadline"; co_return absl::OkStatus(); } @@ -265,13 +256,12 @@ bycorf::Task CheckTimerCancelAfterFire(bycorf::Worker& worker) { auto timer = bycorf::CancellableSleepFor(worker, 30ms); bycorf::TimerCancelHandle handle = timer.CancelHandle(); const absl::Status fired = co_await timer; - Check(fired.ok(), "timer fires before the cancel"); + EXPECT_TRUE(fired.ok()) << "timer fires before the cancel"; // Cancel-after-fire and double cancel are safe no-ops on live shared state. handle.Cancel(); handle.Cancel(); bycorf::TimerCancelHandle empty; empty.Cancel(); - Check(true, "cancel-after-fire and double cancel are safe"); co_return absl::OkStatus(); } @@ -280,9 +270,10 @@ bycorf::Task CheckTimerCancelBeforeAwait(bycorf::Worker& worker) { timer.CancelHandle().Cancel(); const auto started = std::chrono::steady_clock::now(); const absl::Status fired = co_await timer; - Check(fired.code() == absl::StatusCode::kCancelled, - "cancel-before-await resolves kCancelled"); - Check(ElapsedMs(started) < 100, "cancel-before-await never hits the ring"); + EXPECT_TRUE(fired.code() == absl::StatusCode::kCancelled) + << "cancel-before-await resolves kCancelled"; + EXPECT_TRUE(ElapsedMs(started) < 100) + << "cancel-before-await never hits the ring"; co_return absl::OkStatus(); } @@ -303,12 +294,12 @@ bycorf::Task CheckConnectSuccess(bycorf::Worker& worker) { bycorf::TcpListener listener; absl::Status bound = listener.Bind(&worker, "127.0.0.1", 0); if (!bound.ok()) { - Check(false, "connect-success: listener bind"); + ADD_FAILURE() << "connect-success: listener bind"; co_return bound; } auto port = BoundPort(listener); if (!port.ok()) { - Check(false, "connect-success: bound port"); + ADD_FAILURE() << "connect-success: bound port"; co_return port.status(); } @@ -317,12 +308,12 @@ bycorf::Task CheckConnectSuccess(bycorf::Worker& worker) { worker, "127.0.0.1", *port, 2s, &cancellation); cancellation.Cancel(); cancellation.Cancel(); - Check(client.ok(), "connect-success: ConnectTcp to local listener"); + EXPECT_TRUE(client.ok()) << "connect-success: ConnectTcp to local listener"; if (!client.ok()) { co_return client.status(); } auto accepted = co_await listener.Accept(); - Check(accepted.ok(), "connect-success: listener accepts"); + EXPECT_TRUE(accepted.ok()) << "connect-success: listener accepts"; if (!accepted.ok()) { co_return accepted.status(); } @@ -331,26 +322,26 @@ bycorf::Task CheckConnectSuccess(bycorf::Worker& worker) { constexpr std::string_view kPing = "ping"; absl::Status written = co_await client->WriteAll(std::span( reinterpret_cast(kPing.data()), kPing.size())); - Check(written.ok(), "connect-success: client write"); + EXPECT_TRUE(written.ok()) << "connect-success: client write"; std::array buffer{}; auto server_read = co_await server.ReadSome(buffer); - Check(server_read.ok() && *server_read == kPing.size() && - std::memcmp(buffer.data(), kPing.data(), kPing.size()) == 0, - "connect-success: server read"); + EXPECT_TRUE(server_read.ok() && *server_read == kPing.size() && + std::memcmp(buffer.data(), kPing.data(), kPing.size()) == 0) + << "connect-success: server read"; if (server_read.ok()) { absl::Status echoed = co_await server.WriteAll( std::span(buffer.data(), *server_read)); - Check(echoed.ok(), "connect-success: server echo"); + EXPECT_TRUE(echoed.ok()) << "connect-success: server echo"; } auto client_read = co_await client->ReadSome(buffer); - Check(client_read.ok() && *client_read == kPing.size() && - std::memcmp(buffer.data(), kPing.data(), kPing.size()) == 0, - "connect-success: client read echo"); + EXPECT_TRUE(client_read.ok() && *client_read == kPing.size() && + std::memcmp(buffer.data(), kPing.data(), kPing.size()) == 0) + << "connect-success: client read echo"; auto peer = client->PeerAddress(); - Check(peer.ok() && *peer == "127.0.0.1:" + std::to_string(*port), - "connect-success: peer address"); + EXPECT_TRUE(peer.ok() && *peer == "127.0.0.1:" + std::to_string(*port)) + << "connect-success: peer address"; server.Close().IgnoreError(); client->Close().IgnoreError(); @@ -365,21 +356,21 @@ bycorf::Task CheckConnectRefused(bycorf::Worker& worker) { bycorf::TcpListener listener; absl::Status bound = listener.Bind(&worker, "127.0.0.1", 0); if (!bound.ok()) { - Check(false, "connect-refused: listener bind"); + ADD_FAILURE() << "connect-refused: listener bind"; co_return bound; } auto port = BoundPort(listener); listener.Close().IgnoreError(); if (!port.ok()) { - Check(false, "connect-refused: bound port"); + ADD_FAILURE() << "connect-refused: bound port"; co_return port.status(); } const auto started = std::chrono::steady_clock::now(); auto client = co_await bycorf::ConnectTcp(worker, "127.0.0.1", *port, 5s); - Check( - !client.ok() && client.status().code() == absl::StatusCode::kUnavailable, - "connect-refused: ECONNREFUSED maps to kUnavailable"); - Check(ElapsedMs(started) < 2000, "connect-refused: fails fast"); + EXPECT_TRUE(!client.ok() && + client.status().code() == absl::StatusCode::kUnavailable) + << "connect-refused: ECONNREFUSED maps to kUnavailable"; + EXPECT_TRUE(ElapsedMs(started) < 2000) << "connect-refused: fails fast"; co_return absl::OkStatus(); } @@ -399,22 +390,21 @@ bycorf::Task CheckConnectTimeout(bycorf::Worker& worker) { "[warn] connect-timeout: immediate unreachable after {}ms (no route " "to 10/8 in this environment)", elapsed); - Check(true, "connect-timeout: immediate unreachable (see warning)"); co_return absl::OkStatus(); } - Check(!client.ok() && - client.status().code() == absl::StatusCode::kDeadlineExceeded, - "connect-timeout: resolves kDeadlineExceeded"); - Check(elapsed >= 280 && elapsed < 10000, - "connect-timeout: deadline bounds the wait"); + EXPECT_TRUE(!client.ok() && + client.status().code() == absl::StatusCode::kDeadlineExceeded) + << "connect-timeout: resolves kDeadlineExceeded"; + EXPECT_TRUE(elapsed >= 280 && elapsed < 10000) + << "connect-timeout: deadline bounds the wait"; co_return absl::OkStatus(); } bycorf::Task CheckConnectRejectsHostname(bycorf::Worker& worker) { auto client = co_await bycorf::ConnectTcp(worker, "localhost", 17699, 100ms); - Check(!client.ok() && - client.status().code() == absl::StatusCode::kInvalidArgument, - "connect: non-numeric host is rejected without DNS"); + EXPECT_TRUE(!client.ok() && + client.status().code() == absl::StatusCode::kInvalidArgument) + << "connect: non-numeric host is rejected without DNS"; co_return absl::OkStatus(); } @@ -431,13 +421,13 @@ bycorf::Task CheckConnectFastFailDeadlineRetired( bycorf::TcpListener listener; absl::Status bound = listener.Bind(&worker, "127.0.0.1", 0); if (!bound.ok()) { - Check(false, "deadline-retired(fail): listener bind"); + ADD_FAILURE() << "deadline-retired(fail): listener bind"; co_return bound; } auto port = BoundPort(listener); listener.Close().IgnoreError(); if (!port.ok()) { - Check(false, "deadline-retired(fail): bound port"); + ADD_FAILURE() << "deadline-retired(fail): bound port"; co_return port.status(); } @@ -447,17 +437,14 @@ bycorf::Task CheckConnectFastFailDeadlineRetired( auto client = co_await bycorf::ConnectTcp(worker, "127.0.0.1", *port, 1s); if (client.ok() || client.status().code() != absl::StatusCode::kUnavailable) { - Check(false, - "deadline-retired(fail): refused connect maps to " - "kUnavailable"); + ADD_FAILURE() << "deadline-retired(fail): refused connect maps to " + "kUnavailable"; co_return absl::OkStatus(); } } - Check(ElapsedMs(started) < 2000, - "deadline-retired(fail): refused loop stays fast"); + EXPECT_TRUE(ElapsedMs(started) < 2000) + << "deadline-retired(fail): refused loop stays fast"; (void)co_await bycorf::SleepFor(worker, 1500ms); - Check(true, - "deadline-retired(fail): no late timeout dispatch after teardown"); co_return absl::OkStatus(); } @@ -469,12 +456,12 @@ bycorf::Task CheckConnectFastSuccessDeadlineRetired( bycorf::TcpListener listener; absl::Status bound = listener.Bind(&worker, "127.0.0.1", 0); if (!bound.ok()) { - Check(false, "deadline-retired(ok): listener bind"); + ADD_FAILURE() << "deadline-retired(ok): listener bind"; co_return bound; } auto port = BoundPort(listener); if (!port.ok()) { - Check(false, "deadline-retired(ok): bound port"); + ADD_FAILURE() << "deadline-retired(ok): bound port"; co_return port.status(); } @@ -485,20 +472,19 @@ bycorf::Task CheckConnectFastSuccessDeadlineRetired( // userspace accept is needed for the connect to win instantly. auto client = co_await bycorf::ConnectTcp(worker, "127.0.0.1", *port, 1s); if (!client.ok()) { - Check(false, "deadline-retired(ok): connect succeeds"); + ADD_FAILURE() << "deadline-retired(ok): connect succeeds"; co_return client.status(); } client->Close().IgnoreError(); } - Check(ElapsedMs(started) < 2000, - "deadline-retired(ok): connect loop stays fast"); + EXPECT_TRUE(ElapsedMs(started) < 2000) + << "deadline-retired(ok): connect loop stays fast"; (void)co_await bycorf::SleepFor(worker, 1500ms); - Check(true, "deadline-retired(ok): no late timeout dispatch after teardown"); for (int attempt = 0; attempt < kAttempts; ++attempt) { auto accepted = co_await listener.Accept(); if (!accepted.ok()) { - Check(false, "deadline-retired(ok): drain accepted sockets"); + ADD_FAILURE() << "deadline-retired(ok): drain accepted sockets"; break; } bycorf::TcpStream server(*accepted); @@ -522,8 +508,8 @@ bycorf::Task CheckConnectCancellation(bycorf::Worker& worker) { before.Cancel(); auto cancelled = co_await bycorf::ConnectTcpCancellable(worker, "127.0.0.1", 1, 5s, &before); - Check(!cancelled.ok() && absl::IsCancelled(cancelled.status()), - "connect: cancel before submission"); + EXPECT_TRUE(!cancelled.ok() && absl::IsCancelled(cancelled.status())) + << "connect: cancel before submission"; // A full, zero-length listen backlog drops subsequent SYNs on loopback. // Exercise a real pending connect without relying on an external route or @@ -531,7 +517,8 @@ bycorf::Task CheckConnectCancellation(bycorf::Worker& worker) { bycorf::TcpListener listener; auto bound = listener.Bind(&worker, "127.0.0.1", 0); if (!bound.ok()) co_return bound; - Check(::listen(listener.NativeFd(), 0) == 0, "connect-cancel: tiny backlog"); + EXPECT_TRUE(::listen(listener.NativeFd(), 0) == 0) + << "connect-cancel: tiny backlog"; auto port = BoundPort(listener); if (!port.ok()) co_return port.status(); auto filler = co_await bycorf::ConnectTcp(worker, "127.0.0.1", *port, 1s); @@ -543,58 +530,77 @@ bycorf::Task CheckConnectCancellation(bycorf::Worker& worker) { const auto started = std::chrono::steady_clock::now(); auto client = co_await bycorf::ConnectTcpCancellable( worker, "127.0.0.1", *port, 1s, cancellation.get()); - Check(!client.ok() && absl::IsCancelled(client.status()), - "connect: pending attempt resolves Cancelled"); - Check(ElapsedMs(started) < 500, "connect: cancellation beats deadline"); + EXPECT_TRUE(!client.ok() && absl::IsCancelled(client.status())) + << "connect: pending attempt resolves Cancelled"; + EXPECT_TRUE(ElapsedMs(started) < 500) + << "connect: cancellation beats deadline"; if (client.ok()) client->Close().IgnoreError(); } (void)co_await bycorf::SleepFor(worker, 1100ms); - Check(true, "connect: cancelled frames have no late deadline completion"); filler->Close().IgnoreError(); listener.Close().IgnoreError(); co_return absl::OkStatus(); } -bycorf::Task RunAllChecks(bycorf::Worker& worker) { - co_await CheckTimerFire(worker); - co_await CheckTimerCancelOnWorker(worker); - co_await CheckTimerCancelForeignThread(worker); - co_await CheckTimerCancelAfterFire(worker); - co_await CheckTimerCancelBeforeAwait(worker); - co_await CheckConnectSuccess(worker); - const auto cancellation = co_await CheckConnectCancellation(worker); - Check(cancellation.ok(), "connect: cancellation scenario completed"); - co_await CheckConnectRefused(worker); - co_await CheckConnectTimeout(worker); - co_await CheckConnectRejectsHostname(worker); - co_await CheckConnectFastFailDeadlineRetired(worker); - co_await CheckConnectFastSuccessDeadlineRetired(worker); - spdlog::info("checks complete: {} failure(s)", g_failures); - worker.RequestStop(); - co_return absl::OkStatus(); +TEST(RuntimeStartup, StopBeforeWorkerInit) { CheckStopDuringStartup(true); } +TEST(RuntimeStartup, StopBeforeWorkerRun) { CheckStopDuringStartup(false); } +TEST(TcpServiceStop, BeforeRun) { CheckTcpServiceStop(true); } +TEST(TcpServiceStop, AfterListenerStartup) { CheckTcpServiceStop(false); } + +TEST(Timer, Fire) { + bycorf::test::RunOnWorker(CheckTimerFire, {.recv_buffer_count_ = 64}); } -} // namespace +TEST(Timer, CancelOnWorker) { + bycorf::test::RunOnWorker(CheckTimerCancelOnWorker, + {.recv_buffer_count_ = 64}); +} -int main() { - CheckStopDuringStartup(/*before_init=*/true); - CheckStopDuringStartup(/*before_init=*/false); - CheckTcpServiceStop(/*before_run=*/true); - CheckTcpServiceStop(/*before_run=*/false); - bycorf::Runtime runtime; - runtime.Start( - 1, - [](unsigned, bycorf::Worker& worker) -> int { - bycorf::WorkerOptions options; - options.recv_buffer_count_ = 64; - if (!worker.Init(options).ok()) { - return 1; - } - worker.Spawn(RunAllChecks(worker)); - worker.Run(); - return g_failures == 0 ? 0 : 1; - }, - /*pin_workers=*/false); - runtime.WaitUntilStopped(); - return runtime.exit_code(); +TEST(Timer, CancelForeignThread) { + bycorf::test::RunOnWorker(CheckTimerCancelForeignThread, + {.recv_buffer_count_ = 64}); } + +TEST(Timer, CancelAfterFire) { + bycorf::test::RunOnWorker(CheckTimerCancelAfterFire, + {.recv_buffer_count_ = 64}); +} + +TEST(Timer, CancelBeforeAwait) { + bycorf::test::RunOnWorker(CheckTimerCancelBeforeAwait, + {.recv_buffer_count_ = 64}); +} + +TEST(Connect, Success) { + bycorf::test::RunOnWorker(CheckConnectSuccess, {.recv_buffer_count_ = 64}); +} + +TEST(Connect, Cancellation) { + bycorf::test::RunOnWorker(CheckConnectCancellation, + {.recv_buffer_count_ = 64}); +} + +TEST(Connect, Refused) { + bycorf::test::RunOnWorker(CheckConnectRefused, {.recv_buffer_count_ = 64}); +} + +TEST(Connect, Timeout) { + bycorf::test::RunOnWorker(CheckConnectTimeout, {.recv_buffer_count_ = 64}); +} + +TEST(Connect, RejectsHostname) { + bycorf::test::RunOnWorker(CheckConnectRejectsHostname, + {.recv_buffer_count_ = 64}); +} + +TEST(Connect, FastFailDeadlineRetired) { + bycorf::test::RunOnWorker(CheckConnectFastFailDeadlineRetired, + {.recv_buffer_count_ = 64}); +} + +TEST(Connect, FastSuccessDeadlineRetired) { + bycorf::test::RunOnWorker(CheckConnectFastSuccessDeadlineRetired, + {.recv_buffer_count_ = 64}); +} + +} // namespace diff --git a/example/connection_storage_check.cpp b/tests/connection_storage_test.cpp similarity index 66% rename from example/connection_storage_check.cpp rename to tests/connection_storage_test.cpp index 4727188..c0d0bef 100644 --- a/example/connection_storage_check.cpp +++ b/tests/connection_storage_test.cpp @@ -20,30 +20,21 @@ #include #include -#include #include #include #include "bycorf/io/storage.h" #include "bycorf/net/connection.h" -#include "bycorf/runtime/runtime.h" +#include "runtime_test.h" namespace { using namespace std::chrono_literals; -int failures = 0; - -void Check(bool passed, const char* message) { - std::cout << (passed ? "PASS " : "FAIL ") << message << '\n'; - if (!passed) ++failures; -} - bycorf::Task CheckStorage(bycorf::Worker& worker) { int sockets[2]; if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, sockets) != 0) { - Check(false, "create socket pair"); - worker.RequestStop(); + ADD_FAILURE() << "create socket pair"; co_return absl::InternalError("socketpair failed"); } @@ -60,44 +51,38 @@ bycorf::Task CheckStorage(bycorf::Worker& worker) { auto* connection = worker.AddConnection(std::move(initial)); bycorf::BorrowConnectionStorage(connection); bycorf::BorrowConnectionStorage(connection); - Check(worker.EnsureRecvArmed(connection).ok(), "arm real pending receive"); + EXPECT_TRUE(worker.EnsureRecvArmed(connection).ok()) + << "arm real pending receive"; worker.BeginClose(connection, absl::CancelledError("test close"), bycorf::CloseMode::kLocalClose); - Check(fcntl(sockets[0], F_GETFD) == -1 && errno == EBADF, - "storage borrow allows immediate transport close"); + EXPECT_TRUE(fcntl(sockets[0], F_GETFD) == -1 && errno == EBADF) + << "storage borrow allows immediate transport close"; close(sockets[1]); // Returning to the event loop lets cancellation completions and reclamation // run while this session is suspended on a timer instead of socket I/O. (void)co_await bycorf::SleepFor(worker, 20ms); - Check(!storage_alive.expired(), "closed storage survives suspended session"); + EXPECT_TRUE(!storage_alive.expired()) + << "closed storage survives suspended session"; if (!storage_alive.expired()) { - Check(connection->closed_ && !connection->recv_armed_, - "receive cancellation drains while storage stays borrowed"); + EXPECT_TRUE(connection->closed_ && !connection->recv_armed_) + << "receive cancellation drains while storage stays borrowed"; bycorf::ReleaseConnectionStorage(connection); (void)co_await bycorf::SleepFor(worker, 20ms); - Check(!storage_alive.expired(), "one remaining borrower prevents reclaim"); + EXPECT_TRUE(!storage_alive.expired()) + << "one remaining borrower prevents reclaim"; if (!storage_alive.expired()) { bycorf::ReleaseConnectionStorage(connection); (void)co_await bycorf::SleepFor(worker, 20ms); - Check(storage_alive.expired(), "last release permits worker reclamation"); + EXPECT_TRUE(storage_alive.expired()) + << "last release permits worker reclamation"; } } - worker.RequestStop(); co_return absl::OkStatus(); } -} // namespace - -int main() { - bycorf::Runtime runtime; - runtime.Start(1, [](unsigned, bycorf::Worker& worker) { - if (!worker.Init().ok()) return 1; - worker.Spawn(CheckStorage(worker)); - worker.Run(); - worker.Shutdown(); - return failures == 0 ? 0 : 1; - }); - runtime.WaitUntilStopped(); - return runtime.exit_code(); +TEST(ConnectionStorage, BorrowSurvivesCloseUntilLastRelease) { + bycorf::test::RunOnWorker(CheckStorage); } + +} // namespace diff --git a/tests/dpdk_adaptive_check.cpp b/tests/dpdk_adaptive_test.cpp similarity index 81% rename from tests/dpdk_adaptive_check.cpp rename to tests/dpdk_adaptive_test.cpp index 79e1226..a013a9b 100644 --- a/tests/dpdk_adaptive_check.cpp +++ b/tests/dpdk_adaptive_test.cpp @@ -14,6 +14,7 @@ * limitations under the License. */ +#include #include #include #include @@ -23,6 +24,7 @@ #include #include #include +#include #include #include @@ -98,9 +100,9 @@ extern "C" int __wrap_rte_eth_dev_rx_intr_disable(uint16_t, uint16_t) { return 0; } -int main(int argc, char** argv) { - if (argc != 2) return 2; - scenario = argv[1]; +namespace { +int RunScenario(std::string_view selected) { + scenario = selected; const bool poll = scenario == "poll"; const bool mlx5 = scenario == "mlx5-empty"; const bool runtime = scenario.starts_with("runtime-"); @@ -144,4 +146,26 @@ int main(int argc, char** argv) { } close(notification_fd); std::cout << "PASS " << scenario << '\n'; + return 0; } + +class DpdkAdaptiveDeathTest : public testing::TestWithParam {}; + +TEST_P(DpdkAdaptiveDeathTest, HandlesInterruptFailure) { + // EAL, backend selection, and FreeBSD bootstrap have process lifetime. + // GoogleTest discovery must only list cases, never initialize those states. + const auto selected = GetParam(); + ASSERT_EXIT(std::exit(RunScenario(selected)), testing::ExitedWithCode(0), ""); +} + +INSTANTIATE_TEST_SUITE_P(Scenarios, DpdkAdaptiveDeathTest, + testing::Values("configure", "descriptor", "enable", + "disable", "poll", "runtime-enable", + "runtime-disable", "mlx5-empty"), + [](const testing::TestParamInfo& info) { + std::string name = info.param; + for (auto& character : name) + if (character == '-') character = '_'; + return name; + }); +} // namespace diff --git a/tests/freebsd_local_connect_check.cpp b/tests/freebsd_local_connect_test.cpp similarity index 94% rename from tests/freebsd_local_connect_check.cpp rename to tests/freebsd_local_connect_test.cpp index 734c662..beb2c0d 100644 --- a/tests/freebsd_local_connect_check.cpp +++ b/tests/freebsd_local_connect_test.cpp @@ -3,6 +3,7 @@ // Real BSD sockets and packet input, without DPDK, TAP, or kernel TCP. #include +#include #include #include @@ -69,9 +70,8 @@ void Exchange(struct socket* source, struct socket* target) { actual == expected, "segmented binary local payload"); } -} // namespace -int main() { +void CheckLocalConnect() { bycorf_bsd_host host{ [](size_t n, size_t alignment) -> void* { void* p = nullptr; @@ -167,3 +167,15 @@ int main() { "PASS BSD local connect, binary duplex transfer, EOF, refusal and " "physical source validation"); } + +TEST(FreeBsdLocalConnectDeathTest, DuplexTransferRejectionEofAndRefusal) { + // The native stack cannot be reinitialized in-process. Its C callbacks use + // fail-fast checks; the exit assertion reports those failures for this case. + ASSERT_EXIT( + { + CheckLocalConnect(); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} +} // namespace diff --git a/tests/io_uring_backpressure_check.cpp b/tests/io_uring_backpressure_check.cpp deleted file mode 100644 index 1c0d060..0000000 --- a/tests/io_uring_backpressure_check.cpp +++ /dev/null @@ -1,408 +0,0 @@ -/* - * Copyright (C) 2026 EloqData Inc. - * - * 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 - * - * https://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. - */ - -#include -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include - -#include "bycorf/io/io_uring_backend.h" -#include "bycorf/runtime/worker.h" - -namespace { - -#define CHECK(condition) \ - do { \ - if (!(condition)) { \ - std::cerr << "FAIL line " << __LINE__ << ": " #condition << '\n'; \ - std::exit(1); \ - } \ - } while (false) - -struct Completion final : bycorf::IoCompletion { - unsigned calls = 0; - int result = 0; - - void Complete(bycorf::Worker&, int value, unsigned flags) override { - CHECK(++calls == 1); - CHECK(flags == bycorf::kCompletionNone); - result = value; - } -}; - -struct Fixture { - Fixture() { - const auto status = backend.Init( - {.ring_entries_ = 2, .recv_buffer_count_ = 0}, &worker, -1); - if (!status.ok()) std::cerr << status << '\n'; - CHECK(status.ok()); - } - - bycorf::Worker worker; - bycorf::IoUringBackend backend; -}; - -template -void Drain(bycorf::IoUringBackend& backend, Predicate done) { - const auto deadline = - std::chrono::steady_clock::now() + std::chrono::seconds(5); - while (!done()) { - CHECK(std::chrono::steady_clock::now() < deadline); - const auto status = backend.Submit(); - CHECK(status.ok() || absl::IsUnavailable(status)); - backend.Poll(); - } -} - -void CheckSendsAndWake() { - Fixture source, peer; - int sockets[2]; - CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, - sockets) == 0); - constexpr std::size_t kCount = 64; - std::array tags; - std::array payload; - std::array vectors; - std::array messages{}; - // Submit the initial wake poll so both SQ slots are available to the burst. - CHECK(source.backend.Submit().ok()); - for (std::size_t i = 0; i < kCount; ++i) { - payload[i] = std::byte(i); - vectors[i] = {.iov_base = &payload[i], .iov_len = 1}; - messages[i].msg_iov = &vectors[i]; - messages[i].msg_iovlen = 1; - // Exercise both preparation callbacks in the real SQ and the pending FIFO. - // Indirect sendmsg arguments remain caller-owned through completion. - CHECK((i % 2 == 0 ? source.backend.SubmitSend( - {.fd_ = sockets[0]}, - std::span(payload).subspan(i, 1), &tags[i]) - : source.backend.SubmitSendMsg({.fd_ = sockets[0]}, - &messages[i], &tags[i])) - .ok()); - } - source.backend.WakeRemote(peer.backend.WakeHandle()); - for (const auto& tag : tags) CHECK(tag.calls == 0); - - // A completion-only check misses hidden io_uring_enter calls that free SQ - // slots without reaping. The peer must see no data before the event loop - // explicitly submits this burst, even though it exceeds the ring capacity. - std::array received; - CHECK(recv(sockets[1], received.data(), received.size(), MSG_DONTWAIT) == -1); - CHECK(errno == EAGAIN || errno == EWOULDBLOCK); - CHECK(peer.backend.Submit().ok()); - CHECK(!peer.backend.Poll()); - - bool peer_woken = false; - Drain(source.backend, [&] { - // DEFER_TASKRUN can defer MSG_RING delivery until the target ring enters - // the kernel, even when both rings share a thread. Drive the peer as its - // worker would: Submit runs deferred task work; Poll consumes ready CQEs. - const auto status = peer.backend.Submit(); - CHECK(status.ok() || absl::IsUnavailable(status)); - peer_woken |= peer.backend.Poll(); - for (const auto& tag : tags) - if (tag.calls == 0) return false; - return peer_woken; - }); - for (const auto& tag : tags) CHECK(tag.result == 1); - CHECK(recv(sockets[1], received.data(), received.size(), MSG_DONTWAIT) == - static_cast(received.size())); - CHECK(received == payload); - CHECK(close(sockets[0]) == 0); - CHECK(close(sockets[1]) == 0); - std::cout << "PASS deferred sends, FIFO payloads and queued peer wake\n"; -} - -void CheckQueuedCancellation() { - Fixture fixture; - const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); - CHECK(event >= 0); - std::array tags; - for (std::size_t i = 0; i + 1 < tags.size(); ++i) - CHECK(fixture.backend.SubmitPoll(event, POLLIN, &tags[i]).ok()); - const __kernel_timespec timeout{.tv_sec = 60, .tv_nsec = 0}; - CHECK(fixture.backend.SubmitTimeout(timeout, &tags.back()).ok()); - // Both the target and its cancel are behind a full SQ. Cancellation must - // still deliver the target's terminal completion before its tag is released. - CHECK(fixture.backend.SubmitCancel(&tags.back()).ok()); - for (const auto& tag : tags) CHECK(tag.calls == 0); - Drain(fixture.backend, [&] { return tags.back().calls != 0; }); - CHECK(tags.back().result == -ECANCELED); - for (std::size_t i = 0; i + 1 < tags.size(); ++i) CHECK(tags[i].calls == 0); - - const std::uint64_t one = 1; - CHECK(write(event, &one, sizeof(one)) == sizeof(one)); - Drain(fixture.backend, [&] { - for (const auto& tag : tags) - if (tag.calls == 0) return false; - return true; - }); - for (std::size_t i = 0; i + 1 < tags.size(); ++i) - CHECK((tags[i].result & POLLIN) != 0); - CHECK(fixture.backend.Submit().ok()); - fixture.backend.Poll(); - CHECK(tags.back().calls == 1); - CHECK(close(event) == 0); - std::cout << "PASS queued cancellation and exactly-once completions\n"; -} - -void CheckClosedDescriptorReuse() { - Fixture fixture; - int original[2], replacement[2]; - CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, - original) == 0); - CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, - replacement) == 0); - const std::array payload{std::byte{0x5a}}; - std::array tags; - // Initialization occupies one SQ slot; the first send is in the real SQ, - // while later sends are software-queued. Neither may outlive this fd's - // identity, and cancellation callbacks must remain deferred. - for (auto& tag : tags) - CHECK(fixture.backend.SubmitSend({.fd_ = original[0]}, payload, &tag).ok()); - fixture.backend.CancelUnsubmittedForFd(original[0]); - for (const auto& tag : tags) CHECK(tag.calls == 0); - CHECK(close(original[0]) == 0); - CHECK(dup2(replacement[0], original[0]) == original[0]); - CHECK(close(replacement[0]) == 0); - - Drain(fixture.backend, [&] { - for (const auto& tag : tags) - if (tag.calls == 0) return false; - return true; - }); - for (const auto& tag : tags) CHECK(tag.result == -ECANCELED); - std::byte received{}; - CHECK(recv(original[1], &received, 1, MSG_DONTWAIT) == 0); - CHECK(recv(replacement[1], &received, 1, MSG_DONTWAIT) == -1); - CHECK(errno == EAGAIN || errno == EWOULDBLOCK); - - Completion fresh; - CHECK(fixture.backend.SubmitSend({.fd_ = original[0]}, payload, &fresh).ok()); - Drain(fixture.backend, [&] { return fresh.calls != 0; }); - CHECK(fresh.result == 1); - CHECK(recv(replacement[1], &received, 1, MSG_DONTWAIT) == 1); - CHECK(received == payload.front()); - CHECK(close(original[0]) == 0); - CHECK(close(original[1]) == 0); - CHECK(close(replacement[1]) == 0); - std::cout << "PASS cancellation before close protects reused descriptors\n"; -} - -void CheckQueuedReceiveCancellation() { - Fixture fixture; - const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); - CHECK(event >= 0); - std::array blockers; - for (auto& tag : blockers) - CHECK(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); - - int sockets[2]; - CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, - sockets) == 0); - bycorf::Connection connection; - connection.worker_ = &fixture.worker; - connection.file_.fd_ = sockets[0]; - connection.closed_ = false; - CHECK(send(sockets[1], "x", 1, MSG_NOSIGNAL) == 1); - CHECK(fixture.backend.StartRecvMultishot(&connection).ok()); - CHECK(connection.recv_armed_ && connection.inflight_ops_ == 1); - connection.recv_paused_ = true; - Completion cancelled; - CHECK(fixture.backend.SubmitCancelRecv(&connection, &cancelled).ok()); - CHECK(cancelled.calls == 0); - CHECK(connection.recv_armed_ && connection.inflight_ops_ == 1); - Drain(fixture.backend, [&] { return cancelled.calls != 0; }); - CHECK(cancelled.result == 0); - CHECK(!connection.recv_armed_ && connection.inflight_ops_ == 0); - CHECK(connection.received_buffers_.empty()); - char received = 0; - CHECK(recv(sockets[0], &received, 1, MSG_DONTWAIT) == 1); - CHECK(received == 'x'); - - const std::uint64_t one = 1; - CHECK(write(event, &one, sizeof(one)) == sizeof(one)); - Drain(fixture.backend, [&] { - for (const auto& tag : blockers) - if (tag.calls == 0) return false; - return true; - }); - CHECK(close(event) == 0); - CHECK(close(sockets[0]) == 0); - CHECK(close(sockets[1]) == 0); - std::cout << "PASS queued receive cancellation drains connection ownership\n"; -} - -void CheckLateCancelBeforeTagReuse() { - Fixture fixture; - const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); - CHECK(event >= 0); - int sockets[2]; - CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, - sockets) == 0); - bycorf::Connection connection; - connection.worker_ = &fixture.worker; - connection.file_.fd_ = sockets[0]; - connection.closed_ = false; - CHECK(fixture.backend.StartRecvMultishot(&connection).ok()); - CHECK(fixture.backend.Submit().ok()); - CHECK(!fixture.backend.HasPending()); - - std::array blockers; - for (auto& tag : blockers) - CHECK(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); - Completion cancelled; - CHECK(fixture.backend.SubmitCancelRecv(&connection, &cancelled).ok()); - CHECK(cancelled.calls == 0); - CHECK(send(sockets[1], "x", 1, MSG_NOSIGNAL) == 1); - // Run deferred kernel completion work without publishing or submitting the - // full SQ/backlog. The receive therefore finishes naturally while its cancel - // is still wholly owned by the backend's submission queue. - alarm(2); - CHECK(io_uring_enter(fixture.backend.WakeHandle(), 0, 1, - IORING_ENTER_GETEVENTS, nullptr) >= 0); - alarm(0); - CHECK(fixture.backend.Poll()); - CHECK(!connection.recv_armed_ && connection.inflight_ops_ == 0); - CHECK(connection.received_buffers_.size() == 1); - CHECK(connection.read_buffer_.front() == std::byte{'x'}); - // This must be delivered by the same Poll, before any queued cancel can - // enter the kernel. Merely checking the next read would allow FIFO submission - // to hide a stale cancel that happens to run before the reused tag is armed. - CHECK(cancelled.calls == 1 && cancelled.result == -ENOENT); - - connection.received_buffers_.clear(); - CHECK(fixture.backend.StartRecvMultishot(&connection).ok()); - CHECK(connection.recv_armed_ && connection.inflight_ops_ == 1); - CHECK(send(sockets[1], "y", 1, MSG_NOSIGNAL) == 1); - const std::uint64_t one = 1; - CHECK(write(event, &one, sizeof(one)) == sizeof(one)); - Drain(fixture.backend, [&] { - for (const auto& tag : blockers) - if (tag.calls == 0) return false; - return !connection.received_buffers_.empty(); - }); - CHECK(cancelled.calls == 1); - CHECK(!connection.recv_armed_ && connection.inflight_ops_ == 0); - CHECK(connection.read_buffer_.front() == std::byte{'y'}); - CHECK(close(event) == 0); - CHECK(close(sockets[0]) == 0); - CHECK(close(sockets[1]) == 0); - std::cout - << "PASS natural completion retires queued cancel before tag reuse\n"; -} - -void CheckPendingWaitAndShutdown() { - Fixture fixture; - const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); - CHECK(event >= 0); - { - std::array tags; - for (auto& tag : tags) - CHECK(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); - CHECK(fixture.backend.SubmitCancel(&tags.back()).ok()); - CHECK(fixture.backend.HasPending()); - // None of these polls can complete. Blocking here would strand both the - // unsubmitted work and the software cancellation completion behind it. - alarm(2); - CHECK(fixture.backend.Wait(-1)); - alarm(0); - for (const auto& tag : tags) CHECK(tag.calls == 0); - fixture.backend.Shutdown(); - CHECK(!fixture.backend.HasPending()); - CHECK(!fixture.backend.Poll()); - } - CHECK(close(event) == 0); - - // Old tags are now destroyed. Reusing the backend must not issue abandoned - // SQEs or dispatch their synthetic completions into the old caller's state. - CHECK(fixture.backend - .Init({.ring_entries_ = 2, .recv_buffer_count_ = 0}, - &fixture.worker, -1) - .ok()); - Completion fresh; - const __kernel_timespec timeout{.tv_sec = 0, .tv_nsec = 1}; - CHECK(fixture.backend.SubmitTimeout(timeout, &fresh).ok()); - Drain(fixture.backend, [&] { return fresh.calls != 0; }); - CHECK(fresh.result == -ETIME); - std::cout << "PASS pending work prevents park and shutdown drops old state\n"; -} - -void CheckStorageReads() { - Fixture fixture; - char path[] = "/tmp/bycorf-uring-backpressure-XXXXXX"; - const int file = mkstemp(path); - CHECK(file >= 0); - std::array payload; - for (std::size_t i = 0; i < payload.size(); ++i) payload[i] = std::byte(i); - CHECK(write(file, payload.data(), payload.size()) == - static_cast(payload.size())); - CHECK(fixture.backend.RegisterFixedFiles(1).ok()); - Completion opened; - CHECK(fixture.backend.SubmitOpenDirect(path, O_RDONLY, 0, {0}, &opened).ok()); - Drain(fixture.backend, [&] { return opened.calls != 0; }); - CHECK(opened.result == 0); - CHECK(unlink(path) == 0); - CHECK(close(file) == 0); - - std::array tags; - std::array received{}; - const iovec region{.iov_base = received.data(), .iov_len = received.size()}; - CHECK(fixture.backend.RegisterBuffers(std::span(®ion, 1)).ok()); - for (std::size_t i = 0; i < tags.size(); ++i) { - const auto buffer = std::span(received).subspan(i * 4, 4); - CHECK((i % 2 == 0 - ? fixture.backend.SubmitRead({0}, buffer, i * 4, &tags[i]) - : fixture.backend.SubmitReadFixed( - {0}, {.data_ = buffer.data(), .size_ = buffer.size()}, - i * 4, &tags[i])) - .ok()); - } - for (const auto& tag : tags) CHECK(tag.calls == 0); - Drain(fixture.backend, [&] { - for (const auto& tag : tags) - if (tag.calls == 0) return false; - return true; - }); - for (const auto& tag : tags) CHECK(tag.result == 4); - CHECK(received == payload); - Completion closed; - CHECK(fixture.backend.SubmitCloseDirect({0}, &closed).ok()); - Drain(fixture.backend, [&] { return closed.calls != 0; }); - CHECK(closed.result == 0); - std::cout << "PASS queued fixed-file reads preserve buffers and offsets\n"; -} - -} // namespace - -int main() { - CheckSendsAndWake(); - CheckQueuedCancellation(); - CheckClosedDescriptorReuse(); - CheckQueuedReceiveCancellation(); - CheckLateCancelBeforeTagReuse(); - CheckPendingWaitAndShutdown(); - CheckStorageReads(); - return 0; -} diff --git a/tests/io_uring_backpressure_test.cpp b/tests/io_uring_backpressure_test.cpp new file mode 100644 index 0000000..de96c82 --- /dev/null +++ b/tests/io_uring_backpressure_test.cpp @@ -0,0 +1,403 @@ +/* + * Copyright (C) 2026 EloqData Inc. + * + * 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 + * + * https://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. + */ + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "bycorf/io/io_uring_backend.h" +#include "bycorf/runtime/worker.h" + +namespace { + +struct Completion final : bycorf::IoCompletion { + unsigned calls = 0; + int result = 0; + + void Complete(bycorf::Worker&, int value, unsigned flags) override { + ASSERT_TRUE(++calls == 1); + ASSERT_TRUE(flags == bycorf::kCompletionNone); + result = value; + } +}; + +struct Fixture { + void Init() { + const auto status = backend.Init( + {.ring_entries_ = 2, .recv_buffer_count_ = 0}, &worker, -1); + if (!status.ok()) std::cerr << status << '\n'; + ASSERT_TRUE(status.ok()); + } + + bycorf::Worker worker; + bycorf::IoUringBackend backend; +}; + +template +void Drain(bycorf::IoUringBackend& backend, Predicate done) { + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!done()) { + ASSERT_TRUE(std::chrono::steady_clock::now() < deadline); + const auto status = backend.Submit(); + ASSERT_TRUE(status.ok() || absl::IsUnavailable(status)); + backend.Poll(); + } +} + +TEST(IoUringBackpressure, SendsAndWake) { + Fixture source, peer; + ASSERT_NO_FATAL_FAILURE(source.Init()); + ASSERT_NO_FATAL_FAILURE(peer.Init()); + int sockets[2]; + ASSERT_TRUE(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + sockets) == 0); + constexpr std::size_t kCount = 64; + std::array tags; + std::array payload; + std::array vectors; + std::array messages{}; + // Submit the initial wake poll so both SQ slots are available to the burst. + ASSERT_TRUE(source.backend.Submit().ok()); + for (std::size_t i = 0; i < kCount; ++i) { + payload[i] = std::byte(i); + vectors[i] = {.iov_base = &payload[i], .iov_len = 1}; + messages[i].msg_iov = &vectors[i]; + messages[i].msg_iovlen = 1; + // Exercise both preparation callbacks in the real SQ and the pending FIFO. + // Indirect sendmsg arguments remain caller-owned through completion. + ASSERT_TRUE((i % 2 == 0 ? source.backend.SubmitSend( + {.fd_ = sockets[0]}, + std::span(payload).subspan(i, 1), &tags[i]) + : source.backend.SubmitSendMsg( + {.fd_ = sockets[0]}, &messages[i], &tags[i])) + .ok()); + } + source.backend.WakeRemote(peer.backend.WakeHandle()); + for (const auto& tag : tags) ASSERT_TRUE(tag.calls == 0); + + // A completion-only check misses hidden io_uring_enter calls that free SQ + // slots without reaping. The peer must see no data before the event loop + // explicitly submits this burst, even though it exceeds the ring capacity. + std::array received; + ASSERT_TRUE( + recv(sockets[1], received.data(), received.size(), MSG_DONTWAIT) == -1); + ASSERT_TRUE(errno == EAGAIN || errno == EWOULDBLOCK); + ASSERT_TRUE(peer.backend.Submit().ok()); + ASSERT_TRUE(!peer.backend.Poll()); + + bool peer_woken = false; + Drain(source.backend, [&] { + // DEFER_TASKRUN can defer MSG_RING delivery until the target ring enters + // the kernel, even when both rings share a thread. Drive the peer as its + // worker would: Submit runs deferred task work; Poll consumes ready CQEs. + const auto status = peer.backend.Submit(); + EXPECT_TRUE(status.ok() || absl::IsUnavailable(status)) << status; + peer_woken |= peer.backend.Poll(); + for (const auto& tag : tags) + if (tag.calls == 0) return false; + return peer_woken; + }); + for (const auto& tag : tags) ASSERT_TRUE(tag.result == 1); + ASSERT_TRUE(recv(sockets[1], received.data(), received.size(), + MSG_DONTWAIT) == static_cast(received.size())); + ASSERT_TRUE(received == payload); + ASSERT_TRUE(close(sockets[0]) == 0); + ASSERT_TRUE(close(sockets[1]) == 0); + std::cout << "PASS deferred sends, FIFO payloads and queued peer wake\n"; +} + +TEST(IoUringBackpressure, QueuedCancellation) { + Fixture fixture; + ASSERT_NO_FATAL_FAILURE(fixture.Init()); + const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + ASSERT_TRUE(event >= 0); + std::array tags; + for (std::size_t i = 0; i + 1 < tags.size(); ++i) + ASSERT_TRUE(fixture.backend.SubmitPoll(event, POLLIN, &tags[i]).ok()); + const __kernel_timespec timeout{.tv_sec = 60, .tv_nsec = 0}; + ASSERT_TRUE(fixture.backend.SubmitTimeout(timeout, &tags.back()).ok()); + // Both the target and its cancel are behind a full SQ. Cancellation must + // still deliver the target's terminal completion before its tag is released. + ASSERT_TRUE(fixture.backend.SubmitCancel(&tags.back()).ok()); + for (const auto& tag : tags) ASSERT_TRUE(tag.calls == 0); + Drain(fixture.backend, [&] { return tags.back().calls != 0; }); + ASSERT_TRUE(tags.back().result == -ECANCELED); + for (std::size_t i = 0; i + 1 < tags.size(); ++i) + ASSERT_TRUE(tags[i].calls == 0); + + const std::uint64_t one = 1; + ASSERT_TRUE(write(event, &one, sizeof(one)) == sizeof(one)); + Drain(fixture.backend, [&] { + for (const auto& tag : tags) + if (tag.calls == 0) return false; + return true; + }); + for (std::size_t i = 0; i + 1 < tags.size(); ++i) + ASSERT_TRUE((tags[i].result & POLLIN) != 0); + ASSERT_TRUE(fixture.backend.Submit().ok()); + fixture.backend.Poll(); + ASSERT_TRUE(tags.back().calls == 1); + ASSERT_TRUE(close(event) == 0); + std::cout << "PASS queued cancellation and exactly-once completions\n"; +} + +TEST(IoUringBackpressure, ClosedDescriptorReuse) { + Fixture fixture; + ASSERT_NO_FATAL_FAILURE(fixture.Init()); + int original[2], replacement[2]; + ASSERT_TRUE(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + original) == 0); + ASSERT_TRUE(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + replacement) == 0); + const std::array payload{std::byte{0x5a}}; + std::array tags; + // Initialization occupies one SQ slot; the first send is in the real SQ, + // while later sends are software-queued. Neither may outlive this fd's + // identity, and cancellation callbacks must remain deferred. + for (auto& tag : tags) + ASSERT_TRUE( + fixture.backend.SubmitSend({.fd_ = original[0]}, payload, &tag).ok()); + fixture.backend.CancelUnsubmittedForFd(original[0]); + for (const auto& tag : tags) ASSERT_TRUE(tag.calls == 0); + ASSERT_TRUE(close(original[0]) == 0); + ASSERT_TRUE(dup2(replacement[0], original[0]) == original[0]); + ASSERT_TRUE(close(replacement[0]) == 0); + + Drain(fixture.backend, [&] { + for (const auto& tag : tags) + if (tag.calls == 0) return false; + return true; + }); + for (const auto& tag : tags) ASSERT_TRUE(tag.result == -ECANCELED); + std::byte received{}; + ASSERT_TRUE(recv(original[1], &received, 1, MSG_DONTWAIT) == 0); + ASSERT_TRUE(recv(replacement[1], &received, 1, MSG_DONTWAIT) == -1); + ASSERT_TRUE(errno == EAGAIN || errno == EWOULDBLOCK); + + Completion fresh; + ASSERT_TRUE( + fixture.backend.SubmitSend({.fd_ = original[0]}, payload, &fresh).ok()); + Drain(fixture.backend, [&] { return fresh.calls != 0; }); + ASSERT_TRUE(fresh.result == 1); + ASSERT_TRUE(recv(replacement[1], &received, 1, MSG_DONTWAIT) == 1); + ASSERT_TRUE(received == payload.front()); + ASSERT_TRUE(close(original[0]) == 0); + ASSERT_TRUE(close(original[1]) == 0); + ASSERT_TRUE(close(replacement[1]) == 0); + std::cout << "PASS cancellation before close protects reused descriptors\n"; +} + +TEST(IoUringBackpressure, QueuedReceiveCancellation) { + Fixture fixture; + ASSERT_NO_FATAL_FAILURE(fixture.Init()); + const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + ASSERT_TRUE(event >= 0); + std::array blockers; + for (auto& tag : blockers) + ASSERT_TRUE(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); + + int sockets[2]; + ASSERT_TRUE(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + sockets) == 0); + bycorf::Connection connection; + connection.worker_ = &fixture.worker; + connection.file_.fd_ = sockets[0]; + connection.closed_ = false; + ASSERT_TRUE(send(sockets[1], "x", 1, MSG_NOSIGNAL) == 1); + ASSERT_TRUE(fixture.backend.StartRecvMultishot(&connection).ok()); + ASSERT_TRUE(connection.recv_armed_ && connection.inflight_ops_ == 1); + connection.recv_paused_ = true; + Completion cancelled; + ASSERT_TRUE(fixture.backend.SubmitCancelRecv(&connection, &cancelled).ok()); + ASSERT_TRUE(cancelled.calls == 0); + ASSERT_TRUE(connection.recv_armed_ && connection.inflight_ops_ == 1); + Drain(fixture.backend, [&] { return cancelled.calls != 0; }); + ASSERT_TRUE(cancelled.result == 0); + ASSERT_TRUE(!connection.recv_armed_ && connection.inflight_ops_ == 0); + ASSERT_TRUE(connection.received_buffers_.empty()); + char received = 0; + ASSERT_TRUE(recv(sockets[0], &received, 1, MSG_DONTWAIT) == 1); + ASSERT_TRUE(received == 'x'); + + const std::uint64_t one = 1; + ASSERT_TRUE(write(event, &one, sizeof(one)) == sizeof(one)); + Drain(fixture.backend, [&] { + for (const auto& tag : blockers) + if (tag.calls == 0) return false; + return true; + }); + ASSERT_TRUE(close(event) == 0); + ASSERT_TRUE(close(sockets[0]) == 0); + ASSERT_TRUE(close(sockets[1]) == 0); + std::cout << "PASS queued receive cancellation drains connection ownership\n"; +} + +TEST(IoUringBackpressure, LateCancelBeforeTagReuse) { + Fixture fixture; + ASSERT_NO_FATAL_FAILURE(fixture.Init()); + const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + ASSERT_TRUE(event >= 0); + int sockets[2]; + ASSERT_TRUE(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + sockets) == 0); + bycorf::Connection connection; + connection.worker_ = &fixture.worker; + connection.file_.fd_ = sockets[0]; + connection.closed_ = false; + ASSERT_TRUE(fixture.backend.StartRecvMultishot(&connection).ok()); + ASSERT_TRUE(fixture.backend.Submit().ok()); + ASSERT_TRUE(!fixture.backend.HasPending()); + + std::array blockers; + for (auto& tag : blockers) + ASSERT_TRUE(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); + Completion cancelled; + ASSERT_TRUE(fixture.backend.SubmitCancelRecv(&connection, &cancelled).ok()); + ASSERT_TRUE(cancelled.calls == 0); + ASSERT_TRUE(send(sockets[1], "x", 1, MSG_NOSIGNAL) == 1); + // Run deferred kernel completion work without publishing or submitting the + // full SQ/backlog. The receive therefore finishes naturally while its cancel + // is still wholly owned by the backend's submission queue. + alarm(2); + ASSERT_TRUE(io_uring_enter(fixture.backend.WakeHandle(), 0, 1, + IORING_ENTER_GETEVENTS, nullptr) >= 0); + alarm(0); + ASSERT_TRUE(fixture.backend.Poll()); + ASSERT_TRUE(!connection.recv_armed_ && connection.inflight_ops_ == 0); + ASSERT_TRUE(connection.received_buffers_.size() == 1); + ASSERT_TRUE(connection.read_buffer_.front() == std::byte{'x'}); + // This must be delivered by the same Poll, before any queued cancel can + // enter the kernel. Merely checking the next read would allow FIFO submission + // to hide a stale cancel that happens to run before the reused tag is armed. + ASSERT_TRUE(cancelled.calls == 1 && cancelled.result == -ENOENT); + + connection.received_buffers_.clear(); + ASSERT_TRUE(fixture.backend.StartRecvMultishot(&connection).ok()); + ASSERT_TRUE(connection.recv_armed_ && connection.inflight_ops_ == 1); + ASSERT_TRUE(send(sockets[1], "y", 1, MSG_NOSIGNAL) == 1); + const std::uint64_t one = 1; + ASSERT_TRUE(write(event, &one, sizeof(one)) == sizeof(one)); + Drain(fixture.backend, [&] { + for (const auto& tag : blockers) + if (tag.calls == 0) return false; + return !connection.received_buffers_.empty(); + }); + ASSERT_TRUE(cancelled.calls == 1); + ASSERT_TRUE(!connection.recv_armed_ && connection.inflight_ops_ == 0); + ASSERT_TRUE(connection.read_buffer_.front() == std::byte{'y'}); + ASSERT_TRUE(close(event) == 0); + ASSERT_TRUE(close(sockets[0]) == 0); + ASSERT_TRUE(close(sockets[1]) == 0); + std::cout + << "PASS natural completion retires queued cancel before tag reuse\n"; +} + +TEST(IoUringBackpressure, PendingWaitAndShutdown) { + Fixture fixture; + ASSERT_NO_FATAL_FAILURE(fixture.Init()); + const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + ASSERT_TRUE(event >= 0); + { + std::array tags; + for (auto& tag : tags) + ASSERT_TRUE(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); + ASSERT_TRUE(fixture.backend.SubmitCancel(&tags.back()).ok()); + ASSERT_TRUE(fixture.backend.HasPending()); + // None of these polls can complete. Blocking here would strand both the + // unsubmitted work and the software cancellation completion behind it. + alarm(2); + ASSERT_TRUE(fixture.backend.Wait(-1)); + alarm(0); + for (const auto& tag : tags) ASSERT_TRUE(tag.calls == 0); + fixture.backend.Shutdown(); + ASSERT_TRUE(!fixture.backend.HasPending()); + ASSERT_TRUE(!fixture.backend.Poll()); + } + ASSERT_TRUE(close(event) == 0); + + // Old tags are now destroyed. Reusing the backend must not issue abandoned + // SQEs or dispatch their synthetic completions into the old caller's state. + ASSERT_TRUE(fixture.backend + .Init({.ring_entries_ = 2, .recv_buffer_count_ = 0}, + &fixture.worker, -1) + .ok()); + Completion fresh; + const __kernel_timespec timeout{.tv_sec = 0, .tv_nsec = 1}; + ASSERT_TRUE(fixture.backend.SubmitTimeout(timeout, &fresh).ok()); + Drain(fixture.backend, [&] { return fresh.calls != 0; }); + ASSERT_TRUE(fresh.result == -ETIME); + std::cout << "PASS pending work prevents park and shutdown drops old state\n"; +} + +TEST(IoUringBackpressure, StorageReads) { + Fixture fixture; + ASSERT_NO_FATAL_FAILURE(fixture.Init()); + char path[] = "/tmp/bycorf-uring-backpressure-XXXXXX"; + const int file = mkstemp(path); + ASSERT_TRUE(file >= 0); + std::array payload; + for (std::size_t i = 0; i < payload.size(); ++i) payload[i] = std::byte(i); + ASSERT_TRUE(write(file, payload.data(), payload.size()) == + static_cast(payload.size())); + ASSERT_TRUE(fixture.backend.RegisterFixedFiles(1).ok()); + Completion opened; + ASSERT_TRUE( + fixture.backend.SubmitOpenDirect(path, O_RDONLY, 0, {0}, &opened).ok()); + Drain(fixture.backend, [&] { return opened.calls != 0; }); + ASSERT_TRUE(opened.result == 0); + ASSERT_TRUE(unlink(path) == 0); + ASSERT_TRUE(close(file) == 0); + + std::array tags; + std::array received{}; + const iovec region{.iov_base = received.data(), .iov_len = received.size()}; + ASSERT_TRUE(fixture.backend.RegisterBuffers(std::span(®ion, 1)).ok()); + for (std::size_t i = 0; i < tags.size(); ++i) { + const auto buffer = std::span(received).subspan(i * 4, 4); + ASSERT_TRUE( + (i % 2 == 0 ? fixture.backend.SubmitRead({0}, buffer, i * 4, &tags[i]) + : fixture.backend.SubmitReadFixed( + {0}, {.data_ = buffer.data(), .size_ = buffer.size()}, + i * 4, &tags[i])) + .ok()); + } + for (const auto& tag : tags) ASSERT_TRUE(tag.calls == 0); + Drain(fixture.backend, [&] { + for (const auto& tag : tags) + if (tag.calls == 0) return false; + return true; + }); + for (const auto& tag : tags) ASSERT_TRUE(tag.result == 4); + ASSERT_TRUE(received == payload); + Completion closed; + ASSERT_TRUE(fixture.backend.SubmitCloseDirect({0}, &closed).ok()); + Drain(fixture.backend, [&] { return closed.calls != 0; }); + ASSERT_TRUE(closed.result == 0); + std::cout << "PASS queued fixed-file reads preserve buffers and offsets\n"; +} + +} // namespace diff --git a/tests/received_buffer_queue_check.cpp b/tests/received_buffer_queue_test.cpp similarity index 76% rename from tests/received_buffer_queue_check.cpp rename to tests/received_buffer_queue_test.cpp index d292551..35e0dc7 100644 --- a/tests/received_buffer_queue_check.cpp +++ b/tests/received_buffer_queue_test.cpp @@ -14,11 +14,10 @@ * limitations under the License. */ -#include +#include + #include -#include #include -#include #include #include "bycorf/net/connection.h" @@ -26,25 +25,17 @@ namespace { using ReceiveQueue = decltype(bycorf::Connection::received_buffers_); -void Check(bool ok, - std::source_location location = std::source_location::current()) { - if (!ok) { - std::cerr << "received buffer queue check failed at line " - << location.line() << "\n"; - std::abort(); - } -} bool Equal(const bycorf::ReceivedBuffer& a, const bycorf::ReceivedBuffer& b) { return a.buffer_id_ == b.buffer_id_ && a.size_ == b.size_ && a.offset_ == b.offset_; } void Compare(const ReceiveQueue& queue, const std::deque& reference) { - Check(queue.size() == reference.size()); + ASSERT_TRUE(queue.size() == reference.size()); for (std::uint32_t i = 0; i < queue.size(); ++i) - Check(Equal(queue[i], reference[i])); + ASSERT_TRUE(Equal(queue[i], reference[i])); } -void CheckInlineAndHeapMoves() { +TEST(ReceivedBufferQueue, InlineAndHeapMoves) { for (bool source_heap : {false, true}) { for (bool destination_heap : {false, true}) { ReceiveQueue destination; @@ -62,14 +53,14 @@ void CheckInlineAndHeapMoves() { expected.push_back(value); } destination = std::move(source); - Check(source.empty() && source.capacity() == 1); + ASSERT_TRUE(source.empty() && source.capacity() == 1); // Reusing and destroying the source must not alias an inline element. source.push_back({.buffer_id_ = 77, .size_ = 7}); Compare(destination, expected); } Compare(destination, expected); ReceiveQueue constructed(std::move(destination)); - Check(destination.empty() && destination.capacity() == 1); + ASSERT_TRUE(destination.empty() && destination.capacity() == 1); destination.push_back({.buffer_id_ = 88}); Compare(constructed, expected); auto& same = constructed; @@ -80,18 +71,18 @@ void CheckInlineAndHeapMoves() { ReceiveQueue small; for (unsigned cycle = 0; cycle < 100; ++cycle) { small.push_back({.buffer_id_ = 1}); - Check(small.capacity() == 1); + ASSERT_TRUE(small.capacity() == 1); small.push_back({.buffer_id_ = 2}); - Check(small.capacity() == 8); + ASSERT_TRUE(small.capacity() == 8); small.pop_front(); - Check(small.front().buffer_id_ == 2 && small.capacity() == 8); + ASSERT_TRUE(small.front().buffer_id_ == 2 && small.capacity() == 8); small.pop_front(); - Check(small.empty() && small.capacity() == 8); + ASSERT_TRUE(small.empty() && small.capacity() == 8); for (unsigned i = 0; i < 6; ++i) { small.push_back({.buffer_id_ = 3}); small.pop_front(); } - Check(small.empty() && small.capacity() == 1); + ASSERT_TRUE(small.empty() && small.capacity() == 1); } } // No default construction is needed, even for over-aligned values. Its copy @@ -107,10 +98,11 @@ template void CheckGenericQueue() { using Queue = bycorf::InlineRingQueue; auto compare = [](const Queue& q, const std::deque& expected) { - Check(q.size() == expected.size()); - Check(q.capacity() >= q.size()); - if (!q.empty()) Check(q.front() == expected.front()); - for (std::uint32_t i = 0; i < q.size(); ++i) Check(q[i] == expected[i]); + ASSERT_TRUE(q.size() == expected.size()); + ASSERT_TRUE(q.capacity() >= q.size()); + if (!q.empty()) ASSERT_TRUE(q.front() == expected.front()); + for (std::uint32_t i = 0; i < q.size(); ++i) + ASSERT_TRUE(q[i] == expected[i]); }; // Exercise wrapped inline and heap sources against both destination modes. for (bool source_heap : {false, true}) { @@ -133,13 +125,13 @@ void CheckGenericQueue() { expected.push_back(T(100 + i)); } destination = std::move(source); - Check(source.empty() && source.capacity() == N); + ASSERT_TRUE(source.empty() && source.capacity() == N); source.push_back(T(777)); compare(destination, expected); } compare(destination, expected); Queue constructed(std::move(destination)); - Check(destination.empty() && destination.capacity() == N); + ASSERT_TRUE(destination.empty() && destination.capacity() == N); destination.push_back(T(888)); compare(constructed, expected); auto& same = constructed; @@ -167,11 +159,11 @@ void CheckGenericQueue() { const auto capacity = source.capacity(); source.clear(); destination = std::move(source); - Check(destination.empty() && destination.capacity() == capacity); - Check(source.empty() && source.capacity() == N); + ASSERT_TRUE(destination.empty() && destination.capacity() == capacity); + ASSERT_TRUE(source.empty() && source.capacity() == N); source.push_back(T(7)); destination.push_back(T(8)); - Check(source.front() == T(7) && destination.front() == T(8)); + ASSERT_TRUE(source.front() == T(7) && destination.front() == T(8)); } } Queue queue; @@ -209,37 +201,42 @@ void CheckGenericQueue() { expected.push_back(T(i)); compare(queue, expected); } - Check(queue.capacity() == N); + ASSERT_TRUE(queue.capacity() == N); queue.push_back(T(123)); - Check(queue.capacity() > N); + ASSERT_TRUE(queue.capacity() > N); const auto capacity = queue.capacity(); queue.clear(); - Check(queue.empty() && queue.capacity() == capacity); + ASSERT_TRUE(queue.empty() && queue.capacity() == capacity); for (unsigned i = 0; i < capacity * 2; ++i) { queue.push_back(T(i)); - Check(queue.front() == T(i)); + ASSERT_TRUE(queue.front() == T(i)); queue.pop_front(); } - Check(queue.capacity() == N); + ASSERT_TRUE(queue.capacity() == N); } -template -void CheckGenericTypes() { - CheckGenericQueue(); - CheckGenericQueue(); +template +struct QueueConfig { + using ValueType = T; + static constexpr std::uint32_t kCapacity = N; +}; +template +class InlineRingQueue : public testing::Test {}; +using QueueConfigs = testing::Types< + QueueConfig, QueueConfig, + QueueConfig, QueueConfig, + QueueConfig, QueueConfig, + QueueConfig, QueueConfig, QueueConfig, + QueueConfig, QueueConfig, QueueConfig>; +TYPED_TEST_SUITE(InlineRingQueue, QueueConfigs); + +TYPED_TEST(InlineRingQueue, MovesGrowthAndShrink) { + CheckGenericQueue(); } -} // namespace -int main() { - CheckGenericTypes<1>(); - CheckGenericTypes<2>(); - CheckGenericTypes<4>(); - CheckGenericTypes<8>(); - CheckGenericTypes<16>(); - CheckGenericTypes<32>(); - CheckInlineAndHeapMoves(); +TEST(ReceivedBufferQueue, FifoWrapGrowthAndShrink) { ReceiveQueue queue; - Check(queue.empty() && queue.capacity() == 1); + ASSERT_TRUE(queue.empty() && queue.capacity() == 1); std::deque reference; std::mt19937 random(20261019); // Alternate sustained growth and draining, with partial front consumption. @@ -255,7 +252,7 @@ int main() { queue.push_back(buffer); reference.push_back(buffer); } else { - Check(Equal(queue.front(), reference.front())); + ASSERT_TRUE(Equal(queue.front(), reference.front())); queue.pop_front(); reference.pop_front(); } @@ -283,23 +280,22 @@ int main() { reference.pop_front(); Compare(queue, reference); } - Check(queue.capacity() < peak && queue.capacity() >= queue.size()); + ASSERT_TRUE(queue.capacity() < peak && queue.capacity() >= queue.size()); auto moved = std::move(queue); - Check(queue.empty() && queue.capacity() == 1); + ASSERT_TRUE(queue.empty() && queue.capacity() == 1); Compare(moved, reference); queue.push_back({.buffer_id_ = 99}); queue = std::move(moved); - Check(moved.empty() && moved.capacity() == 1); + ASSERT_TRUE(moved.empty() && moved.capacity() == 1); Compare(queue, reference); queue.clear(); - Check(queue.empty()); + ASSERT_TRUE(queue.empty()); for (unsigned i = 0; i < 10000; ++i) { queue.push_back({.buffer_id_ = 1, .size_ = 32}); - Check(queue.front().size_ == 32); + ASSERT_TRUE(queue.front().size_ == 32); queue.pop_front(); } - Check(queue.capacity() == 1); - std::cout - << "PASS FIFO, wrap, growth, delayed shrink, partial offsets, " - "move, clear, generic types and inline capacities 1/2/4/8/16/32\n"; + ASSERT_TRUE(queue.capacity() == 1); } + +} // namespace diff --git a/tests/runtime_test.h b/tests/runtime_test.h new file mode 100644 index 0000000..35d8aca --- /dev/null +++ b/tests/runtime_test.h @@ -0,0 +1,51 @@ +// Copyright (C) 2026 EloqData Inc. +// SPDX-License-Identifier: Apache-2.0 + +#ifndef BYCORF_TESTS_RUNTIME_TEST_H_ +#define BYCORF_TESTS_RUNTIME_TEST_H_ + +#include + +#include "bycorf/runtime/runtime.h" + +namespace bycorf::test { + +using WorkerCheck = Task (*)(Worker&); + +inline Task CompleteCheck(Worker& worker, WorkerCheck check, + bool& completed) { + const auto status = co_await check(worker); + EXPECT_TRUE(status.ok()) << status; + completed = true; + worker.RequestStop(); + co_return status; +} + +// Keep coroutine frames and callbacks on their worker through teardown. A +// GoogleTest assertion in a coroutine must be nonfatal (EXPECT_*): ASSERT_* +// returns normally and cannot be used in a coroutine. Check returned statuses +// here as well, so an early co_return cannot silently pass a case. +inline void RunOnWorker(WorkerCheck check, WorkerOptions options = {}) { + Runtime runtime; + runtime.Start( + 1, + [&](unsigned, Worker& worker) { + const auto initialized = worker.Init(options); + EXPECT_TRUE(initialized.ok()) << initialized; + if (!initialized.ok()) return 1; + bool completed = false; + worker.Spawn(CompleteCheck(worker, check, completed)); + worker.Run(); + worker.Shutdown(); + worker.DestroyDetachedTasks(); + EXPECT_TRUE(completed) << "worker stopped before the case completed"; + return completed ? 0 : 1; + }, + /*pin_workers=*/false); + runtime.WaitUntilStopped(); + EXPECT_EQ(runtime.exit_code(), 0); +} + +} // namespace bycorf::test + +#endif // BYCORF_TESTS_RUNTIME_TEST_H_ diff --git a/tests/spdk_backpressure_stub_check.cpp b/tests/spdk_backpressure_stub_test.cpp similarity index 92% rename from tests/spdk_backpressure_stub_check.cpp rename to tests/spdk_backpressure_stub_test.cpp index a2bbb2f..1d57df0 100644 --- a/tests/spdk_backpressure_stub_check.cpp +++ b/tests/spdk_backpressure_stub_test.cpp @@ -18,6 +18,7 @@ // device discovery, environment initialization and NVMe commands are replaced; // the pending FIFO, polling and callback ownership stay real. #include +#include #include #include #include @@ -38,6 +39,8 @@ #include "bycorf/runtime/worker.h" namespace { +// Stub C callbacks cannot return GoogleTest fatal assertions or unwind. +// Fail the isolated child immediately; ASSERT_EXIT attributes it to its case. #define CHECK(condition) \ do { \ if (!(condition)) { \ @@ -607,9 +610,33 @@ void CheckSyntheticShutdown(bycorf::Worker& worker) { } std::cout << "PASS shutdown delivers synthetic open/close/no-VWC flush\n"; } -} // namespace +void CheckShutdown(bycorf::Worker& worker, std::span buffer) { + std::vector shutdown(5000); + for (std::size_t i = 0; i < shutdown.size(); ++i) { + CHECK((i % 2 == 0 ? worker.SubmitRead({0}, buffer, 0, &shutdown[i]) + : worker.SubmitWrite({0}, buffer, 0, &shutdown[i])) + .ok()); + CHECK(shutdown[i].calls == 0); + } + Completion other; + CHECK(worker.SubmitRead({2}, std::span(buffer).first(512), 0, &other).ok()); + CHECK(other.calls == 0); + qpairs[0]->completion_tokens = shutdown.size(); + qpairs[1]->completion_tokens = 1; + worker.Shutdown(); + for (const auto& tag : shutdown) CHECK(tag.calls == 1 && tag.result == 4096); + CHECK(other.calls == 1 && other.result == 512); + CHECK(qpairs.empty()); + std::cout << "PASS shutdown drains in-flight and deferred I/O on the owning " + "worker\n"; +} + +void CheckSyntheticCompletions(bycorf::Worker& worker, std::span) { + worker.Shutdown(); + CheckSyntheticShutdown(worker); +} -int main() { +void RunCase(void (*check)(bycorf::Worker&, std::span)) { CHECK(bycorf::ConfigureIoBackends({.spdk_storage = true}).ok()); for (auto path : paths) CHECK(bycorf::ProbeSpdkStorage(path).ok()); CHECK(environment_calls == 1); @@ -635,32 +662,94 @@ int main() { // Stub commands do not DMA: ordinary memory suffices, with no hugepages, // device binding, root privileges or PCI discovery required. std::array buffer{}; - CheckQueues(worker, buffer); - CheckErrors(worker, buffer); - CheckLargeBursts(worker, buffer); - CheckLateSubmissionFifo(worker, buffer); - CheckCompletionOwnership(worker, buffer); - CheckRetryErrorReentry(worker, buffer); - CheckChannelReopen(worker, buffer); - - std::vector shutdown(5000); - for (std::size_t i = 0; i < shutdown.size(); ++i) { - CHECK((i % 2 == 0 ? worker.SubmitRead({0}, buffer, 0, &shutdown[i]) - : worker.SubmitWrite({0}, buffer, 0, &shutdown[i])) - .ok()); - CHECK(shutdown[i].calls == 0); - } - Completion other; - CHECK(worker.SubmitRead({2}, std::span(buffer).first(512), 0, &other).ok()); - CHECK(other.calls == 0); - qpairs[0]->completion_tokens = shutdown.size(); - qpairs[1]->completion_tokens = 1; + check(worker, buffer); worker.Shutdown(); - for (const auto& tag : shutdown) CHECK(tag.calls == 1 && tag.result == 4096); - CHECK(other.calls == 1 && other.result == 512); CHECK(qpairs.empty()); - std::cout << "PASS shutdown drains in-flight and deferred I/O on the owning " - "worker\n"; - CheckSyntheticShutdown(worker); close(wake_fd); } + +// SPDK metadata and the allocator selection are process-global. Each child +// starts with fresh stubs and qpairs even when the binary runs every case or +// repeats them in shuffled order. +TEST(SpdkBackpressureDeathTest, Queues) { + ASSERT_EXIT( + { + RunCase(CheckQueues); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +TEST(SpdkBackpressureDeathTest, Errors) { + ASSERT_EXIT( + { + RunCase(CheckErrors); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +TEST(SpdkBackpressureDeathTest, LargeBursts) { + ASSERT_EXIT( + { + RunCase(CheckLargeBursts); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +TEST(SpdkBackpressureDeathTest, LateSubmissionFifo) { + ASSERT_EXIT( + { + RunCase(CheckLateSubmissionFifo); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +TEST(SpdkBackpressureDeathTest, CompletionOwnership) { + ASSERT_EXIT( + { + RunCase(CheckCompletionOwnership); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +TEST(SpdkBackpressureDeathTest, RetryErrorReentry) { + ASSERT_EXIT( + { + RunCase(CheckRetryErrorReentry); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +TEST(SpdkBackpressureDeathTest, ChannelReopen) { + ASSERT_EXIT( + { + RunCase(CheckChannelReopen); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +TEST(SpdkBackpressureDeathTest, Shutdown) { + ASSERT_EXIT( + { + RunCase(CheckShutdown); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +TEST(SpdkBackpressureDeathTest, SyntheticCompletions) { + ASSERT_EXIT( + { + RunCase(CheckSyntheticCompletions); + std::exit(0); + }, + testing::ExitedWithCode(0), ""); +} + +} // namespace diff --git a/tests/task_transfer_check.cpp b/tests/task_transfer_test.cpp similarity index 59% rename from tests/task_transfer_check.cpp rename to tests/task_transfer_test.cpp index 2ea021e..45d3d4a 100644 --- a/tests/task_transfer_check.cpp +++ b/tests/task_transfer_test.cpp @@ -11,14 +11,16 @@ * limitations under the License. */ +#include #include -#include #include #include +#include #include #include #include +#include #include "bycorf/runtime/task.h" @@ -70,39 +72,62 @@ Task CancelledParent() { Lifetime lifetime; co_return *co_await Delayed(7); } -bool Run(Task task, std::uint64_t expected) { +bool RunTask(Task task, std::uint64_t expected) { auto handle = std::move(task).ReleaseHandle(); handle.resume(); while (!handle.done()) { - if (!pending) return false; + if (!pending) { + handle.destroy(); + return false; + } std::exchange(pending, {}).resume(); } const bool okay = handle.promise().value_ == expected; handle.destroy(); return okay && live == 0 && !pending; } -int Check() { - if (!Run(Sequential(100000), 4999950000ULL)) return 1; - std::cout << "PASS 100000 sequential awaits\n"; - if (!Run(Nested(100000), 100001)) return 1; - std::cout << "PASS 100000 nested awaits\n"; - bycorf::MutableCurrentTaskClass() = bycorf::TaskClass::kBackground; - if (!Run(Mixed(), 20ULL * 49995000 + 190)) return 1; - bycorf::MutableCurrentTaskClass() = bycorf::TaskClass::kForeground; - if (background_registrations == 0) return 1; - std::cout - << "PASS real suspensions, move-only results and background hooks\n"; - { - auto task = CancelledParent(); - auto handle = std::move(task).ReleaseHandle(); - handle.resume(); - if (handle.done() || !pending || live != 2) return 1; - // The event owner cancels its outstanding wake before reclaiming the root. +class TaskTransfer : public testing::Test { + protected: + void SetUp() override { pending = {}; - handle.destroy(); - if (live != 0) return 1; + live = background_registrations = forgotten = 0; + bycorf::MutableCurrentTaskClass() = bycorf::TaskClass::kForeground; + } + void TearDown() override { + EXPECT_EQ(live, 0u); + EXPECT_FALSE(pending); + bycorf::MutableCurrentTaskClass() = bycorf::TaskClass::kForeground; } - std::cout << "PASS suspended child ownership on root destruction\n"; +}; + +TEST_F(TaskTransfer, SequentialAwaitsBoundNativeStack) { + EXPECT_TRUE(RunTask(Sequential(100000), 4999950000ULL)); +} + +TEST_F(TaskTransfer, NestedAwaitsBoundNativeStack) { + EXPECT_TRUE(RunTask(Nested(100000), 100001)); +} + +TEST_F(TaskTransfer, SuspensionsMoveOnlyResultsAndBackgroundHooks) { + bycorf::MutableCurrentTaskClass() = bycorf::TaskClass::kBackground; + EXPECT_TRUE(RunTask(Mixed(), 20ULL * 49995000 + 190)); + EXPECT_GT(background_registrations, 0u); +} + +TEST_F(TaskTransfer, RootDestructionOwnsSuspendedChild) { + auto task = CancelledParent(); + auto handle = std::move(task).ReleaseHandle(); + handle.resume(); + EXPECT_FALSE(handle.done()); + EXPECT_TRUE(pending); + EXPECT_EQ(live, 2u); + // The event owner cancels its outstanding wake before reclaiming the root. + pending = {}; + handle.destroy(); + EXPECT_EQ(live, 0u); +} + +TEST_F(TaskTransfer, DetachedCompletionDestroysRootOnce) { unsigned completions = 0; auto task = Sequential(100000); task.SetCompletionCallback(&completions, @@ -111,9 +136,13 @@ int Check() { handle.destroy(); }); std::move(task).ReleaseHandle().resume(); - if (completions != 1 || live != 0 || forgotten == 0) return 1; - std::cout << "PASS detached completion destroys its root exactly once\n"; - completions = 0; + EXPECT_EQ(completions, 1u); + EXPECT_EQ(live, 0u); + EXPECT_GT(forgotten, 0u); +} + +TEST_F(TaskTransfer, ReentrantCompletionsRunEveryChild) { + unsigned completions = 0; auto reentrant = Sequential(1000); reentrant.SetCompletionCallback( &completions, [](void* context, auto handle) noexcept { @@ -130,9 +159,8 @@ int Check() { } }); std::move(reentrant).ReleaseHandle().resume(); - if (completions != 3 || live != 0) return 1; - std::cout << "PASS reentrant completions run every child\n"; - return 0; + EXPECT_EQ(completions, 3u); + EXPECT_EQ(live, 0u); } } // namespace @@ -146,24 +174,26 @@ void ForgetTaskScheduling(std::coroutine_handle<>) noexcept { ++forgotten; } } // namespace bycorf int main(int argc, char** argv) { - if (argc == 2) return Check(); - const pid_t child = fork(); - if (child < 0) return 1; - if (child == 0) { - // Set before exec so the runtime's native stack starts with a bounded size. - const rlimit stack{512 * 1024, 512 * 1024}; - const rlimit core{0, 0}; - if (setrlimit(RLIMIT_STACK, &stack) || setrlimit(RLIMIT_CORE, &core)) - _exit(127); - execl(argv[0], argv[0], "--child", nullptr); - _exit(127); - } - int status = 0; - if (waitpid(child, &status, 0) != child) return 1; - if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) { - std::cerr << "FAIL bounded-stack Task checks, wait status=" << status - << '\n'; + constexpr char kChild[] = "--bycorf-bounded-stack"; + const bool child = argc > 1 && std::strcmp(argv[argc - 1], kChild) == 0; + if (child) argv[--argc] = nullptr; + // Preserve filters, shuffle/repeat, and XML output across exec; GoogleTest + // removes its own flags from argv during initialization. + std::vector arguments(argv, argv + argc); + testing::InitGoogleTest(&argc, argv); + if (child || GTEST_FLAG_GET(list_tests)) return RUN_ALL_TESTS(); + + // The limit must precede exec so the native stack mapping starts bounded. + // Discovery only lists cases and must never run this stress setup. + const rlimit stack{512 * 1024, 512 * 1024}; + const rlimit core{0, 0}; + if (setrlimit(RLIMIT_STACK, &stack) || setrlimit(RLIMIT_CORE, &core)) { + std::cerr << "cannot set bounded-stack test limits\n"; return 1; } - return 0; + arguments.push_back(const_cast(kChild)); + arguments.push_back(nullptr); + execv("/proc/self/exe", arguments.data()); + std::cerr << "cannot exec bounded-stack tests\n"; + return 1; } diff --git a/tests/tcp_buffered_read_check.cpp b/tests/tcp_buffered_read_test.cpp similarity index 93% rename from tests/tcp_buffered_read_check.cpp rename to tests/tcp_buffered_read_test.cpp index 157fbd1..a3f7a89 100644 --- a/tests/tcp_buffered_read_check.cpp +++ b/tests/tcp_buffered_read_test.cpp @@ -29,7 +29,7 @@ #include "absl/cleanup/cleanup.h" #include "bycorf/io/storage.h" #include "bycorf/net/tcp_stream.h" -#include "bycorf/runtime/runtime.h" +#include "runtime_test.h" namespace { @@ -251,34 +251,15 @@ absl::Status CheckCloseDoesNotShrink(bycorf::Worker& worker) { "close discards queued buffers without shrinking storage"); } -bycorf::Task RunChecks(bycorf::Worker& worker, bool& passed) { - auto status = co_await CheckReads(worker); - if (status.ok()) status = CheckCloseDoesNotShrink(worker); - passed = status.ok(); - if (!status.ok()) std::cerr << status << '\n'; - worker.RequestStop(); - co_return status; +TEST(TcpBufferedRead, QueuedAndLiveReads) { + bycorf::test::RunOnWorker(CheckReads); } -} // namespace - -int main() { - bycorf::Runtime runtime; - runtime.Start( - 1, - [](unsigned, bycorf::Worker& worker) { - auto status = worker.Init(); - if (!status.ok()) { - std::cerr << status << '\n'; - return 1; - } - bool passed = false; - worker.Spawn(RunChecks(worker, passed)); - worker.Run(); - worker.Shutdown(); - return passed ? 0 : 1; - }, - false); - runtime.WaitUntilStopped(); - return runtime.exit_code(); +TEST(TcpBufferedRead, CloseDoesNotShrink) { + bycorf::test::RunOnWorker( + [](bycorf::Worker& worker) -> bycorf::Task { + co_return CheckCloseDoesNotShrink(worker); + }); } + +} // namespace