test_iostream.cpp
Go to the documentation of this file.
1 /*
2  tests/test_iostream.cpp -- Usage of scoped_output_redirect
3 
4  Copyright (c) 2017 Henry F. Schreiner
5 
6  All rights reserved. Use of this source code is governed by a
7  BSD-style license that can be found in the LICENSE file.
8 */
9 
10 #include <pybind11/iostream.h>
11 
12 #include "pybind11_tests.h"
13 
14 #include <atomic>
15 #include <iostream>
16 #include <mutex>
17 #include <string>
18 #include <thread>
19 
20 void noisy_function(const std::string &msg, bool flush) {
21 
22  std::cout << msg;
23  if (flush) {
24  std::cout << std::flush;
25  }
26 }
27 
28 void noisy_funct_dual(const std::string &msg, const std::string &emsg) {
29  std::cout << msg;
30  std::cerr << emsg;
31 }
32 
33 // object to manage C++ thread
34 // simply repeatedly write to std::cerr until stopped
35 // redirect is called at some point to test the safety of scoped_estream_redirect
36 struct TestThread {
37  TestThread() : stop_{false} {
38  auto thread_f = [this] {
39  static std::mutex cout_mutex;
40  while (!stop_) {
41  {
42  // #HelpAppreciated: Work on iostream.h thread safety.
43  // Without this lock, the clang ThreadSanitizer (tsan) reliably reports a
44  // data race, and this test is predictably flakey on Windows.
45  // For more background see the discussion under
46  // https://github.com/pybind/pybind11/pull/2982 and
47  // https://github.com/pybind/pybind11/pull/2995.
48  const std::lock_guard<std::mutex> lock(cout_mutex);
49  std::cout << "x" << std::flush;
50  }
51  std::this_thread::sleep_for(std::chrono::microseconds(50));
52  }
53  };
54  t_ = new std::thread(std::move(thread_f));
55  }
56 
57  ~TestThread() { delete t_; }
58 
59  void stop() { stop_ = true; }
60 
61  void join() const {
62  py::gil_scoped_release gil_lock;
63  t_->join();
64  }
65 
66  void sleep() {
67  py::gil_scoped_release gil_lock;
68  std::this_thread::sleep_for(std::chrono::milliseconds(50));
69  }
70 
71  std::thread *t_{nullptr};
72  std::atomic<bool> stop_;
73 };
74 
75 TEST_SUBMODULE(iostream, m) {
76 
78 
79  // test_evals
80 
81  m.def("captured_output_default", [](const std::string &msg) {
82  py::scoped_ostream_redirect redir;
83  std::cout << msg << std::flush;
84  });
85 
86  m.def("captured_output", [](const std::string &msg) {
87  py::scoped_ostream_redirect redir(std::cout, py::module_::import("sys").attr("stdout"));
88  std::cout << msg << std::flush;
89  });
90 
91  m.def("guard_output",
93  py::call_guard<py::scoped_ostream_redirect>(),
94  py::arg("msg"),
95  py::arg("flush") = true);
96 
97  m.def("captured_err", [](const std::string &msg) {
98  py::scoped_ostream_redirect redir(std::cerr, py::module_::import("sys").attr("stderr"));
99  std::cerr << msg << std::flush;
100  });
101 
102  m.def("noisy_function", &noisy_function, py::arg("msg"), py::arg("flush") = true);
103 
104  m.def("dual_guard",
106  py::call_guard<py::scoped_ostream_redirect, py::scoped_estream_redirect>(),
107  py::arg("msg"),
108  py::arg("emsg"));
109 
110  m.def("raw_output", [](const std::string &msg) { std::cout << msg << std::flush; });
111 
112  m.def("raw_err", [](const std::string &msg) { std::cerr << msg << std::flush; });
113 
114  m.def("captured_dual", [](const std::string &msg, const std::string &emsg) {
115  py::scoped_ostream_redirect redirout(std::cout, py::module_::import("sys").attr("stdout"));
116  py::scoped_ostream_redirect redirerr(std::cerr, py::module_::import("sys").attr("stderr"));
117  std::cout << msg << std::flush;
118  std::cerr << emsg << std::flush;
119  });
120 
121  py::class_<TestThread>(m, "TestThread")
122  .def(py::init<>())
123  .def("stop", &TestThread::stop)
124  .def("join", &TestThread::join)
125  .def("sleep", &TestThread::sleep);
126 }
void noisy_funct_dual(const std::string &msg, const std::string &emsg)
Matrix3f m
std::atomic< bool > stop_
class_< detail::OstreamRedirect > add_ostream_redirect(module_ m, const std::string &name="ostream_redirect")
Definition: iostream.h:258
std::thread * t_
EIGEN_DEVICE_FUNC EIGEN_STRONG_INLINE const ArgReturnType arg() const
void join() const
TEST_SUBMODULE(iostream, m)
void noisy_function(const std::string &msg, bool flush)


gtsam
Author(s):
autogenerated on Tue Jul 4 2023 02:37:46