folly::gen::detail::ClosableMPMCQueue< T > Class Template Reference

#include <Parallel-inl.h>

Public Member Functions
	ClosableMPMCQueue (size_t capacity)
	~ClosableMPMCQueue ()
void	openProducer ()
void	openConsumer ()
void	closeInputProducer ()
void	closeOutputConsumer ()
size_t	producers () const
size_t	consumers () const
template<typename... Args>
bool	writeUnlessFull (Args &&...args) noexcept
template<typename... Args>
bool	writeUnlessClosed (Args &&...args)
bool	readUnlessEmpty (T &out)
bool	readUnlessClosed (T &out)

Private Attributes
MPMCQueue< T >	queue_
std::atomic< size_t >	producers_ {0}
std::atomic< size_t >	consumers_ {0}
folly::EventCount	wakeProducer_
folly::EventCount	wakeConsumer_

Detailed Description

template<typename T>
class folly::gen::detail::ClosableMPMCQueue< T >

Definition at line 33 of file Parallel-inl.h.

Constructor & Destructor Documentation

template<typename T>

folly::gen::detail::ClosableMPMCQueue< T >::ClosableMPMCQueue ( size_t  capacity )

inlineexplicit

Definition at line 41 of file Parallel-inl.h.

: queue_(capacity) {}

template<typename T>

folly::gen::detail::ClosableMPMCQueue< T >::~ClosableMPMCQueue ( )

inline

Definition at line 43 of file Parallel-inl.h.

                       {
    CHECK(!producers());
    CHECK(!consumers());
  }

Member Function Documentation

template<typename T>

void folly::gen::detail::ClosableMPMCQueue< T >::closeInputProducer ( )

inline

Definition at line 55 of file Parallel-inl.h.

Referenced by folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::closeInputProducer(), folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::Executor(), and folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::~Executor().

                            {
    size_t producers = producers_--;
    CHECK(producers);
    if (producers == 1) { // last producer
      wakeConsumer_.notifyAll();
    }
  }

template<typename T>

void folly::gen::detail::ClosableMPMCQueue< T >::closeOutputConsumer ( )

inline

Definition at line 63 of file Parallel-inl.h.

Referenced by folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::closeOutputConsumer(), folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::Executor(), and folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::~Executor().

                             {
    size_t consumers = consumers_--;
    CHECK(consumers);
    if (consumers == 1) { // last consumer
      wakeProducer_.notifyAll();
    }
  }

template<typename T>

size_t folly::gen::detail::ClosableMPMCQueue< T >::consumers ( ) const

inline

Definition at line 75 of file Parallel-inl.h.

Referenced by folly::gen::detail::ClosableMPMCQueue< InputDecayed >::closeOutputConsumer(), folly::gen::detail::ClosableMPMCQueue< InputDecayed >::writeUnlessClosed(), folly::gen::detail::ClosableMPMCQueue< InputDecayed >::~ClosableMPMCQueue(), and folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::~Executor().

                           {
    return consumers_.load(std::memory_order_acquire);
  }

template<typename T>

void folly::gen::detail::ClosableMPMCQueue< T >::openConsumer ( )

inline

Definition at line 51 of file Parallel-inl.h.

Referenced by folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::Executor().

                      {
    ++consumers_;
  }

template<typename T>

void folly::gen::detail::ClosableMPMCQueue< T >::openProducer ( )

inline

Definition at line 48 of file Parallel-inl.h.

Referenced by folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::Executor().

                      {
    ++producers_;
  }

template<typename T>

size_t folly::gen::detail::ClosableMPMCQueue< T >::producers ( ) const

inline

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

Referenced by folly::gen::detail::ClosableMPMCQueue< InputDecayed >::closeInputProducer(), folly::gen::detail::ClosableMPMCQueue< InputDecayed >::readUnlessClosed(), folly::gen::detail::ClosableMPMCQueue< InputDecayed >::~ClosableMPMCQueue(), and folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::~Executor().

                           {
    return producers_.load(std::memory_order_acquire);
  }

