+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-
-#include "test/librados_test_stub/TestRadosClient.h"
-#include "test/librados_test_stub/TestIoCtxImpl.h"
-#include "librados/AioCompletionImpl.h"
-#include "include/assert.h"
-#include "common/ceph_json.h"
-#include "common/Finisher.h"
-#include <boost/bind.hpp>
-#include <boost/thread.hpp>
-#include <errno.h>
-
-#include <atomic>
-
-static int get_concurrency() {
- int concurrency = 0;
- char *env = getenv("LIBRADOS_CONCURRENCY");
- if (env != NULL) {
- concurrency = atoi(env);
- }
- if (concurrency == 0) {
- concurrency = boost::thread::thread::hardware_concurrency();
- }
- if (concurrency == 0) {
- concurrency = 1;
- }
- return concurrency;
-}
-
-namespace librados {
-
-static void finish_aio_completion(AioCompletionImpl *c, int r) {
- c->lock.Lock();
- c->complete = true;
- c->rval = r;
- c->lock.Unlock();
-
- rados_callback_t cb_complete = c->callback_complete;
- void *cb_complete_arg = c->callback_complete_arg;
- if (cb_complete) {
- cb_complete(c, cb_complete_arg);
- }
-
- rados_callback_t cb_safe = c->callback_safe;
- void *cb_safe_arg = c->callback_safe_arg;
- if (cb_safe) {
- cb_safe(c, cb_safe_arg);
- }
-
- c->lock.Lock();
- c->callback_complete = NULL;
- c->callback_safe = NULL;
- c->cond.Signal();
- c->put_unlock();
-}
-
-class AioFunctionContext : public Context {
-public:
- AioFunctionContext(const TestRadosClient::AioFunction &callback,
- Finisher *finisher, AioCompletionImpl *c)
- : m_callback(callback), m_finisher(finisher), m_comp(c)
- {
- if (m_comp != NULL) {
- m_comp->get();
- }
- }
-
- void finish(int r) override {
- int ret = m_callback();
- if (m_comp != NULL) {
- if (m_finisher != NULL) {
- m_finisher->queue(new FunctionContext(boost::bind(
- &finish_aio_completion, m_comp, ret)));
- } else {
- finish_aio_completion(m_comp, ret);
- }
- }
- }
-private:
- TestRadosClient::AioFunction m_callback;
- Finisher *m_finisher;
- AioCompletionImpl *m_comp;
-};
-
-TestRadosClient::TestRadosClient(CephContext *cct,
- TestWatchNotify *watch_notify)
- : m_cct(cct->get()), m_watch_notify(watch_notify),
- m_aio_finisher(new Finisher(m_cct))
-{
- get();
-
- // simulate multiple OSDs
- int concurrency = get_concurrency();
- for (int i = 0; i < concurrency; ++i) {
- m_finishers.push_back(new Finisher(m_cct));
- m_finishers.back()->start();
- }
-
- // replicate AIO callback processing
- m_aio_finisher->start();
-}
-
-TestRadosClient::~TestRadosClient() {
- flush_aio_operations();
-
- for (size_t i = 0; i < m_finishers.size(); ++i) {
- m_finishers[i]->stop();
- delete m_finishers[i];
- }
- m_aio_finisher->stop();
- delete m_aio_finisher;
-
- m_cct->put();
- m_cct = NULL;
-}
-
-void TestRadosClient::get() {
- m_refcount++;
-}
-
-void TestRadosClient::put() {
- if (--m_refcount == 0) {
- shutdown();
- delete this;
- }
-}
-
-CephContext *TestRadosClient::cct() {
- return m_cct;
-}
-
-int TestRadosClient::connect() {
- return 0;
-}
-
-void TestRadosClient::shutdown() {
-}
-
-int TestRadosClient::wait_for_latest_osdmap() {
- return 0;
-}
-
-int TestRadosClient::mon_command(const std::vector<std::string>& cmd,
- const bufferlist &inbl,
- bufferlist *outbl, std::string *outs) {
- for (std::vector<std::string>::const_iterator it = cmd.begin();
- it != cmd.end(); ++it) {
- JSONParser parser;
- if (!parser.parse(it->c_str(), it->length())) {
- return -EINVAL;
- }
-
- JSONObjIter j_it = parser.find("prefix");
- if (j_it.end()) {
- return -EINVAL;
- }
-
- if ((*j_it)->get_data() == "osd tier add") {
- return 0;
- } else if ((*j_it)->get_data() == "osd tier cache-mode") {
- return 0;
- } else if ((*j_it)->get_data() == "osd tier set-overlay") {
- return 0;
- } else if ((*j_it)->get_data() == "osd tier remove-overlay") {
- return 0;
- } else if ((*j_it)->get_data() == "osd tier remove") {
- return 0;
- }
- }
- return -ENOSYS;
-}
-
-void TestRadosClient::add_aio_operation(const std::string& oid,
- bool queue_callback,
- const AioFunction &aio_function,
- AioCompletionImpl *c) {
- AioFunctionContext *ctx = new AioFunctionContext(
- aio_function, queue_callback ? m_aio_finisher : NULL, c);
- get_finisher(oid)->queue(ctx);
-}
-
-struct WaitForFlush {
- int flushed() {
- if (--count == 0) {
- aio_finisher->queue(new FunctionContext(boost::bind(
- &finish_aio_completion, c, 0)));
- delete this;
- }
- return 0;
- }
-
- std::atomic<int64_t> count = { 0 };
- Finisher *aio_finisher;
- AioCompletionImpl *c;
-};
-
-void TestRadosClient::flush_aio_operations() {
- AioCompletionImpl *comp = new AioCompletionImpl();
- flush_aio_operations(comp);
- comp->wait_for_safe();
- comp->put();
-}
-
-void TestRadosClient::flush_aio_operations(AioCompletionImpl *c) {
- c->get();
-
- WaitForFlush *wait_for_flush = new WaitForFlush();
- wait_for_flush->count = m_finishers.size();
- wait_for_flush->aio_finisher = m_aio_finisher;
- wait_for_flush->c = c;
-
- for (size_t i = 0; i < m_finishers.size(); ++i) {
- AioFunctionContext *ctx = new AioFunctionContext(
- boost::bind(&WaitForFlush::flushed, wait_for_flush),
- nullptr, nullptr);
- m_finishers[i]->queue(ctx);
- }
-}
-
-int TestRadosClient::aio_watch_flush(AioCompletionImpl *c) {
- c->get();
- Context *ctx = new FunctionContext(boost::bind(
- &TestRadosClient::finish_aio_completion, this, c, _1));
- get_watch_notify()->aio_flush(this, ctx);
- return 0;
-}
-
-void TestRadosClient::finish_aio_completion(AioCompletionImpl *c, int r) {
- librados::finish_aio_completion(c, r);
-}
-
-Finisher *TestRadosClient::get_finisher(const std::string &oid) {
- std::size_t h = m_hash(oid);
- return m_finishers[h % m_finishers.size()];
-}
-
-} // namespace librados
Code Review / stor4nfv.git / blobdiff