ceph/src/librados/PoolAsyncCompletionImpl.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) 2004-2012 Sage Weil <sage@newdream.net>
   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_LIBRADOS_POOLASYNCCOMPLETIONIMPL_H
  16 #define CEPH_LIBRADOS_POOLASYNCCOMPLETIONIMPL_H
  17
  18 #include "common/ceph_mutex.h"
  19
  20 #include <boost/intrusive_ptr.hpp>
  21
  22 #include "include/rados/librados.h"
  23 #include "include/rados/librados.hpp"
  24
  25 namespace librados {
  26   struct PoolAsyncCompletionImpl {
  27     ceph::mutex lock = ceph::make_mutex("PoolAsyncCompletionImpl lock");
  28     ceph::condition_variable cond;
  29     int ref = 1;
  30     int rval = 0;
  31     bool released = false;
  32     bool done = false;
  33
  34     rados_callback_t callback = nullptr;
  35     void *callback_arg = nullptr;
  36
  37     PoolAsyncCompletionImpl() = default;
  38
  39     int set_callback(void *cb_arg, rados_callback_t cb) {
  40       std::scoped_lock l(lock);
  41       callback = cb;
  42       callback_arg = cb_arg;
  43       return 0;
  44     }
  45     int wait() {
  46       std::unique_lock l(lock);
  47       while (!done)
  48         cond.wait(l);
  49       return 0;
  50     }
  51     int is_complete() {
  52       std::scoped_lock l(lock);
  53       return done;
  54     }
  55     int get_return_value() {
  56       std::scoped_lock l(lock);
  57       return rval;
  58     }
  59     void get() {
  60       std::scoped_lock l(lock);
  61       ceph_assert(ref > 0);
  62       ref++;
  63     }
  64     void release() {
  65       std::scoped_lock l(lock);
  66       ceph_assert(!released);
  67       released = true;
  68     }
  69     void put() {
  70       std::unique_lock l(lock);
  71       int n = --ref;
  72       l.unlock();
  73       if (!n)
  74         delete this;
  75     }
  76   };
  77
  78   inline void intrusive_ptr_add_ref(PoolAsyncCompletionImpl* p) {
  79     p->get();
  80   }
  81   inline void intrusive_ptr_release(PoolAsyncCompletionImpl* p) {
  82     p->put();
  83   }
  84
  85   class CB_PoolAsync_Safe {
  86     boost::intrusive_ptr<PoolAsyncCompletionImpl> p;
  87
  88   public:
  89     explicit CB_PoolAsync_Safe(boost::intrusive_ptr<PoolAsyncCompletionImpl> p)
  90       : p(p) {}
  91     ~CB_PoolAsync_Safe() = default;
  92
  93     void operator()(int r) {
  94       auto c(std::move(p));
  95       std::unique_lock l(c->lock);
  96       c->rval = r;
  97       c->done = true;
  98       c->cond.notify_all();
  99
 100       if (c->callback) {
 101         rados_callback_t cb = c->callback;
 102         void *cb_arg = c->callback_arg;
 103         l.unlock();
 104         cb(c.get(), cb_arg);
 105         l.lock();
 106       }
 107     }
 108   };
 109 }
 110 #endif