composition_2.cpp File Reference

#include <algorithm>
#include <cassert>
#include <iostream>
#include <vector>
#include <folly/experimental/pushmi/examples/pool.h>
#include <folly/experimental/pushmi/o/transform.h>
#include <folly/experimental/pushmi/o/via.h>
#include <folly/experimental/pushmi/strand.h>

Go to the source code of this file.

Classes
struct	f_t
struct	g_t

Functions
f_t	f ()
g_t	g (f_t)
template<class CPUExecutor , class IOExecutor >
void	lisp (CPUExecutor cpu, IOExecutor io)
template<class CPUExecutor , class IOExecutor >
void	sugar (CPUExecutor cpu, IOExecutor io)
template<class CPUExecutor , class IOExecutor >
void	pipe (CPUExecutor cpu, IOExecutor io)
int	main ()

Function Documentation

f_t f

(

)

Definition at line 30 of file composition_2.cpp.

Referenced by lisp(), pipe(), and sugar().

        {
  return {};
}

g_t g

(

f_t

)

Definition at line 34 of file composition_2.cpp.

Referenced by folly::detail::NodeRecycler< NodeType, NodeAlloc, typename std::enable_if< !NodeType::template DestroyIsNoOp< NodeAlloc >::value >::type >::add(), folly::TestExecutor::addImpl(), folly::threadlocal_detail::StaticMetaBase::allocate(), folly::SimpleAllocator::allocate(), folly::detail::BufferedStat< DigestT, ClockT >::append(), folly::detail::DigestBuilder< DigestT >::append(), proxygen::HTTPMessage::atomicDumpMessage(), BENCHMARK(), folly::AsyncUDPSocket::bind(), folly::detail::DigestBuilder< DigestT >::build(), folly::ssl::cleanup(), folly::detail::ConcurrentHashMapSegment< KeyType, ValueType, ShardBits, HashFn, KeyEqual, Allocator, Atom, Mutex >::clear(), folly::NotificationQueue< MessageT >::Consumer::consumeUntilDrained(), folly::NotificationQueue< MessageT >::SimpleConsumer::consumeUntilDrained(), folly::SimpleAllocator::deallocate(), folly::threadlocal_detail::StaticMetaBase::destroy(), QueueTest::destroyCallback(), folly::portability::ssl::DH_set0_pqg(), folly::detail::BufferedStat< DigestT, ClockT >::doUpdate(), wangle::ConnectionManager::DrainHelper::drainConnections(), wangle::ConnectionManager::dropAllConnections(), wangle::ConnectionManager::dropConnections(), BaselinePQ< T, PriorityQueue, Mutex >::empty(), testing::gmock_matchers_test::EMString(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::ensureSignal(), folly::detail::ConcurrentHashMapSegment< KeyType, ValueType, ShardBits, HashFn, KeyEqual, Allocator, Atom, Mutex >::erase_internal(), TestAsyncTransport::failPendingWrites(), testing::gmock_matchers_test::FindBacktrackingMaxBPM(), proxygen::HTTPSession::flowControlTimeoutExpired(), folly::detail::BufferedStat< DigestT, ClockT >::flush(), folly::compression::generateRandomList(), folly::detail::BufferedDigest< DigestT, ClockT >::get(), folly::detail::BufferedSlidingWindow< DigestT, ClockT >::get(), wangle::SSLSessionCacheManager::getLocalCache(), wangle::SSLUtil::getRSAExIndex(), folly::SharedPromise< T >::getSemiFuture(), wangle::SSLUtil::getSSLCtxExIndex(), wangle::SSLUtil::getSSLSessionExStrIndex(), HTTPDownstreamTest< SPDY3_1CodecPair >::gracefulShutdown(), folly::ConcurrentSkipList< T, Comp, NodeAlloc, MAX_HEIGHT >::growHeight(), folly::HandshakeCallback::handshakeErr(), folly::HandshakeCallback::handshakeSuc(), folly::ssl::init(), folly::NotificationQueue< MessageT >::Consumer::init(), folly::detail::ConcurrentHashMapSegment< KeyType, ValueType, ShardBits, HashFn, KeyEqual, Allocator, Atom, Mutex >::insert_internal(), proxygen::HTTPSession::invokeOnAllTransactions(), folly::SharedPromise< T >::isFulfilled(), lisp(), wangle::ShardedLocalSSLSessionCache::lookupSession(), folly::ssl::markInitialized(), folly::detail::ConcurrentHashMapSegment< KeyType, ValueType, ShardBits, HashFn, KeyEqual, Allocator, Atom, Mutex >::max_load_factor(), testing::gmock_generated_function_mockers_test::MockFoo::MockFoo(), proxygen::HTTPTransaction::onEgressBodyFirstByte(), proxygen::HTTPTransaction::onEgressBodyLastByte(), proxygen::HTTPTransaction::onEgressHeaderFirstByte(), proxygen::HTTPTransaction::onEgressLastByteAck(), proxygen::HTTPTransaction::onEgressTimeout(), proxygen::HTTPTransaction::onEgressTrackedByte(), proxygen::HTTPTransaction::onError(), proxygen::HTTPTransaction::onExTransaction(), proxygen::HTTPSession::onGoaway(), proxygen::HTTPTransaction::onGoaway(), proxygen::HTTP1xCodec::onHeadersComplete(), proxygen::HTTPTransaction::onIngressBody(), proxygen::HTTPTransaction::onIngressHeadersComplete(), proxygen::HTTPTransaction::onIngressTimeout(), proxygen::HTTPTransaction::onIngressWindowUpdate(), proxygen::HTTPTransaction::onPushedTransaction(), proxygen::HTTPSession::onSettings(), folly::threadlocal_detail::StaticMetaBase::onThreadExit(), proxygen::HTTPTransaction::onWriteReady(), proxygen::HTTP2Codec::parseHeadersDecodeFrames(), proxygen::HTTPTransaction::pauseEgress(), proxygen::HTTPTransaction::pauseIngress(), testing::gmock_generated_actions_test::SubstractAction::Perform(), pipe(), GlobalLockPQ< T >::pop(), proxygen::HTTPTransaction::processIngressBody(), proxygen::HTTPTransaction::processIngressChunkComplete(), proxygen::HTTPTransaction::processIngressChunkHeader(), proxygen::HTTPTransaction::processIngressEOM(), proxygen::HTTPTransaction::processIngressHeadersComplete(), proxygen::HTTPTransaction::processIngressTrailers(), proxygen::HTTPTransaction::processIngressUpgrade(), GlobalLockPQ< T >::push(), folly::threadlocal_detail::StaticMetaBase::pushBackLocked(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessageImpl(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessagesImpl(), folly::detail::NodeRecycler< NodeType, NodeAlloc, typename std::enable_if< !NodeType::template DestroyIsNoOp< NodeAlloc >::value >::type >::releaseRef(), wangle::ShardedLocalSSLSessionCache::removeSession(), folly::threadlocal_detail::StaticMetaBase::reserve(), proxygen::HTTPTransaction::resumeEgress(), proxygen::HTTPTransaction::resumeIngress(), proxygen::HTTPSession::resumeTransactions(), folly::rcu_domain< Tag >::retire(), runFairness(), runTest(), proxygen::HTTPTransaction::sendAbort(), proxygen::HTTPTransaction::sendEOM(), folly::SharedPromise< T >::setInterruptHandler(), folly::ssl::setLockTypes(), folly::ssl::setLockTypesAndInit(), folly::SharedPromise< T >::setTry(), shared_ptr_test(), wangle::SSLSessionCacheManager::shutdown(), folly::FunctionScheduler::shutdown(), TestAsyncTransport::shutdownWriteNow(), BaselinePQ< T, PriorityQueue, Mutex >::size(), folly::SharedPromise< T >::size(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::size(), testing::internal::MatchMatrix::SpaceIndex(), folly::NotificationQueue< MessageT >::Consumer::stopConsuming(), wangle::ShardedLocalSSLSessionCache::storeSession(), sugar(), folly::rcu_domain< Tag >::synchronize(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::syncSignalAndQueue(), TEST(), testing::gmock_generated_actions_test::TEST(), TEST_F(), testing::gmock_matchers_test::TEST_F(), HTTPDownstreamTest< C >::testChunks(), folly::EventBaseManager::trackEventBase(), BaselinePQ< T, PriorityQueue, Mutex >::try_peek(), BaselinePQ< T, PriorityQueue, Mutex >::try_pop(), BaselinePQ< T, PriorityQueue, Mutex >::try_push(), testing::internal::MaxBipartiteMatchState::TryAugment(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::tryConsume(), tryretire(), unique_ptr_test(), folly::EventBaseManager::untrackEventBase(), folly::detail::BufferedStat< DigestT, ClockT >::updateIfExpired(), proxygen::HTTPTransaction::validateIngressStateTransition(), folly::EventBaseManager::withEventBaseSet(), proxygen::HTTPSession::writeTimeoutExpired(), testing::gmock_generated_function_mockers_test::FooInterface::~FooInterface(), wangle::LocalSSLSessionCache::~LocalSSLSessionCache(), and folly::SimpleAllocator::~SimpleAllocator().

           {
  return {};
}

template<class CPUExecutor , class IOExecutor >

void lisp	(	CPUExecutor	cpu,
		IOExecutor	io
	)

Definition at line 40 of file composition_2.cpp.

References f(), g(), folly::pushmi::strands(), submit, folly::pushmi::operators::transform, and folly::pushmi::operators::via.

Referenced by main().

                                          {
  // f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
  // the functions)
  op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })(
      op::transform([](auto) { return f(); })(cpu)));

  // f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu
  // task runs g)
  op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })(
      op::via(mi::strands(cpu))(op::transform([](auto) { return f(); })(cpu))));

  // f on io  - g on cpu
  op::submit([](g_t) {})(op::transform([](f_t ft) { return g(ft); })(
      op::via(mi::strands(cpu))(op::transform([](auto) { return f(); })(io))));
}

