ConcurrentFlowSingleSender Class Reference

Inheritance diagram for ConcurrentFlowSingleSender:

Protected Types
using	TNT = mi::invoke_result_t< decltype(make_time), mi::time_source<> &, NT & >

Protected Member Functions
void	reset ()
void	join ()
template<class MakeTokens >
void	cancellation_test (std::chrono::system_clock::time_point at, MakeTokens make_tokens)
Protected Member Functions inherited from testing::Test
	Test ()
virtual void	SetUp ()
virtual void	TearDown ()
	Test ()
virtual void	SetUp ()
virtual void	TearDown ()
	Test ()
virtual void	SetUp ()
virtual void	TearDown ()

Protected Attributes
NT	ntproduce_ {mi::new_thread()}
mi::time_source	timeproduce_ {}
TNT	tnt_ {make_time(timeproduce_, ntproduce_)}
NT	ntcancel_ {mi::new_thread()}
mi::time_source	timecancel_ {}
TNT	tcncl_ {make_time(timecancel_, ntcancel_)}
std::atomic< int >	signals_ {0}
std::atomic< int >	terminal_ {0}
std::atomic< int >	cancel_ {0}
std::chrono::system_clock::time_point	at_ {mi::now(tnt_) + 100ms}

Additional Inherited Members
Public Types inherited from testing::Test
typedef internal::SetUpTestCaseFunc	SetUpTestCaseFunc
typedef internal::TearDownTestCaseFunc	TearDownTestCaseFunc
typedef internal::SetUpTestCaseFunc	SetUpTestCaseFunc
typedef internal::TearDownTestCaseFunc	TearDownTestCaseFunc
typedef internal::SetUpTestCaseFunc	SetUpTestCaseFunc
typedef internal::TearDownTestCaseFunc	TearDownTestCaseFunc
Public Member Functions inherited from testing::Test
virtual	~Test ()
virtual	~Test ()
virtual	~Test ()
Static Public Member Functions inherited from testing::Test
static void	SetUpTestCase ()
static void	TearDownTestCase ()
static bool	HasFatalFailure ()
static bool	HasNonfatalFailure ()
static bool	HasFailure ()
static void	RecordProperty (const std::string &key, const std::string &value)
static void	RecordProperty (const std::string &key, int value)
static void	SetUpTestCase ()
static void	TearDownTestCase ()
static bool	HasFatalFailure ()
static bool	HasNonfatalFailure ()
static bool	HasFailure ()
static void	RecordProperty (const std::string &key, const std::string &value)
static void	RecordProperty (const std::string &key, int value)
static void	SetUpTestCase ()
static void	TearDownTestCase ()
static bool	HasFatalFailure ()
static bool	HasNonfatalFailure ()
static bool	HasFailure ()
static void	RecordProperty (const std::string &key, const std::string &value)
static void	RecordProperty (const std::string &key, int value)

Detailed Description

Definition at line 130 of file FlowTest.cpp.

Member Typedef Documentation

using ConcurrentFlowSingleSender::TNT = mi::invoke_result_t<decltype(make_time), mi::time_source<>&, NT&>

protected

Definition at line 132 of file FlowTest.cpp.

Member Function Documentation

template<class MakeTokens >

void ConcurrentFlowSingleSender::cancellation_test ( std::chrono::system_clock::time_point  at, MakeTokens  make_tokens  )

inlineprotected

Definition at line 147 of file FlowTest.cpp.

