ConcurrentFlowManySender Class Reference

Inheritance diagram for ConcurrentFlowManySender:

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

Protected Member Functions
void	reset ()
void	join ()
void	cancellation_test (std::chrono::system_clock::time_point at)
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 FlowManyTest.cpp.

Member Typedef Documentation

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

protected

Definition at line 132 of file FlowManyTest.cpp.

Member Function Documentation

void ConcurrentFlowManySender::cancellation_test ( std::chrono::system_clock::time_point  at )

inlineprotected

Definition at line 146 of file FlowManyTest.cpp.

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

                                                             {
    auto f = mi::MAKE(flow_many_sender)([&](auto out) {
      using Out = decltype(out);

      // boolean cancellation
      struct producer {
        producer(Out out, TNT tnt, bool s)
            : out(std::move(out)), tnt(std::move(tnt)), stop(s) {}
        Out out;
        TNT tnt;
        std::atomic<bool> stop;
      };
      auto p = std::make_shared<producer>(std::move(out), tnt_, false);

      struct Data : mi::receiver<> {
        explicit Data(std::shared_ptr<producer> p) : p(std::move(p)) {}
        std::shared_ptr<producer> p;
      };

      auto up = mi::MAKE(receiver)(
          Data{p},
          [&](auto& data, auto requested) {
            signals_ += 1000000;
            if (requested < 1) {
              return;
            }
            // submit work to happen later
            data.p->tnt | op::submit_at(at_, [p = data.p](auto) {
              // check boolean to select signal
              if (!p->stop) {
                ::mi::set_value(p->out, 42);
                ::mi::set_done(p->out);
              }
            });
          },
          [&](auto& data, auto) noexcept {
            signals_ += 100000;
            data.p->stop.store(true);
            data.p->tnt |
                op::submit([p = data.p](auto) { ::mi::set_done(p->out); });
            ++cancel_;
          },
          [&](auto& data) {
            signals_ += 10000;
            data.p->stop.store(true);
            data.p->tnt |
                op::submit([p = data.p](auto) { ::mi::set_done(p->out); });
            ++cancel_;
          });

      tnt_ | op::submit([p, sup = std::move(up)](auto) mutable {
        // pass reference for cancellation.
        ::mi::set_starting(p->out, std::move(sup));
      });
    });
    f |
        op::submit(mi::MAKE(flow_receiver)(
            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;
              mi::set_value(up, 1);
              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 ConcurrentFlowManySender::join ( )

inlineprotected

Definition at line 141 of file FlowManyTest.cpp.

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

void ConcurrentFlowManySender::reset ( )

inlineprotected

Definition at line 134 of file FlowManyTest.cpp.

References now().

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

Member Data Documentation

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

protected

Definition at line 237 of file FlowManyTest.cpp.

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

protected

Definition at line 236 of file FlowManyTest.cpp.

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

protected

Definition at line 230 of file FlowManyTest.cpp.

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

protected

Definition at line 226 of file FlowManyTest.cpp.

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

protected

Definition at line 234 of file FlowManyTest.cpp.

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

protected

Definition at line 232 of file FlowManyTest.cpp.

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

protected

Definition at line 235 of file FlowManyTest.cpp.

mi::time_source ConcurrentFlowManySender::timecancel_ {}

protected

Definition at line 231 of file FlowManyTest.cpp.

mi::time_source ConcurrentFlowManySender::timeproduce_ {}

protected

Definition at line 227 of file FlowManyTest.cpp.

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

protected

Definition at line 228 of file FlowManyTest.cpp.

---

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

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