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