ceph/src/common/QueueRing.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #ifndef QUEUE_RING_H
   5 #define QUEUE_RING_H
   6
   7 #include "common/ceph_mutex.h"
   8
   9 #include <list>
  10 #include <atomic>
  11 #include <vector>
  12
  13 template <class T>
  14 class QueueRing {
  15   struct QueueBucket {
  16     ceph::mutex lock = ceph::make_mutex("QueueRing::QueueBucket::lock");
  17     ceph::condition_variable cond;
  18     typename std::list<T> entries;
  19
  20     QueueBucket() {}
  21     QueueBucket(const QueueBucket& rhs) {
  22       entries = rhs.entries;
  23     }
  24
  25     void enqueue(const T& entry) {
  26       lock.lock();
  27       if (entries.empty()) {
  28         cond.notify_all();
  29       }
  30       entries.push_back(entry);
  31       lock.unlock();
  32     }
  33
  34     void dequeue(T *entry) {
  35       std::unique_lock l(lock);
  36       while (entries.empty()) {
  37         cond.wait(l);
  38       };
  39       ceph_assert(!entries.empty());
  40       *entry = entries.front();
  41       entries.pop_front();
  42     };
  43   };
  44
  45   std::vector<QueueBucket> buckets;
  46   int num_buckets;
  47
  48   std::atomic<int64_t> cur_read_bucket = { 0 };
  49   std::atomic<int64_t> cur_write_bucket = { 0 };
  50
  51 public:
  52   QueueRing(int n) : buckets(n), num_buckets(n) {
  53   }
  54
  55   void enqueue(const T& entry) {
  56     buckets[++cur_write_bucket % num_buckets].enqueue(entry);
  57   };
  58
  59   void dequeue(T *entry) {
  60     buckets[++cur_read_bucket % num_buckets].dequeue(entry);
  61   }
  62 };
  63
  64 #endif