int main

(

void

)

Definition at line 101 of file composition_2.cpp.

References lisp(), max, folly::pushmi::pipe, and sugar().

           {
  mi::pool cpuPool{std::max(1u, std::thread::hardware_concurrency())};
  mi::pool ioPool{std::max(1u, std::thread::hardware_concurrency())};

  lisp(cpuPool.executor(), ioPool.executor());
  sugar(cpuPool.executor(), ioPool.executor());
  pipe(cpuPool.executor(), ioPool.executor());

  ioPool.wait();
  cpuPool.wait();

  std::cout << "OK" << std::endl;
}

template<class CPUExecutor , class IOExecutor >

void pipe	(	CPUExecutor	cpu,
		IOExecutor	io
	)

Definition at line 74 of file composition_2.cpp.

References f(), g(), folly::pushmi::strands(), submit, folly::pushmi::operators::transform, and folly::pushmi::operators::via.

Referenced by BENCHMARK(), folly::Subprocess::findByChildFd(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::NotificationQueue(), folly::Subprocess::spawn(), folly::Subprocess::spawnInternal(), TEST(), and folly::TEST().

                                          {
  // f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
  // the functions)
  mi::pipe(
      cpu,
      op::transform([](auto) { return f(); }),
      op::transform([](f_t ft) { return g(ft); }),
      op::submit([](g_t) {}));

  // f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu
  // task runs g)
  mi::pipe(
      cpu,
      op::transform([](auto) { return f(); }),
      op::via(mi::strands(cpu)),
      op::transform([](f_t ft) { return g(ft); }),
      op::submit([](g_t) {}));

  // f on io  - g on cpu
  mi::pipe(
      io,
      op::transform([](auto) { return f(); }),
      op::via(mi::strands(cpu)),
      op::transform([](f_t ft) { return g(ft); }),
      op::submit([](g_t) {}));
}

template<class CPUExecutor , class IOExecutor >

void sugar	(	CPUExecutor	cpu,
		IOExecutor	io
	)

Definition at line 57 of file composition_2.cpp.

References f(), g(), folly::pushmi::strands(), submit, folly::pushmi::operators::transform, and folly::pushmi::operators::via.

Referenced by main().

                                           {
  // f on cpu - g on cpu (implicit: a single task on the cpu executor runs all
  // the functions)
  cpu | op::transform([](auto) { return f(); }) |
      op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {});

  // f on cpu - g on cpu (explicit: the first cpu task runs f and a second cpu
  // task runs g)
  cpu | op::transform([](auto) { return f(); }) | op::via(mi::strands(cpu)) |
      op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {});

  // f on io  - g on cpu
  io | op::transform([](auto) { return f(); }) | op::via(mi::strands(cpu)) |
      op::transform([](f_t ft) { return g(ft); }) | op::submit([](g_t) {});
}