ceph/src/libradosstriper/MultiAioCompletionImpl.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2014 Sebastien Ponce <sebastien.ponce@cern.ch>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #ifndef CEPH_LIBRADOSSTRIPERSTRIPER_MULTIAIOCOMPLETIONIMPL_H
  16 #define CEPH_LIBRADOSSTRIPERSTRIPER_MULTIAIOCOMPLETIONIMPL_H
  17
  18 #include <list>
  19 #include <mutex>
  20 #include "common/ceph_mutex.h"
  21 #include "include/radosstriper/libradosstriper.hpp"
  22
  23 namespace libradosstriper {
  24
  25 struct MultiAioCompletionImpl {
  26
  27   ceph::mutex lock = ceph::make_mutex("MultiAioCompletionImpl lock", false);
  28   ceph::condition_variable cond;
  29   int ref, rval;
  30   int pending_complete, pending_safe;
  31   rados_callback_t callback_complete, callback_safe;
  32   void *callback_complete_arg, *callback_safe_arg;
  33   bool building;       ///< true if we are still building this completion
  34   bufferlist bl;       /// only used for read case in C api of rados striper
  35   std::list<bufferlist*> bllist; /// keep temporary buffer lists used for destriping
  36
  37   MultiAioCompletionImpl()
  38   : ref(1), rval(0),
  39     pending_complete(0), pending_safe(0),
  40     callback_complete(0), callback_safe(0),
  41     callback_complete_arg(0), callback_safe_arg(0),
  42     building(true) {};
  43
  44   ~MultiAioCompletionImpl() {
  45     // deallocate temporary buffer lists
  46     for (std::list<bufferlist*>::iterator it = bllist.begin();
  47          it != bllist.end();
  48          it++) {
  49       delete *it;
  50     }
  51     bllist.clear();
  52   }
  53
  54   int set_complete_callback(void *cb_arg, rados_callback_t cb) {
  55     std::scoped_lock l{lock};
  56     callback_complete = cb;
  57     callback_complete_arg = cb_arg;
  58     return 0;
  59   }
  60   int set_safe_callback(void *cb_arg, rados_callback_t cb) {
  61     std::scoped_lock l{lock};
  62     callback_safe = cb;
  63     callback_safe_arg = cb_arg;
  64     return 0;
  65   }
  66   int wait_for_complete() {
  67     std::unique_lock l{lock};
  68     cond.wait(l, [this] { return !pending_complete; });
  69     return 0;
  70   }
  71   int wait_for_safe() {
  72     std::unique_lock l{lock};
  73     cond.wait(l, [this] { return !pending_safe; });
  74     return 0;
  75   }
  76   bool is_complete() {
  77     std::scoped_lock l{lock};
  78     return pending_complete == 0;
  79   }
  80   bool is_safe() {
  81     std::scoped_lock l{lock};
  82     return pending_safe == 0;
  83   }
  84   void wait_for_complete_and_cb() {
  85     std::unique_lock l{lock};
  86     cond.wait(l, [this] { return !pending_complete && !callback_complete; });
  87   }
  88   void wait_for_safe_and_cb() {
  89     std::unique_lock l{lock};
  90     cond.wait(l, [this] { return !pending_safe && !callback_safe; });
  91   }
  92   bool is_complete_and_cb() {
  93     std::scoped_lock l{lock};
  94     return ((0 == pending_complete) && !callback_complete);
  95   }
  96   bool is_safe_and_cb() {
  97     std::scoped_lock l{lock};
  98     return ((0 == pending_safe) && !callback_safe);
  99   }
 100   int get_return_value() {
 101     std::scoped_lock l{lock};
 102     return rval;
 103   }
 104   void get() {
 105     std::scoped_lock l{lock};
 106     _get();
 107   }
 108   void _get() {
 109     ceph_assert(ceph_mutex_is_locked(lock));
 110     ceph_assert(ref > 0);
 111     ++ref;
 112   }
 113   void put() {
 114     lock.lock();
 115     put_unlock();
 116   }
 117   void put_unlock() {
 118     ceph_assert(ref > 0);
 119     int n = --ref;
 120     lock.unlock();
 121     if (!n)
 122       delete this;
 123   }
 124   void add_request() {
 125     std::scoped_lock l{lock};
 126     pending_complete++;
 127     _get();
 128     pending_safe++;
 129     _get();
 130   }
 131   void add_safe_request() {
 132     std::scoped_lock l{lock};
 133     pending_complete++;
 134     _get();
 135   }
 136   void complete() {
 137     ceph_assert(ceph_mutex_is_locked(lock));
 138     if (callback_complete) {
 139       callback_complete(this, callback_complete_arg);
 140       callback_complete = 0;
 141     }
 142     cond.notify_all();
 143   }
 144   void safe() {
 145     ceph_assert(ceph_mutex_is_locked(lock));
 146     if (callback_safe) {
 147       callback_safe(this, callback_safe_arg);
 148       callback_safe = 0;
 149     }
 150     cond.notify_all();
 151   };
 152
 153   void complete_request(ssize_t r);
 154   void safe_request(ssize_t r);
 155   void finish_adding_requests();
 156 };
 157
 158 inline void intrusive_ptr_add_ref(MultiAioCompletionImpl* ptr)
 159 {
 160   ptr->get();
 161 }
 162
 163 inline void intrusive_ptr_release(MultiAioCompletionImpl* ptr)
 164 {
 165   ptr->put();
 166 }
 167 }
 168
 169 #endif // CEPH_LIBRADOSSTRIPERSTRIPER_MULTIAIOCOMPLETIONIMPL_H