folly::pushmi::time_source_queue< E, TP, NF, Executor > Class Template Reference

#include <time_source.h>

Inheritance diagram for folly::pushmi::time_source_queue< E, TP, NF, Executor >:

Public Types
using	time_point = std::decay_t< TP >
Public Types inherited from folly::pushmi::time_source_queue_base< E, TP >
using	time_point = std::decay_t< TP >

Public Member Functions
	~time_source_queue ()
	time_source_queue (std::weak_ptr< time_source_shared< E, time_point >> source, NF nf, Executor ex)
void	dispatch () override
auto	shared_from_that ()
template<class Exec >
void	value (Exec &&)
template<class AE >
void	error (AE e) noexcept
void	done ()
Public Member Functions inherited from folly::pushmi::time_source_queue_base< E, TP >
virtual	~time_source_queue_base ()
time_heap_item< E, TP > &	top ()

Public Attributes
std::weak_ptr< time_source_shared< E, time_point > >	source_
NF	nf_
Executor	ex_
Public Attributes inherited from folly::pushmi::time_source_queue_base< E, TP >
bool	dispatching_ = false
bool	pending_ = false
std::priority_queue< time_heap_item< E, TP >, std::vector< time_heap_item< E, TP > >, std::greater<> >	heap_

Detailed Description

template<class E, class TP, class NF, class Executor>
class folly::pushmi::time_source_queue< E, TP, NF, Executor >

Definition at line 106 of file time_source.h.

Member Typedef Documentation

template<class E , class TP , class NF , class Executor >

using folly::pushmi::time_source_queue< E, TP, NF, Executor >::time_point = std::decay_t<TP>

Definition at line 108 of file time_source.h.

Constructor & Destructor Documentation

template<class E , class TP , class NF , class Executor >

folly::pushmi::time_source_queue< E, TP, NF, Executor >::~time_source_queue ( )

inline

Definition at line 109 of file time_source.h.

{}

template<class E , class TP , class NF , class Executor >

folly::pushmi::time_source_queue< E, TP, NF, Executor >::time_source_queue ( std::weak_ptr< time_source_shared< E, time_point >>  source, NF  nf, Executor  ex  )

inline

Definition at line 110 of file time_source.h.

      : source_(std::move(source)), nf_(std::move(nf)), ex_(std::move(ex)) {}

Member Function Documentation

template<class E , class TP , class NF , class Executor >

void folly::pushmi::time_source_queue< E, TP, NF, Executor >::dispatch ( )

overridevirtual

Implements folly::pushmi::time_source_queue_base< E, TP >.

Definition at line 238 of file time_source.h.

References folly::pushmi::submit.

                                                      {
  submit(
      ex_, time_source_queue_receiver<E, TP, NF, Executor>{shared_from_that()});
}

template<class E , class TP , class NF , class Executor >

void folly::pushmi::time_source_queue< E, TP, NF, Executor >::done ( )

inline

Definition at line 205 of file time_source.h.

References folly::gen::guard(), and s.

              {
    auto s = source_.lock();
    std::unique_lock<std::mutex> guard{s->lock_};

    if (!this->dispatching_ || this->pending_) {
      std::abort();
    }
    this->dispatching_ = false;

    // add back to pending_ to get the remaining items dispatched
    s->pending_.push_back(this->shared_from_this());
    this->pending_ = true;
    if (this->heap_.top().when <= s->earliest_) {
      // this is the earliest, tell worker to reset earliest_
      ++s->dirty_;
      s->wake_.notify_one();
    }
  }

template<class E , class TP , class NF , class Executor >

template<class AE >

void folly::pushmi::time_source_queue< E, TP, NF, Executor >::error ( AE  e )

inlinenoexcept

Definition at line 187 of file time_source.h.

References folly::pushmi::detail::as_const, folly::gen::guard(), folly::gen::move, s, folly::pushmi::set_error, folly::pushmi::top, and folly::pushmi::time_heap_item< E, TP >::what.

                            {
    auto s = source_.lock();
    std::unique_lock<std::mutex> guard{s->lock_};

    if (!this->dispatching_ || this->pending_) {
      std::abort();
    }

    while (!this->heap_.empty()) {
      auto what{std::move(this->top().what)};
      this->heap_.pop();
      --s->items_;
      guard.unlock();
      set_error(what, detail::as_const(e));
      guard.lock();
    }
    this->dispatching_ = false;
  }

template<class E , class TP , class NF , class Executor >

auto folly::pushmi::time_source_queue< E, TP, NF, Executor >::shared_from_that ( )

inline

Definition at line 121 of file time_source.h.

                          {
    return std::static_pointer_cast<time_source_queue<E, TP, NF, Executor>>(
        this->shared_from_this());
  }

template<class E , class TP , class NF , class Executor >

template<class Exec >

void folly::pushmi::time_source_queue< E, TP, NF, Executor >::value ( Exec &&  )

inline

Definition at line 127 of file time_source.h.

References folly::gen::guard(), folly::gen::move, s, folly::pushmi::set_done, folly::pushmi::set_error, folly::pushmi::set_value, start, folly::pushmi::top, and folly::pushmi::time_heap_item< E, TP >::what.

                     {
    auto s = source_.lock();

    if (s->t_.get_id() == std::this_thread::get_id()) {
      // Executor is not allowed to use the time_source thread
      std::abort();
    }

    //
    // pull ready items from the heap in order.

    // drain anything queued within the next 50ms before
    // going back to the pending queue.
    auto start = nf_() + std::chrono::milliseconds(50);

    std::unique_lock<std::mutex> guard{s->lock_};

    if (!this->dispatching_ || this->pending_) {
      std::abort();
    }

    if (this->heap_.empty()) {
      return;
    }
    auto that = shared_from_that();
    auto subEx = time_source_executor<E, TP, NF, Executor>{s, that};
    while (!this->heap_.empty() && this->heap_.top().when <= start) {
      auto item{std::move(this->top())};
      this->heap_.pop();
      guard.unlock();
      std::this_thread::sleep_until(item.when);
      set_value(item.what, any_time_executor_ref<E, TP>{subEx});
      set_done(item.what);
      guard.lock();
      // allows set_value to queue nested items
      --s->items_;
    }

    if (this->heap_.empty()) {
      // if this is empty, tell worker to check for the done condition.
      ++s->dirty_;
      s->wake_.notify_one();
    } else {
      if (!!s->error_) {
        while (!this->heap_.empty()) {
          try {
            auto what{std::move(this->top().what)};
            this->heap_.pop();
            --s->items_;
            guard.unlock();
            set_error(what, *s->error_);
            guard.lock();
          } catch (...) {
            // we already have an error, ignore this one.
          }
        }
      }
    }
  }

Member Data Documentation

template<class E , class TP , class NF , class Executor >

Executor folly::pushmi::time_source_queue< E, TP, NF, Executor >::ex_

Definition at line 117 of file time_source.h.

template<class E , class TP , class NF , class Executor >

NF folly::pushmi::time_source_queue< E, TP, NF, Executor >::nf_

Definition at line 116 of file time_source.h.

template<class E , class TP , class NF , class Executor >

std::weak_ptr<time_source_shared<E, time_point> > folly::pushmi::time_source_queue< E, TP, NF, Executor >::source_

Definition at line 115 of file time_source.h.

---

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

• proxygen/folly/folly/experimental/pushmi/time_source.h