ceph/src/test/librados_test_stub/TestRadosClient.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "test/librados_test_stub/TestRadosClient.h"
   5 #include "test/librados_test_stub/TestIoCtxImpl.h"
   6 #include "librados/AioCompletionImpl.h"
   7 #include "include/assert.h"
   8 #include "common/ceph_json.h"
   9 #include "common/Finisher.h"
  10 #include <boost/bind.hpp>
  11 #include <boost/thread.hpp>
  12 #include <errno.h>
  13
  14 static int get_concurrency() {
  15   int concurrency = 0;
  16   char *env = getenv("LIBRADOS_CONCURRENCY");
  17   if (env != NULL) {
  18     concurrency = atoi(env);
  19   }
  20   if (concurrency == 0) {
  21     concurrency = boost::thread::thread::hardware_concurrency();
  22   }
  23   if (concurrency == 0) {
  24     concurrency = 1;
  25   }
  26   return concurrency;
  27 }
  28
  29 namespace librados {
  30
  31 static void finish_aio_completion(AioCompletionImpl *c, int r) {
  32   c->lock.Lock();
  33   c->complete = true;
  34   c->rval = r;
  35   c->lock.Unlock();
  36
  37   rados_callback_t cb_complete = c->callback_complete;
  38   void *cb_complete_arg = c->callback_complete_arg;
  39   if (cb_complete) {
  40     cb_complete(c, cb_complete_arg);
  41   }
  42
  43   rados_callback_t cb_safe = c->callback_safe;
  44   void *cb_safe_arg = c->callback_safe_arg;
  45   if (cb_safe) {
  46     cb_safe(c, cb_safe_arg);
  47   }
  48
  49   c->lock.Lock();
  50   c->callback_complete = NULL;
  51   c->callback_safe = NULL;
  52   c->cond.Signal();
  53   c->put_unlock();
  54 }
  55
  56 class AioFunctionContext : public Context {
  57 public:
  58   AioFunctionContext(const TestRadosClient::AioFunction &callback,
  59                      Finisher *finisher, AioCompletionImpl *c)
  60     : m_callback(callback), m_finisher(finisher), m_comp(c)
  61   {
  62     if (m_comp != NULL) {
  63       m_comp->get();
  64     }
  65   }
  66
  67   void finish(int r) override {
  68     int ret = m_callback();
  69     if (m_comp != NULL) {
  70       if (m_finisher != NULL) {
  71         m_finisher->queue(new FunctionContext(boost::bind(
  72           &finish_aio_completion, m_comp, ret)));
  73       } else {
  74         finish_aio_completion(m_comp, ret);
  75       }
  76     }
  77   }
  78 private:
  79   TestRadosClient::AioFunction m_callback;
  80   Finisher *m_finisher;
  81   AioCompletionImpl *m_comp;
  82 };
  83
  84 TestRadosClient::TestRadosClient(CephContext *cct,
  85                                  TestWatchNotify *watch_notify)
  86   : m_cct(cct->get()), m_watch_notify(watch_notify),
  87     m_aio_finisher(new Finisher(m_cct))
  88 {
  89   get();
  90
  91   // simulate multiple OSDs
  92   int concurrency = get_concurrency();
  93   for (int i = 0; i < concurrency; ++i) {
  94     m_finishers.push_back(new Finisher(m_cct));
  95     m_finishers.back()->start();
  96   }
  97
  98   // replicate AIO callback processing
  99   m_aio_finisher->start();
 100 }
 101
 102 TestRadosClient::~TestRadosClient() {
 103   flush_aio_operations();
 104
 105   for (size_t i = 0; i < m_finishers.size(); ++i) {
 106     m_finishers[i]->stop();
 107     delete m_finishers[i];
 108   }
 109   m_aio_finisher->stop();
 110   delete m_aio_finisher;
 111
 112   m_cct->put();
 113   m_cct = NULL;
 114 }
 115
 116 void TestRadosClient::get() {
 117   m_refcount.inc();
 118 }
 119
 120 void TestRadosClient::put() {
 121   if (m_refcount.dec() == 0) {
 122     shutdown();
 123     delete this;
 124   }
 125 }
 126
 127 CephContext *TestRadosClient::cct() {
 128   return m_cct;
 129 }
 130
 131 int TestRadosClient::connect() {
 132   return 0;
 133 }
 134
 135 void TestRadosClient::shutdown() {
 136 }
 137
 138 int TestRadosClient::wait_for_latest_osdmap() {
 139   return 0;
 140 }
 141
 142 int TestRadosClient::mon_command(const std::vector<std::string>& cmd,
 143                                  const bufferlist &inbl,
 144                                  bufferlist *outbl, std::string *outs) {
 145   for (std::vector<std::string>::const_iterator it = cmd.begin();
 146        it != cmd.end(); ++it) {
 147     JSONParser parser;
 148     if (!parser.parse(it->c_str(), it->length())) {
 149       return -EINVAL;
 150     }
 151
 152     JSONObjIter j_it = parser.find("prefix");
 153     if (j_it.end()) {
 154       return -EINVAL;
 155     }
 156
 157     if ((*j_it)->get_data() == "osd tier add") {
 158       return 0;
 159     } else if ((*j_it)->get_data() == "osd tier cache-mode") {
 160       return 0;
 161     } else if ((*j_it)->get_data() == "osd tier set-overlay") {
 162       return 0;
 163     } else if ((*j_it)->get_data() == "osd tier remove-overlay") {
 164       return 0;
 165     } else if ((*j_it)->get_data() == "osd tier remove") {
 166       return 0;
 167     }
 168   }
 169   return -ENOSYS;
 170 }
 171
 172 void TestRadosClient::add_aio_operation(const std::string& oid,
 173                                         bool queue_callback,
 174                                         const AioFunction &aio_function,
 175                                         AioCompletionImpl *c) {
 176   AioFunctionContext *ctx = new AioFunctionContext(
 177     aio_function, queue_callback ? m_aio_finisher : NULL, c);
 178   get_finisher(oid)->queue(ctx);
 179 }
 180
 181 struct WaitForFlush {
 182   int flushed() {
 183     if (count.dec() == 0) {
 184       aio_finisher->queue(new FunctionContext(boost::bind(
 185         &finish_aio_completion, c, 0)));
 186       delete this;
 187     }
 188     return 0;
 189   }
 190
 191   atomic_t count;
 192   Finisher *aio_finisher;
 193   AioCompletionImpl *c;
 194 };
 195
 196 void TestRadosClient::flush_aio_operations() {
 197   AioCompletionImpl *comp = new AioCompletionImpl();
 198   flush_aio_operations(comp);
 199   comp->wait_for_safe();
 200   comp->put();
 201 }
 202
 203 void TestRadosClient::flush_aio_operations(AioCompletionImpl *c) {
 204   c->get();
 205
 206   WaitForFlush *wait_for_flush = new WaitForFlush();
 207   wait_for_flush->count.set(m_finishers.size());
 208   wait_for_flush->aio_finisher = m_aio_finisher;
 209   wait_for_flush->c = c;
 210
 211   for (size_t i = 0; i < m_finishers.size(); ++i) {
 212     AioFunctionContext *ctx = new AioFunctionContext(
 213       boost::bind(&WaitForFlush::flushed, wait_for_flush),
 214       nullptr, nullptr);
 215     m_finishers[i]->queue(ctx);
 216   }
 217 }
 218
 219 int TestRadosClient::aio_watch_flush(AioCompletionImpl *c) {
 220   c->get();
 221   Context *ctx = new FunctionContext(boost::bind(
 222     &TestRadosClient::finish_aio_completion, this, c, _1));
 223   get_watch_notify()->aio_flush(this, ctx);
 224   return 0;
 225 }
 226
 227 void TestRadosClient::finish_aio_completion(AioCompletionImpl *c, int r) {
 228   librados::finish_aio_completion(c, r);
 229 }
 230
 231 Finisher *TestRadosClient::get_finisher(const std::string &oid) {
 232   std::size_t h = m_hash(oid);
 233   return m_finishers[h % m_finishers.size()];
 234 }
 235
 236 } // namespace librados