FlowTest.cpp File Reference

#include <type_traits>
#include <chrono>
#include <folly/experimental/pushmi/flow_single_sender.h>
#include <folly/experimental/pushmi/o/submit.h>
#include <folly/experimental/pushmi/entangle.h>
#include <folly/experimental/pushmi/new_thread.h>
#include <folly/experimental/pushmi/time_source.h>
#include <folly/experimental/pushmi/trampoline.h>
#include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>

Go to the source code of this file.

Classes
class	ImmediateFlowSingleSender
class	ConcurrentFlowSingleSender

Macros
#define	MAKE(x)   make_##x

Typedefs
using	NT = decltype(mi::new_thread())

Functions
	TEST_F (ImmediateFlowSingleSender, EarlyCancellation)
	TEST_F (ImmediateFlowSingleSender, LateCancellation)
auto	make_time (mi::time_source<> &t, NT &ex)
	TEST_F (ConcurrentFlowSingleSender, EarlySharedCancellation)
	TEST_F (ConcurrentFlowSingleSender, LateSharedCancellation)
	TEST_F (ConcurrentFlowSingleSender, RacingSharedCancellation)
	TEST_F (ConcurrentFlowSingleSender, EarlyEntangledCancellation)
	TEST_F (ConcurrentFlowSingleSender, LateEntangledCancellation)
	TEST_F (ConcurrentFlowSingleSender, RacingEntangledCancellation)

Macro Definition Documentation

#define MAKE

(

x

)

make_##x

Definition at line 42 of file FlowTest.cpp.

Referenced by ConcurrentFlowSingleSender::cancellation_test(), ImmediateFlowSingleSender::make_consumer(), and ImmediateFlowSingleSender::make_producer().

Typedef Documentation

using NT = decltype(mi::new_thread())

Definition at line 124 of file FlowTest.cpp.

Function Documentation

auto make_time ( mi::time_source<> &  t, NT &  ex  )

inline

Definition at line 126 of file FlowTest.cpp.

References folly::pushmi::time_source< E, TP >::make().

                                                  {
  return t.make(mi::systemNowF{}, [ex]() { return ex; })();
}

TEST_F	(	ImmediateFlowSingleSender	,
		EarlyCancellation
	)

Definition at line 105 of file FlowTest.cpp.

References testing::Eq(), EXPECT_THAT, folly::pushmi::set_done, and submit.

                                                     {
  make_producer() | op::submit(make_consumer([](auto up) {
    // immediately stop producer
    ::mi::set_done(up);
  }));

  EXPECT_THAT(signals_, Eq(10011))
      << "expected that the starting, up.done and out.done signals are each recorded once";
}

TEST_F	(	ImmediateFlowSingleSender	,
		LateCancellation
	)

Definition at line 115 of file FlowTest.cpp.

References testing::Eq(), EXPECT_THAT, and submit.

                                                    {
  make_producer() | op::submit(make_consumer([](auto) {
    // do not stop producer before it is scheduled to run
  }));

  EXPECT_THAT(signals_, Eq(110))
      << "expected that the starting and out.value signals are each recorded once";
}

TEST_F	(	ConcurrentFlowSingleSender	,
		EarlySharedCancellation
	)

Definition at line 246 of file FlowTest.cpp.

References testing::Eq(), EXPECT_THAT, join, folly::pushmi::shared_entangle(), and stop().

                                                            {
  cancellation_test(at_ - 50ms, [](auto& stop, auto& set_stop) {
    return mi::shared_entangle(stop, set_stop);
  });

  join();

  EXPECT_THAT(signals_, Eq(10011));
}

TEST_F	(	ConcurrentFlowSingleSender	,
		LateSharedCancellation
	)

Definition at line 256 of file FlowTest.cpp.

References testing::Eq(), EXPECT_THAT, join, folly::pushmi::shared_entangle(), and stop().

                                                           {
  cancellation_test(at_ + 50ms, [](auto& stop, auto& set_stop) {
    return mi::shared_entangle(stop, set_stop);
  });

  join();

  EXPECT_THAT(signals_, Eq(10111))
      << "expected that the starting, up.done, out.value and out.done signals are each recorded once";
}

TEST_F	(	ConcurrentFlowSingleSender	,
		RacingSharedCancellation
	)

Definition at line 267 of file FlowTest.cpp.

References ASSERT_THAT, testing::Eq(), EXPECT_THAT, join, testing::Lt(), folly::pushmi::shared_entangle(), and stop().

                                                             {
  int total = 0;
  int cancellostrace = 0; // 10111
  int cancelled = 0; // 10011

  for (;;) {
    reset();
    cancellation_test(at_, [](auto& stop, auto& set_stop) {
      return mi::shared_entangle(stop, set_stop);
    });

    // accumulate known signals
    ++total;
    cancellostrace += signals_ == 10111;
    cancelled += signals_ == 10011;

    EXPECT_THAT(total, Eq(cancellostrace + cancelled))
        << signals_ << " <- this set of signals is unrecognized";

    ASSERT_THAT(total, Lt(100))
        // too long, show the signals distribution
        << "total " << total << ", cancel-lost-race " << cancellostrace
        << ", cancelled " << cancelled;

    if (cancellostrace > 4 && cancelled > 4) {
      // yay all known outcomes were observed!
      break;
    }
    // try again
    continue;
  }

  join();
}

TEST_F	(	ConcurrentFlowSingleSender	,
		EarlyEntangledCancellation
	)

Definition at line 302 of file FlowTest.cpp.

References folly::pushmi::entangle(), testing::Eq(), EXPECT_THAT, join, and stop().

                                                               {
  cancellation_test(at_ - 50ms, [](auto& stop, auto& set_stop) {
    return mi::entangle(stop, set_stop);
  });

  join();

  EXPECT_THAT(signals_, Eq(10011));
}

TEST_F	(	ConcurrentFlowSingleSender	,
		LateEntangledCancellation
	)

Definition at line 312 of file FlowTest.cpp.

References folly::pushmi::entangle(), testing::Eq(), EXPECT_THAT, join, and stop().

                                                              {
  cancellation_test(at_ + 50ms, [](auto& stop, auto& set_stop) {
    return mi::entangle(stop, set_stop);
  });

  join();

  EXPECT_THAT(signals_, Eq(10111))
      << "expected that the starting, up.done, out.value and out.done signals are each recorded once";
}

TEST_F	(	ConcurrentFlowSingleSender	,
		RacingEntangledCancellation
	)

Definition at line 323 of file FlowTest.cpp.

References ASSERT_THAT, folly::pushmi::entangle(), testing::Eq(), EXPECT_THAT, join, testing::Lt(), and stop().

                                                                {
  int total = 0;
  int cancellostrace = 0; // 10111
  int cancelled = 0; // 10011

  for (;;) {
    reset();
    cancellation_test(at_, [](auto& stop, auto& set_stop) {
      return mi::entangle(stop, set_stop);
    });

    // accumulate known signals
    ++total;
    cancellostrace += signals_ == 10111;
    cancelled += signals_ == 10011;

    EXPECT_THAT(total, Eq(cancellostrace + cancelled))
        << signals_ << " <- this set of signals is unrecognized";

    ASSERT_THAT(total, Lt(100))
        // too long, show the signals distribution
        << "total " << total << ", cancel-lost-race " << cancellostrace
        << ", cancelled " << cancelled;

    if (cancellostrace > 4 && cancelled > 4) {
      // yay all known outcomes were observed!
      break;
    }
    // try again
    continue;
  }

  join();
}