librados_test_stub: AIO operation callbacks should be via Finisher

librados will execute all AIO callbacks via a single finisher to
prevent blocking the Objecter.  Reproduce this behavior to avoid
deadlocks that only exist when using the test stub.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>

diff --git a/src/test/librados_test_stub/TestIoCtxImpl.cc b/src/test/librados_test_stub/TestIoCtxImpl.cc
index 553afb689b34f2aa07ff07ddc11bd4d4ed546c2e..f810906a19e645aed340ce888dccaae4a6aa05bd 100644 (file)
--- a/src/test/librados_test_stub/TestIoCtxImpl.cc
+++ b/src/test/librados_test_stub/TestIoCtxImpl.cc
@@ -90,7 +90,7 @@ int TestIoCtxImpl::aio_operate(const std::string& oid, TestObjectOperationImpl &
                                int flags) {
   // TODO ignoring snap_context and flags for now
   ops.get();
-  m_client->add_aio_operation(oid, boost::bind(
+  m_client->add_aio_operation(oid, true, boost::bind(
     &TestIoCtxImpl::execute_aio_operations, this, oid, &ops,
     reinterpret_cast<bufferlist*>(NULL)), c);
   return 0;
@@ -102,7 +102,7 @@ int TestIoCtxImpl::aio_operate_read(const std::string& oid,
                                     bufferlist *pbl) {
   // TODO ignoring flags for now
   ops.get();
-  m_client->add_aio_operation(oid, boost::bind(
+  m_client->add_aio_operation(oid, true, boost::bind(
     &TestIoCtxImpl::execute_aio_operations, this, oid, &ops, pbl), c);
   return 0;
 }
@@ -137,11 +137,14 @@ void TestIoCtxImpl::notify_ack(const std::string& o, uint64_t notify_id,
 
 int TestIoCtxImpl::operate(const std::string& oid, TestObjectOperationImpl &ops) {
   AioCompletionImpl *comp = new AioCompletionImpl();
-  int ret = aio_operate(oid, ops, comp, NULL, 0);
-  if (ret == 0) {
-    comp->wait_for_safe();
-    ret = comp->get_return_value();
-  }
+
+  ops.get();
+  m_client->add_aio_operation(oid, false, boost::bind(
+    &TestIoCtxImpl::execute_aio_operations, this, oid, &ops,
+    reinterpret_cast<bufferlist*>(NULL)), comp);
+
+  comp->wait_for_safe();
+  int ret = comp->get_return_value();
   comp->put();
   return ret;
 }
@@ -149,11 +152,13 @@ int TestIoCtxImpl::operate(const std::string& oid, TestObjectOperationImpl &ops)
 int TestIoCtxImpl::operate_read(const std::string& oid, TestObjectOperationImpl &ops,
                                 bufferlist *pbl) {
   AioCompletionImpl *comp = new AioCompletionImpl();
-  int ret = aio_operate_read(oid, ops, comp, 0, pbl);
-  if (ret == 0) {
-    comp->wait_for_complete();
-    ret = comp->get_return_value();
-  }
+
+  ops.get();
+  m_client->add_aio_operation(oid, false, boost::bind(
+    &TestIoCtxImpl::execute_aio_operations, this, oid, &ops, pbl), comp);
+
+  comp->wait_for_complete();
+  int ret = comp->get_return_value();
   comp->put();
   return ret;
 }

diff --git a/src/test/librados_test_stub/TestMemIoCtxImpl.cc b/src/test/librados_test_stub/TestMemIoCtxImpl.cc
index 05103b0c299dd1900541417b047fac75fabd4f93..a9dac606155ae5703a62c22a668d2b68171fef81 100644 (file)
--- a/src/test/librados_test_stub/TestMemIoCtxImpl.cc
+++ b/src/test/librados_test_stub/TestMemIoCtxImpl.cc
@@ -37,7 +37,8 @@ TestIoCtxImpl *TestMemIoCtxImpl::clone() {
 }
 
 int TestMemIoCtxImpl::aio_remove(const std::string& oid, AioCompletionImpl *c) {
-  m_client->add_aio_operation(oid, boost::bind(&TestMemIoCtxImpl::remove, this, oid),
+  m_client->add_aio_operation(oid, true,
+                              boost::bind(&TestMemIoCtxImpl::remove, this, oid),
                               c);
   return 0;
 }

diff --git a/src/test/librados_test_stub/TestRadosClient.cc b/src/test/librados_test_stub/TestRadosClient.cc
index 7db26bc89dba9397eafea69114cc2d86d0588ad6..925dc598385b32c6920163fe84ce38a2bc960ff6 100644 (file)
--- a/src/test/librados_test_stub/TestRadosClient.cc
+++ b/src/test/librados_test_stub/TestRadosClient.cc
@@ -57,8 +57,8 @@ static void finish_aio_completion(AioCompletionImpl *c, int r) {
 class AioFunctionContext : public Context {
 public:
   AioFunctionContext(const TestRadosClient::AioFunction &callback,
-                     AioCompletionImpl *c)
-    : m_callback(callback), m_comp(c)
+                     Finisher *finisher, AioCompletionImpl *c)
+    : m_callback(callback), m_finisher(finisher), m_comp(c)
   {
     if (m_comp != NULL) {
       m_comp->get();
@@ -68,11 +68,17 @@ public:
   virtual void finish(int r) {
     int ret = m_callback();
     if (m_comp != NULL) {
-      finish_aio_completion(m_comp, ret);
+      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;
 };
 
@@ -82,11 +88,16 @@ TestRadosClient::TestRadosClient(CephContext *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 = new Finisher(m_cct);
+  m_aio_finisher->start();
 }
 
 TestRadosClient::~TestRadosClient() {
@@ -96,6 +107,8 @@ TestRadosClient::~TestRadosClient() {
     m_finishers[i]->stop();
     delete m_finishers[i];
   }
+  m_aio_finisher->stop();
+  delete m_aio_finisher;
 
   m_cct->put();
   m_cct = NULL;
@@ -162,9 +175,11 @@ int TestRadosClient::mon_command(const std::vector<std::string>& cmd,
 }
 
 void TestRadosClient::add_aio_operation(const std::string& oid,
+                                        bool queue_callback,
                                        const AioFunction &aio_function,
                                         AioCompletionImpl *c) {
-  AioFunctionContext *ctx = new AioFunctionContext(aio_function, c);
+  AioFunctionContext *ctx = new AioFunctionContext(
+    aio_function, queue_callback ? m_aio_finisher : NULL, c);
   get_finisher(oid)->queue(ctx);
 }
 
@@ -200,7 +215,7 @@ void TestRadosClient::flush_aio_operations(AioCompletionImpl *c) {
   for (size_t i = 0; i < m_finishers.size(); ++i) {
     AioFunctionContext *ctx = new AioFunctionContext(
       boost::bind(&WaitForFlush::flushed, wait_for_flush),
-      NULL);
+      m_aio_finisher, NULL);
     m_finishers[i]->queue(ctx);
   }
 }

diff --git a/src/test/librados_test_stub/TestRadosClient.h b/src/test/librados_test_stub/TestRadosClient.h
index 9bb5b1c90c21ba7672394c63c320c41e6e8e18b6..a0611051cac522b0f242a298d0588fecd2834288 100644 (file)
--- a/src/test/librados_test_stub/TestRadosClient.h
+++ b/src/test/librados_test_stub/TestRadosClient.h
@@ -72,7 +72,7 @@ public:
     return m_watch_notify;
   }
 
-  void add_aio_operation(const std::string& oid,
+  void add_aio_operation(const std::string& oid, bool queue_callback,
                         const AioFunction &aio_function, AioCompletionImpl *c);
   void flush_aio_operations();
   void flush_aio_operations(AioCompletionImpl *c);
@@ -87,6 +87,7 @@ private:
 
   Finisher *get_finisher(const std::string& oid);
 
+  Finisher *m_aio_finisher;
   std::vector<Finisher *> m_finishers;
   boost::hash<std::string> m_hash;