References data, f, folly::pushmi::lock_both(), MAKE, folly::gen::move, folly::pushmi::__adl::noexcept(), folly::pushmi::on_done(), folly::pushmi::on_error(), folly::pushmi::on_starting(), folly::pushmi::on_value(), folly::pushmi::set_done, stop(), submit, folly::pushmi::operators::submit_at, tokens, and folly::fibers::yield().

                              {
    auto f = mi::MAKE(flow_single_sender)([&, make_tokens](auto out) {
      // boolean cancellation
      bool stop = false;
      auto set_stop = [](auto& stop) {
        if (!!stop) {
          *stop = true;
        }
      };
      auto tokens = make_tokens(stop, set_stop);

      using Stopper = decltype(tokens.second);
      struct Data : mi::receiver<> {
        explicit Data(Stopper stopper) : stopper(std::move(stopper)) {}
        Stopper stopper;
      };
      auto up = mi::MAKE(receiver)(
          Data{std::move(tokens.second)},
          [&](auto& data, auto) noexcept {
            signals_ += 100000;
            ++cancel_;
            auto both = lock_both(data.stopper);
            (*(both.first))(both.second);
          },
          mi::on_done([&](auto& data) {
            signals_ += 10000;
            ++cancel_;
            auto both = lock_both(data.stopper);
            (*(both.first))(both.second);
          }));

      // make all the signals come from the same thread
      tnt_ |
          op::submit([&,
                      stoppee = std::move(tokens.first),
                      up_not_a_shadow_howtoeven = std::move(up),
                      out = std::move(out)](auto tnt) mutable {
            // pass reference for cancellation.
            ::mi::set_starting(out, std::move(up_not_a_shadow_howtoeven));

            // submit work to happen later
            tnt |
                op::submit_at(
                    at_,
                    [stoppee = std::move(stoppee),
                     out = std::move(out)](auto) mutable {
                      // check boolean to select signal
                      auto both = lock_both(stoppee);
                      if (!!both.first && !*(both.first)) {
                        ::mi::set_value(out, 42);
                        ::mi::set_done(out);
                      } else {
                        // cancellation is not an error
                        ::mi::set_done(out);
                      }
                    });
          });
    });

    f |
        op::submit(
            mi::on_value([&](int) { signals_ += 100; }),
            mi::on_error([&](auto) noexcept {
              signals_ += 1000;
              ++terminal_;
            }),
            mi::on_done([&]() {
              signals_ += 1;
              ++terminal_;
            }),
            // stop producer before it is scheduled to run
            mi::on_starting([&, at](auto up) {
              signals_ += 10;
              tcncl_ | op::submit_at(at, [up = std::move(up)](auto) mutable {
                ::mi::set_done(up);
              });
            }));

    while (terminal_ == 0 || cancel_ == 0) {
      std::this_thread::yield();
    }
  }

void ConcurrentFlowSingleSender::join ( )

inlineprotected

Definition at line 141 of file FlowTest.cpp.

              {
    timeproduce_.join();
    timecancel_.join();
  }

void ConcurrentFlowSingleSender::reset ( )

inlineprotected

Definition at line 134 of file FlowTest.cpp.

References now().

               {
    at_ = mi::now(tnt_) + 100ms;
    signals_ = 0;
    terminal_ = 0;
    cancel_ = 0;
  }

Member Data Documentation

std::chrono::system_clock::time_point ConcurrentFlowSingleSender::at_ {mi::now(tnt_) + 100ms}

protected

Definition at line 243 of file FlowTest.cpp.

std::atomic<int> ConcurrentFlowSingleSender::cancel_ {0}

protected

Definition at line 242 of file FlowTest.cpp.

NT ConcurrentFlowSingleSender::ntcancel_ {mi::new_thread()}

protected

Definition at line 236 of file FlowTest.cpp.

NT ConcurrentFlowSingleSender::ntproduce_ {mi::new_thread()}

protected

Definition at line 232 of file FlowTest.cpp.

std::atomic<int> ConcurrentFlowSingleSender::signals_ {0}

protected

Definition at line 240 of file FlowTest.cpp.

TNT ConcurrentFlowSingleSender::tcncl_ {make_time(timecancel_, ntcancel_)}

protected

Definition at line 238 of file FlowTest.cpp.

std::atomic<int> ConcurrentFlowSingleSender::terminal_ {0}

protected

Definition at line 241 of file FlowTest.cpp.

mi::time_source ConcurrentFlowSingleSender::timecancel_ {}

protected

Definition at line 237 of file FlowTest.cpp.

mi::time_source ConcurrentFlowSingleSender::timeproduce_ {}

protected

Definition at line 233 of file FlowTest.cpp.

TNT ConcurrentFlowSingleSender::tnt_ {make_time(timeproduce_, ntproduce_)}

protected

Definition at line 234 of file FlowTest.cpp.

---

The documentation for this class was generated from the following file:

• proxygen/folly/folly/experimental/pushmi/test/FlowTest.cpp