concurrency.h

Go to the documentation of this file.

#ifndef CVMFS_UTIL_CONCURRENCY_H_
#define CVMFS_UTIL_CONCURRENCY_H_

#include <pthread.h>

#include <cassert>
#include <queue>
#include <set>
#include <vector>

#include "util/async.h"
#include "util/atomic.h"
#include "util/export.h"
#include "util/mutex.h"
#include "util/single_copy.h"

#ifdef CVMFS_NAMESPACE_GUARD
namespace CVMFS_NAMESPACE_GUARD {
#endif

template<class ItemT>
class Channel : SingleCopy {
 public:
  Channel() {
    int retval = pthread_mutex_init(&lock_, NULL);
    assert(retval == 0);
    retval = pthread_cond_init(&cond_populated_, NULL);
    assert(retval == 0);
  }

  ~Channel() {
    pthread_cond_destroy(&cond_populated_);
    pthread_mutex_destroy(&lock_);
  }

  std::vector<ItemT *> *StartEnqueueing() {
    int retval = pthread_mutex_lock(&lock_);
    assert(retval == 0);
    return &items_;
  }

  void AbortEnqueueing() {
    int retval = pthread_mutex_unlock(&lock_);
    assert(retval == 0);
  }

  void CommitEnqueueing() {
    int retval = pthread_cond_signal(&cond_populated_);
    assert(retval == 0);
    retval = pthread_mutex_unlock(&lock_);
    assert(retval == 0);
  }

  void PushBack(ItemT *item) {
    MutexLockGuard lock_guard(&lock_);
    items_.push_back(item);
    int retval = pthread_cond_signal(&cond_populated_);
    assert(retval == 0);
  }

  ItemT *PopFront() {
    MutexLockGuard lock_guard(&lock_);
    while (items_.size() == 0)
      pthread_cond_wait(&cond_populated_, &lock_);
    ItemT *item = items_[0];
    items_.erase(items_.begin());
    return item;
  }

 private:
  std::vector<ItemT *> items_;
  pthread_mutex_t lock_;
  pthread_cond_t cond_populated_;
};

class CVMFS_EXPORT Lockable : SingleCopy {
 public:
  inline virtual ~Lockable() { pthread_mutex_destroy(&mutex_); }

  void Lock() const { pthread_mutex_lock(&mutex_); }
  int TryLock() const { return pthread_mutex_trylock(&mutex_); }
  void Unlock() const { pthread_mutex_unlock(&mutex_); }

 protected:
  Lockable() {
    const int retval = pthread_mutex_init(&mutex_, NULL);
    assert(retval == 0);
  }

 private:
  mutable pthread_mutex_t mutex_;
};


//
// -----------------------------------------------------------------------------
//


template<typename T>
class SynchronizingCounter : SingleCopy {
 public:
  SynchronizingCounter() : value_(T(0)), maximal_value_(T(0)) { Initialize(); }

  explicit SynchronizingCounter(const T maximal_value)
      : value_(T(0)), maximal_value_(maximal_value) {
    assert(maximal_value > T(0));
    Initialize();
  }

  ~SynchronizingCounter() { Destroy(); }

  T Increment() {
    MutexLockGuard l(mutex_);
    WaitForFreeSlotUnprotected();
    SetValueUnprotected(value_ + T(1));
    return value_;
  }

  T Decrement() {
    MutexLockGuard l(mutex_);
    SetValueUnprotected(value_ - T(1));
    return value_;
  }

  void WaitForZero() const {
    MutexLockGuard l(mutex_);
    while (value_ != T(0)) {
      pthread_cond_wait(&became_zero_, &mutex_);
    }
    assert(value_ == T(0));
  }

  bool HasMaximalValue() const { return maximal_value_ != T(0); }
  T maximal_value() const { return maximal_value_; }

  T operator++() { return Increment(); }
  const T operator++(int) { return Increment() - T(1); }
  T operator--() { return Decrement(); }
  const T operator--(int) { return Decrement() + T(1); }

