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_