folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline Class Reference

Public Member Functions
	ExecutionPipeline (const Predicate &pred, size_t nThreads)
	~ExecutionPipeline ()
void	stop ()
bool	write (Value &&value)
void	blockingWrite (Value &&value)
bool	read (Output &out)
void	blockingRead (Output &out)

Private Member Functions
void	predApplier ()

Private Attributes
std::vector< std::thread >	workers_
std::atomic< bool >	done_ {false}
const Predicate &	pred_
MPMCPipeline< Input, Output >	pipeline_
EventCount	wake_

Detailed Description

template<class Predicate>
template<class Value, class Source, class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>
class folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline

Definition at line 69 of file ParallelMap-inl.h.

Constructor & Destructor Documentation

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::ExecutionPipeline ( const Predicate &  pred, size_t  nThreads  )

inline

Definition at line 77 of file ParallelMap-inl.h.

References i, and folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::predApplier().

          : pred_(pred), pipeline_(nThreads, nThreads) {
        workers_.reserve(nThreads);
        for (size_t i = 0; i < nThreads; i++) {
          workers_.push_back(std::thread([this] { this->predApplier(); }));
        }
      }

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::~ExecutionPipeline ( )

inline

Definition at line 85 of file ParallelMap-inl.h.

References folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::done_, and folly::MPMCPipeline< In, Stages >::sizeGuess().

                           {
        assert(pipeline_.sizeGuess() == 0);
        assert(done_.load());
        for (auto& w : workers_) {
          w.join();
        }
      }

Member Function Documentation

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

void folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::blockingRead ( Output &  out )

inline

Definition at line 116 of file ParallelMap-inl.h.

References folly::MPMCPipeline< In, Stages >::blockingRead().

Referenced by folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::apply(), and folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::foreach().

                                     {
        pipeline_.blockingRead(out);
      }

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

void folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::blockingWrite ( Value &&  value )

inline

Definition at line 107 of file ParallelMap-inl.h.

References folly::MPMCPipeline< In, Stages >::blockingWrite(), folly::EventCount::notify(), and folly::value().

Referenced by folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::apply(), and folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::foreach().

                                        {
        pipeline_.blockingWrite(std::forward<Value>(value));
        wake_.notify();
      }

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

void folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::predApplier ( )

inlineprivate

Definition at line 121 of file ParallelMap-inl.h.

References folly::EventCount::cancelWait(), folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::done_, folly::gen::move, folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::pred_, folly::EventCount::prepareWait(), ticket, and folly::EventCount::wait().

Referenced by folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::ExecutionPipeline().

                         {
        // Each thread takes a value from the pipeline_, runs the
        // predicate and enqueues the result. The pipeline preserves
        // ordering. NOTE: don't use blockingReadStage<0> to read from
        // the pipeline_ as there may not be any: end-of-data is signaled
        // separately using done_/wake_.
        Input in;
        for (;;) {
          auto key = wake_.prepareWait();

          typename MPMCPipeline<Input, Output>::template Ticket<0> ticket;
          if (pipeline_.template readStage<0>(ticket, in)) {
            wake_.cancelWait();
            Output out = pred_(std::move(in));
            pipeline_.template blockingWriteStage<0>(ticket, std::move(out));
            continue;
          }

          if (done_.load(std::memory_order_acquire)) {
            wake_.cancelWait();
            break;
          }

          // Not done_, but no items in the queue.
          wake_.wait(key);
        }
      }

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

bool folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::read ( Output &  out )

inline

Definition at line 112 of file ParallelMap-inl.h.

References folly::MPMCPipeline< In, Stages >::read().

Referenced by folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::apply(), and folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::foreach().

                             {
        return pipeline_.read(out);
      }

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

void folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::stop ( )

inline

Definition at line 93 of file ParallelMap-inl.h.

References folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::done_, and folly::EventCount::notifyAll().

Referenced by folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::apply(), and folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::foreach().

                  {
        // prevent workers from consuming more than we produce.
        done_.store(true, std::memory_order_release);
        wake_.notifyAll();
      }

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

bool folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::write ( Value &&  value )

inline

Definition at line 99 of file ParallelMap-inl.h.

References folly::EventCount::notify(), folly::value(), and folly::MPMCPipeline< In, Stages >::write().

Referenced by folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::apply(), and folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::foreach().

                                {
        bool wrote = pipeline_.write(std::forward<Value>(value));
        if (wrote) {
          wake_.notify();
        }
        return wrote;
      }

Member Data Documentation

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

std::atomic<bool> folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::done_ {false}

private

Definition at line 71 of file ParallelMap-inl.h.

Referenced by folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::predApplier(), folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::stop(), and folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::~ExecutionPipeline().

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

MPMCPipeline<Input, Output> folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::pipeline_

private

Definition at line 73 of file ParallelMap-inl.h.

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

const Predicate& folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::pred_

private

Definition at line 72 of file ParallelMap-inl.h.

Referenced by folly::gen::detail::PMap< Predicate >::compose(), and folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::predApplier().

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

EventCount folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::wake_

private

Definition at line 74 of file ParallelMap-inl.h.

template<class Predicate >

template<class Value , class Source , class Input = typename std::decay<Value>::type, class Output = typename std::decay<invoke_result_t<Predicate, Value>>::type>

std::vector<std::thread> folly::gen::detail::PMap< Predicate >::Generator< Value, Source, Input, Output >::ExecutionPipeline::workers_

private

Definition at line 70 of file ParallelMap-inl.h.

---

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

• proxygen/folly/folly/gen/ParallelMap-inl.h