folly::pushmi::time_source_shared< E, TP > Class Template Reference

#include <time_source.h>

Inheritance diagram for folly::pushmi::time_source_shared< E, TP >:

Public Member Functions
	~time_source_shared ()
	time_source_shared ()
void	insert (std::shared_ptr< time_source_queue_base< E, TP >> queue, time_heap_item< E, TP > item)
Public Member Functions inherited from folly::pushmi::time_source_shared_base< E, TP >
	time_source_shared_base ()

Static Public Member Functions
static void	start (std::shared_ptr< time_source_shared< E, TP >> that)
static void	join (std::shared_ptr< time_source_shared< E, TP >> that)
static void	worker (std::shared_ptr< time_source_shared< E, TP >> that)

Public Attributes
std::thread	t_
std::vector< std::shared_ptr< time_source_queue_base< E, TP > > >	ready_
Public Attributes inherited from folly::pushmi::time_source_shared_base< E, TP >
std::mutex	lock_
std::condition_variable	wake_
std::thread	t_
std::chrono::system_clock::time_point	earliest_
bool	done_
bool	joined_
int	dirty_
int	items_
detail::opt< E >	error_
std::deque< std::shared_ptr< time_source_queue_base< E, TP > > >	pending_

Additional Inherited Members
Public Types inherited from folly::pushmi::time_source_shared_base< E, TP >
using	time_point = std::decay_t< TP >

Detailed Description

template<class E, class TP>
class folly::pushmi::time_source_shared< E, TP >

Definition at line 31 of file time_source.h.

Constructor & Destructor Documentation

template<class E, class TP>

folly::pushmi::time_source_shared< E, TP >::~time_source_shared ( )

inline

Definition at line 304 of file time_source.h.

                        {
    // not allowed to be discarded without joining and completing all queued
    // items
    if (t_.joinable() || this->items_ != 0) {
      std::abort();
    }
  }

template<class E, class TP>

folly::pushmi::time_source_shared< E, TP >::time_source_shared ( )

inline

Definition at line 311 of file time_source.h.

{}

Member Function Documentation

template<class E, class TP>

void folly::pushmi::time_source_shared< E, TP >::insert ( std::shared_ptr< time_source_queue_base< E, TP >>  queue, time_heap_item< E, TP >  item  )

inline

Definition at line 403 of file time_source.h.

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

    // deliver error_ and return
    if (!!this->error_) {
      set_error(item.what, *this->error_);
      return;
    }
    // once join() is called, new work queued to the executor is not safe unless
    // it is nested in an existing item.
    if (!!this->joined_) {
      std::abort();
    };

    queue->heap_.push(std::move(item));
    ++this->items_;

    if (!queue->dispatching_ && !queue->pending_) {
      // add queue to pending pending_ list if it is not already there
      this->pending_.push_back(queue);
      queue->pending_ = true;
    }

    if (queue->heap_.top().when < this->earliest_) {
      // this is the earliest, tell worker to reset earliest_
      ++this->dirty_;
      this->wake_.notify_one();
    }
  }

template<class E, class TP>

static void folly::pushmi::time_source_shared< E, TP >::join ( std::shared_ptr< time_source_shared< E, TP >>  that )

inlinestatic

Definition at line 316 of file time_source.h.

                                                                  {
    std::unique_lock<std::mutex> guard{that->lock_};
    that->done_ = true;
    ++that->dirty_;
    that->wake_.notify_one();
    guard.unlock();
    that->t_.join();
  }

template<class E, class TP>

static void folly::pushmi::time_source_shared< E, TP >::start ( std::shared_ptr< time_source_shared< E, TP >>  that )

inlinestatic

Definition at line 313 of file time_source.h.

                                                                   {
    that->t_ = std::thread{&time_source_shared<E, TP>::worker, that};
  }

template<class E, class TP>

static void folly::pushmi::time_source_shared< E, TP >::worker ( std::shared_ptr< time_source_shared< E, TP >>  that )

inlinestatic

Definition at line 325 of file time_source.h.

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

      // once done_, keep going until empty
      while (!that->done_ || that->items_ > 0) {
        // wait for something to do
        that->wake_.wait_until(guard, that->earliest_, [&]() {
          return that->dirty_ != 0 ||
              std::chrono::system_clock::now() >= that->earliest_;
        });
        that->dirty_ = 0;

        //
        // select ready and empty queues and reset earliest_

        auto start = std::chrono::system_clock::now();
        auto earliest = start + std::chrono::hours(24);
        auto process = time_item_process_pred_fn<E, TP>{&start, &earliest};

        auto process_begin = std::partition(
            that->pending_.begin(), that->pending_.end(), process);
        that->earliest_ = earliest;

        // copy out the queues that have ready items so that the lock
        // is not held during dispatch

        std::copy_if(
            process_begin,
            that->pending_.end(),
            std::back_inserter(that->ready_),
            time_queue_dispatch_pred_fn<E, TP>{});

        // remove processed queues from pending queue.
        that->pending_.erase(process_begin, that->pending_.end());

        // printf("d %lu, %lu, %d, %ld\n", that->pending_.size(),
        // that->ready_.size(), that->items_,
        // std::chrono::duration_cast<std::chrono::milliseconds>(earliest -
        // start).count());

        // dispatch to queues with ready items
        guard.unlock();
        for (auto& q : that->ready_) {
          q->dispatch();
        }
        guard.lock();
        that->ready_.clear();
      }
      that->joined_ = true;
    } catch (...) {
      //
      // block any more items from being enqueued, all new items will be sent
      // this error on the same context that calls submit
      //
      // also dispatch errors to all items already in the queues from the
      // time thread
      std::unique_lock<std::mutex> guard{that->lock_};
      // creates a dependency that std::exception_ptr must be ConvertibleTo E
      // TODO: break this dependency rather than enforce it with concepts
      that->error_ = std::current_exception();
      for (auto& q : that->pending_) {
        while (!q->heap_.empty()) {
          try {
            auto what{std::move(q->top().what)};
            q->heap_.pop();
            --that->items_;
            guard.unlock();
            set_error(what, *that->error_);
            guard.lock();
          } catch (...) {
            // we already have an error, ignore this one.
          }
        }
      }
    }
  }

Member Data Documentation

template<class E, class TP>

std::vector<std::shared_ptr<time_source_queue_base<E, TP> > > folly::pushmi::time_source_shared< E, TP >::ready_

Definition at line 302 of file time_source.h.

template<class E, class TP>

std::thread folly::pushmi::time_source_shared< E, TP >::t_

Definition at line 299 of file time_source.h.

---

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

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