Function test_scoped_critical_section

tests/test_scoped_critical_section.cpp:48–100

// thread-safe manner.

void test_scoped_critical_section(const py::handle &cls) {
    auto barrier = std::barrier(2);
    auto bool_wrapper = cls(false);
    bool output = false;

    {
        // Release the GIL to allow run threads in parallel.
        py::gil_scoped_release gil_release{};

        std::thread t1([&]() {
            // Use gil_scoped_acquire to ensure we have a valid Python thread state
            // before entering the critical section. Otherwise, the critical section
            // will cause a segmentation fault.
            py::gil_scoped_acquire ensure_tstate{};
            // Enter the critical section with the same object as the second thread.
            py::scoped_critical_section lock{bool_wrapper};
            // At this point, the object is locked by this thread via the scoped_critical_section.
            // This barrier will ensure that the second thread waits until this thread has released
            // the critical section before proceeding.
            barrier.arrive_and_wait();
            // Sleep for a short time to simulate some work in the critical section.
            // This sleep is necessary to test the locking mechanism properly.
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            auto *bw = bool_wrapper.cast<BoolWrapper *>();
            bw->set(true);
        });

        std::thread t2([&]() {
            // This thread will wait until the first thread has entered the critical section due to
            // the barrier.
            barrier.arrive_and_wait();
            {
                // Use gil_scoped_acquire to ensure we have a valid Python thread state
                // before entering the critical section. Otherwise, the critical section
                // will cause a segmentation fault.
                py::gil_scoped_acquire ensure_tstate{};
                // Enter the critical section with the same object as the first thread.
                py::scoped_critical_section lock{bool_wrapper};
                // At this point, the critical section is released by the first thread, the value
                // is set to true.
                auto *bw = bool_wrapper.cast<BoolWrapper *>();
                output = bw->get();
            }
        });

        t1.join();
        t2.join();
    }

    if (!output) {
        throw std::runtime_error("Scoped critical section test failed: output is false");
    }
}

void test_scoped_critical_section2(const py::handle &cls) {
    auto barrier = std::barrier(3);