Restoring TestThread code with added std::lock_guard<std::mutex>.

Author: rwgk

tests/test_iostream.cpp

@@ -13,8 +13,11 @@
 
 #include <pybind11/iostream.h>
 #include "pybind11_tests.h"
+#include <atomic>
 #include <iostream>
+#include <mutex>
 #include <string>
+#include <thread>
 
 
 void noisy_function(std::string msg, bool flush) {
@@ -29,6 +32,44 @@ void noisy_funct_dual(std::string msg, std::string emsg) {
     std::cerr << emsg;
 }
 
+// object to manage C++ thread
+// simply repeatedly write to std::cerr until stopped
+// redirect is called at some point to test the safety of scoped_estream_redirect
+struct TestThread {
+    TestThread() : t_{nullptr}, stop_{false} {
+        auto thread_f = [this] {
+            static std::mutex cout_mutex;
+            while (!stop_) {
+                {
+                    const std::lock_guard<std::mutex> lock(cout_mutex);
+                    std::cout << "x" << std::flush;
+                }
+                std::this_thread::sleep_for(std::chrono::microseconds(50));
+            } };
+        t_ = new std::thread(std::move(thread_f));
+    }
+
+    ~TestThread() {
+        delete t_;
+    }
+
+    void stop() { stop_ = true; }
+
+    void join() {
+        py::gil_scoped_release gil_lock;
+        t_->join();
+    }
+
+    void sleep() {
+        py::gil_scoped_release gil_lock;
+        std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    }
+
+    std::thread * t_;
+    std::atomic<bool> stop_;
+};
+
+
 TEST_SUBMODULE(iostream, m) {
 
     add_ostream_redirect(m);
@@ -74,4 +115,10 @@ TEST_SUBMODULE(iostream, m) {
         std::cout << msg << std::flush;
         std::cerr << emsg << std::flush;
     });
+
+    py::class_<TestThread>(m, "TestThread")
+        .def(py::init<>())
+        .def("stop", &TestThread::stop)
+        .def("join", &TestThread::join)
+        .def("sleep", &TestThread::sleep);
 }

tests/test_iostream.py

@@ -306,3 +306,26 @@ def test_redirect_both(capfd):
     assert stderr == ""
     assert stream.getvalue() == msg
     assert stream2.getvalue() == msg2
+
+
+def test_threading():
+    with m.ostream_redirect(stdout=True, stderr=False):
+        # start some threads
+        threads = []
+
+        # start some threads
+        for _j in range(20):
+            threads.append(m.TestThread())
+
+        # give the threads some time to fail
+        threads[0].sleep()
+
+        # stop all the threads
+        for t in threads:
+            t.stop()
+
+        for t in threads:
+            t.join()
+
+        # if a thread segfaults, we don't get here
+        assert True