  T Get() const {
    MutexLockGuard l(mutex_);
    return value_;
  }

  SynchronizingCounter<T> &operator=(const T &other) {
    MutexLockGuard l(mutex_);
    SetValueUnprotected(other);
    return *this;
  }

 protected:
  void SetValueUnprotected(const T new_value);
  void WaitForFreeSlotUnprotected();

 private:
  void Initialize();
  void Destroy();

 private:
  T value_;
  const T maximal_value_;

  mutable pthread_mutex_t mutex_;
  mutable pthread_cond_t became_zero_;
  pthread_cond_t free_slot_;
};


//
// -----------------------------------------------------------------------------
//


template<typename ParamT>
class Observable;


template<typename ParamT>
class Observable : public Callbackable<ParamT>, SingleCopy {
 public:
  typedef typename Callbackable<ParamT>::CallbackTN *CallbackPtr;

 protected:
  typedef std::set<CallbackPtr> Callbacks;

 public:
  virtual ~Observable();

  template<class DelegateT, class ClosureDataT>
  CallbackPtr RegisterListener(
      typename BoundClosure<ParamT, DelegateT, ClosureDataT>::CallbackMethod
          method,
      DelegateT *delegate,
      ClosureDataT data);

  template<class DelegateT>
  CallbackPtr RegisterListener(
      typename BoundCallback<ParamT, DelegateT>::CallbackMethod method,
      DelegateT *delegate);

  CallbackPtr RegisterListener(typename Callback<ParamT>::CallbackFunction fn);

  void UnregisterListener(CallbackPtr callback_object);

  void UnregisterListeners();

 protected:
  Observable();  // don't instantiate this as a stand alone object

  void RegisterListener(CallbackPtr callback_object);

  void NotifyListeners(const ParamT &parameter);

 private:
  Callbacks listeners_;  
  mutable pthread_rwlock_t listeners_rw_lock_;
};


//
// -----------------------------------------------------------------------------
//


CVMFS_EXPORT unsigned int GetNumberOfCpuCores();
static const unsigned int kFallbackNumberOfCpus = 1;


class CVMFS_EXPORT Signal : SingleCopy {
 public:
  Signal();
  ~Signal();
  void Wakeup();
  void Wait();
  bool IsSleeping();

 private:
  bool fired_;
  pthread_mutex_t lock_;
  pthread_cond_t signal_;
};


//
// -----------------------------------------------------------------------------
//


template<class T>
class FifoChannel : protected std::queue<T> {
 public:
  FifoChannel(const size_t maximal_length, const size_t drainout_threshold);
  virtual ~FifoChannel();

  void Enqueue(const T &data);

  const T Dequeue();

  unsigned int Drop();

  inline size_t GetItemCount() const;
  inline bool IsEmpty() const;
  inline size_t GetMaximalItemCount() const;

 private:
  // general configuration
  const size_t maximal_queue_length_;
  const size_t queue_drainout_threshold_;

  // thread synchronisation structures
  mutable pthread_mutex_t mutex_;
  mutable pthread_cond_t queue_is_not_empty_;
  mutable pthread_cond_t queue_is_not_full_;
};


template<class WorkerT>
class ConcurrentWorkers : public Observable<typename WorkerT::returned_data> {
 public:
  // these data types must be defined by the worker class
  typedef typename WorkerT::expected_data expected_data_t;
  typedef typename WorkerT::returned_data returned_data_t;
  typedef typename WorkerT::worker_context worker_context_t;

 protected:
  typedef std::vector<pthread_t> WorkerThreads;

  template<class DataT>
  struct Job {
    explicit Job(const DataT &data) : data(data), is_death_sentence(false) { }
    Job() : data(), is_death_sentence(true) { }
    const DataT data;              
    const bool is_death_sentence;  
  };
  typedef Job<expected_data_t> WorkerJob;
  typedef Job<returned_data_t> CallbackJob;