template<typename T>

bool folly::gen::detail::ClosableMPMCQueue< T >::readUnlessClosed ( T &  out )

inline

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

Referenced by folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Puller::apply(), folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Puller::foreach(), and folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::readUnlessClosed().

                                {
    while (!queue_.readIfNotEmpty(out)) {
      auto key = wakeConsumer_.prepareWait();
      if (!producers()) {
        // wake producers to fill empty space
        wakeProducer_.notify();
        return false;
      }
      wakeConsumer_.wait(key);
    }
    // wake writers blocked by full queue
    wakeProducer_.notify();
    return true;
  }

template<typename T>

bool folly::gen::detail::ClosableMPMCQueue< T >::readUnlessEmpty ( T &  out )

inline

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

Referenced by folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::readUnlessEmpty().

                               {
    if (queue_.read(out)) {
      // wake producers to fill empty space
      wakeProducer_.notify();
      return true;
    }
    return false;
  }

template<typename T>

template<typename... Args>

bool folly::gen::detail::ClosableMPMCQueue< T >::writeUnlessClosed ( Args &&...  args )

inline

Definition at line 90 of file Parallel-inl.h.

Referenced by folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Pusher< all >::compose(), and folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::writeUnlessClosed().

                                         {
    // write if there's room
    while (!queue_.writeIfNotFull(std::forward<Args>(args)...)) {
      // if write fails, check if there are still consumers listening
      auto key = wakeProducer_.prepareWait();
      if (!consumers()) {
        // no consumers left; bail out
        wakeProducer_.cancelWait();
        return false;
      }
      wakeProducer_.wait(key);
    }
    // wake consumers to pick up new value
    wakeConsumer_.notify();
    return true;
  }

template<typename T>

template<typename... Args>

bool folly::gen::detail::ClosableMPMCQueue< T >::writeUnlessFull ( Args &&...  args )

inlinenoexcept

Definition at line 80 of file Parallel-inl.h.

Referenced by folly::gen::detail::Parallel< Ops >::Generator< Input, Source, InputDecayed, Composed, Output, OutputDecayed >::Executor< all >::writeUnlessFull().

                                                {
    if (queue_.write(std::forward<Args>(args)...)) {
      // wake consumers to pick up new value
      wakeConsumer_.notify();
      return true;
    }
    return false;
  }

Member Data Documentation

template<typename T>

std::atomic<size_t> folly::gen::detail::ClosableMPMCQueue< T >::consumers_ {0}

private

Definition at line 36 of file Parallel-inl.h.

Referenced by folly::gen::detail::ClosableMPMCQueue< InputDecayed >::closeOutputConsumer(), folly::gen::detail::ClosableMPMCQueue< InputDecayed >::consumers(), and folly::gen::detail::ClosableMPMCQueue< InputDecayed >::openConsumer().

template<typename T>

std::atomic<size_t> folly::gen::detail::ClosableMPMCQueue< T >::producers_ {0}

private

Definition at line 35 of file Parallel-inl.h.

Referenced by folly::gen::detail::ClosableMPMCQueue< InputDecayed >::closeInputProducer(), folly::gen::detail::ClosableMPMCQueue< InputDecayed >::openProducer(), and folly::gen::detail::ClosableMPMCQueue< InputDecayed >::producers().

template<typename T>

MPMCQueue<T> folly::gen::detail::ClosableMPMCQueue< T >::queue_

private

Definition at line 34 of file Parallel-inl.h.

template<typename T>

folly::EventCount folly::gen::detail::ClosableMPMCQueue< T >::wakeConsumer_

private

Definition at line 38 of file Parallel-inl.h.

template<typename T>

folly::EventCount folly::gen::detail::ClosableMPMCQueue< T >::wakeProducer_

private

Definition at line 37 of file Parallel-inl.h.

---

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

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