tube.h

Go to the documentation of this file.

#ifndef CVMFS_INGESTION_TUBE_H_
#define CVMFS_INGESTION_TUBE_H_

#include <pthread.h>
#include <stdint.h>

#include <cassert>
#include <vector>

#include "util/atomic.h"
#include "util/concurrency.h"
#include "util/pointer.h"
#include "util/single_copy.h"

template <class ItemT>
class Tube : SingleCopy {
 public:
  class Link : SingleCopy {
    friend class Tube<ItemT>;
   public:
    explicit Link(ItemT *item) : item_(item), next_(NULL), prev_(NULL) { }
    ItemT *item() { return item_; }

   private:
    ItemT *item_;
    Link *next_;
    Link *prev_;
  };

  Tube() : limit_(uint64_t(-1)), size_(0) { Init(); }
  explicit Tube(uint64_t limit) : limit_(limit), size_(0) {
    Init();
  }
  ~Tube() {
    Link *cursor = head_;
    do {
      Link *prev = cursor->prev_;
      delete cursor;
      cursor = prev;
    } while (cursor != head_);
    pthread_cond_destroy(&cond_populated_);
    pthread_cond_destroy(&cond_capacious_);
    pthread_cond_destroy(&cond_empty_);
    pthread_mutex_destroy(&lock_);
  }

  Link *EnqueueBack(ItemT *item) {
    assert(item != NULL);
    MutexLockGuard lock_guard(&lock_);
    while (size_ == limit_)
      pthread_cond_wait(&cond_capacious_, &lock_);

    Link *link = new Link(item);
    link->next_ = head_->next_;
    link->prev_ = head_;
    head_->next_->prev_ = link;
    head_->next_ = link;
    size_++;
    int retval = pthread_cond_signal(&cond_populated_);
    assert(retval == 0);
    return link;
  }

  Link *EnqueueFront(ItemT *item) {
    assert(item != NULL);
    MutexLockGuard lock_guard(&lock_);
    while (size_ == limit_)
      pthread_cond_wait(&cond_capacious_, &lock_);

    Link *link = new Link(item);
    link->next_ = head_;
    link->prev_ = head_->prev_;
    head_->prev_->next_ = link;
    head_->prev_ = link;
    size_++;
    int retval = pthread_cond_signal(&cond_populated_);
    assert(retval == 0);
    return link;
  }

  ItemT *Slice(Link *link) {
    MutexLockGuard lock_guard(&lock_);
    return SliceUnlocked(link);
  }

  ItemT *PopFront() {
    MutexLockGuard lock_guard(&lock_);
    while (size_ == 0)
      pthread_cond_wait(&cond_populated_, &lock_);
    return SliceUnlocked(head_->prev_);
  }

  ItemT *PopBack() {
    MutexLockGuard lock_guard(&lock_);
    while (size_ == 0)
      pthread_cond_wait(&cond_populated_, &lock_);
    return SliceUnlocked(head_->next_);
  }

  void Wait() {
    MutexLockGuard lock_guard(&lock_);
    while (size_ > 0)
      pthread_cond_wait(&cond_empty_, &lock_);
  }

  bool IsEmpty() {
    MutexLockGuard lock_guard(&lock_);
    return size_ == 0;
  }

  uint64_t size() {
    MutexLockGuard lock_guard(&lock_);
    return size_;
  }

 private:
  void Init() {
    Link *sentinel = new Link(NULL);
    head_ = sentinel;
    head_->next_ = head_->prev_ = sentinel;

    int retval = pthread_mutex_init(&lock_, NULL);
    assert(retval == 0);
    retval = pthread_cond_init(&cond_populated_, NULL);
    assert(retval == 0);
    retval = pthread_cond_init(&cond_capacious_, NULL);
    assert(retval == 0);
    retval = pthread_cond_init(&cond_empty_, NULL);
    assert(retval == 0);
  }

  ItemT *SliceUnlocked(Link *link) {
    // Cannot delete the sentinel link
    assert(link != head_);
    link->prev_->next_ = link->next_;
    link->next_->prev_ = link->prev_;
    ItemT *item = link->item_;
    delete link;
    size_--;
    int retval = pthread_cond_signal(&cond_capacious_);
    assert(retval == 0);
    if (size_ == 0) {
      retval = pthread_cond_broadcast(&cond_empty_);
      assert(retval == 0);
    }
    return item;
  }


  uint64_t limit_;
  uint64_t size_;
  Link *head_;
  pthread_mutex_t lock_;
  pthread_cond_t cond_populated_;
  pthread_cond_t cond_capacious_;
  pthread_cond_t cond_empty_;
};


template <class ItemT>
class TubeGroup : SingleCopy {
 public:
  TubeGroup() : is_active_(false) {
    atomic_init32(&round_robin_);
  }

  ~TubeGroup() {
    for (unsigned i = 0; i < tubes_.size(); ++i)
      delete tubes_[i];
  }

  void TakeTube(Tube<ItemT> *t) {
    assert(!is_active_);
    tubes_.push_back(t);
  }

  void Activate() {
    assert(!is_active_);
    assert(!tubes_.empty());
    is_active_ = true;
  }

  typename Tube<ItemT>::Link *Dispatch(ItemT *item) {
    assert(is_active_);
    unsigned tube_idx = (tubes_.size() == 1)
                        ? 0 : (item->tag() % tubes_.size());
    return tubes_[tube_idx]->EnqueueBack(item);
  }

  typename Tube<ItemT>::Link *DispatchAny(ItemT *item) {
    assert(is_active_);
    unsigned tube_idx = (tubes_.size() == 1)
                        ? 0 : (atomic_xadd32(&round_robin_, 1) % tubes_.size());
    return tubes_[tube_idx]->EnqueueBack(item);
  }

 private:
  bool is_active_;
  std::vector<Tube<ItemT> *> tubes_;
  atomic_int32 round_robin_;
};

#endif  // CVMFS_INGESTION_TUBE_H_