  struct RunBinding {
    explicit RunBinding(ConcurrentWorkers<WorkerT> *delegate)
        : delegate(delegate) { }
    ConcurrentWorkers<WorkerT> *delegate;  
  };

  struct WorkerRunBinding : RunBinding {
    WorkerRunBinding(ConcurrentWorkers<WorkerT> *delegate,
                     const worker_context_t *worker_context)
        : RunBinding(delegate), worker_context(worker_context) { }
    const worker_context_t *worker_context;
  };

 public:
  ConcurrentWorkers(const size_t number_of_workers,
                    const size_t maximal_queue_length,
                    worker_context_t *worker_context = NULL);
  virtual ~ConcurrentWorkers();

  bool Initialize();

  inline void Schedule(const expected_data_t &data) {
    Schedule(WorkerJob(data));
  }

  void Terminate();

  void WaitForEmptyQueue() const;

  void WaitForTermination();

  inline unsigned int GetNumberOfWorkers() const { return number_of_workers_; }
  inline unsigned int GetNumberOfFailedJobs() const {
    return atomic_read32(&jobs_failed_);
  }

  inline void JobSuccessful(const returned_data_t &data) {
    JobDone(data, true);
  }

  inline void JobFailed(const returned_data_t &data) { JobDone(data, false); }

  void RunCallbackThread();

 protected:
  bool SpawnWorkers();

  static void *RunWorker(void *run_binding);

  static void *RunCallbackThreadWrapper(void *run_binding);

  void ReportStartedWorker() const;

  void Schedule(WorkerJob job);
  void ScheduleDeathSentences();

  void TruncateJobQueue(const bool forget_pending = false);

  inline WorkerJob Acquire();

  void JobDone(const returned_data_t &data, const bool success = true);

  inline void StartRunning() {
    MutexLockGuard guard(status_mutex_);
    running_ = true;
  }
  inline void StopRunning() {
    MutexLockGuard guard(status_mutex_);
    running_ = false;
  }
  inline bool IsRunning() const {
    MutexLockGuard guard(status_mutex_);
    return running_;
  }

 private:
  // general configuration
  const size_t number_of_workers_;  
  const worker_context_t *worker_context_;  

  WorkerRunBinding thread_context_;

  // status information
  bool initialized_;
  bool running_;
  mutable unsigned int workers_started_;
  mutable pthread_mutex_t status_mutex_;
  mutable pthread_cond_t worker_started_;
  mutable pthread_mutex_t jobs_all_done_mutex_;
  mutable pthread_cond_t jobs_all_done_;

  // worker threads
  WorkerThreads worker_threads_;  
  pthread_t callback_thread_;     

  // job queue
  typedef FifoChannel<WorkerJob> JobQueue;
  JobQueue jobs_queue_;
  mutable atomic_int32 jobs_pending_;
  mutable atomic_int32 jobs_failed_;
  mutable atomic_int64 jobs_processed_;

  // callback channel
  typedef FifoChannel<CallbackJob> CallbackQueue;
  CallbackQueue results_queue_;
};


template<class DerivedWorkerT>
class ConcurrentWorker : SingleCopy {
 public:
  virtual ~ConcurrentWorker() { }

  virtual bool Initialize() { return true; }

  virtual void TearDown() { }

  // void operator()(const expected_data &data); // do the actual job of the
  // worker

 protected:
  ConcurrentWorker() : master_(NULL) { }

  inline ConcurrentWorkers<DerivedWorkerT> *master() const { return master_; }

 private:
  friend class ConcurrentWorkers<DerivedWorkerT>;
  void RegisterMaster(ConcurrentWorkers<DerivedWorkerT> *master) {
    master_ = master;
  }

 private:
  ConcurrentWorkers<DerivedWorkerT> *master_;
};

#ifdef CVMFS_NAMESPACE_GUARD
}  // namespace CVMFS_NAMESPACE_GUARD
#endif

#include "util/concurrency_impl.h"

#endif  // CVMFS_UTIL_CONCURRENCY_H_