1 files changed, 161 insertions, 0 deletions
diff --git a/src/test/immutable_object_cache/test_multi_session.cc b/src/test/immutable_object_cache/test_multi_session.cc
new file mode 100644
index 000000000..c0c629ab0
--- /dev/null
+++ b/src/test/immutable_object_cache/test_multi_session.cc
@@ -0,0 +1,161 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+#include <iostream>
+#include <unistd.h>
+
+#include "gtest/gtest.h"
+#include "include/Context.h"
+#include "global/global_init.h"
+#include "global/global_context.h"
+
+#include "test/immutable_object_cache/test_common.h"
+#include "tools/immutable_object_cache/CacheClient.h"
+#include "tools/immutable_object_cache/CacheServer.h"
+
+using namespace ceph::immutable_obj_cache;
+
+class TestMultiSession : public ::testing::Test {
+public:
+  std::string m_local_path;
+  CacheServer* m_cache_server;
+  std::thread* m_cache_server_thread;
+  std::vector<CacheClient*> m_cache_client_vec;
+  WaitEvent m_wait_event;
+  std::atomic<uint64_t> m_send_request_index;
+  std::atomic<uint64_t> m_recv_ack_index;
+  uint64_t m_session_num = 110;
+
+  TestMultiSession() : m_local_path("/tmp/ceph_test_multisession_socket"),
+                       m_cache_server_thread(nullptr),  m_send_request_index(0),
+                       m_recv_ack_index(0) {
+    m_cache_client_vec.resize(m_session_num + 1, nullptr);
+  }
+
+  ~TestMultiSession() {}
+
+  static void SetUpTestCase() {}
+  static void TearDownTestCase() {}
+
+  void SetUp() override {
+    std::remove(m_local_path.c_str());
+    m_cache_server = new CacheServer(g_ceph_context, m_local_path,
+      [this](CacheSession* session_id, ObjectCacheRequest* req){
+        server_handle_request(session_id, req);
+    });
+    ASSERT_TRUE(m_cache_server != nullptr);
+
+    m_cache_server_thread = new std::thread(([this]() {
+        m_wait_event.signal();
+        m_cache_server->run();
+    }));
+
+    // waiting for thread running.
+    m_wait_event.wait();
+
+    // waiting for io_service run.
+    usleep(2);
+  }
+
+  void TearDown() override {
+    for (uint64_t i = 0; i < m_session_num; i++) {
+      if (m_cache_client_vec[i] != nullptr) {
+        m_cache_client_vec[i]->close();
+        delete m_cache_client_vec[i];
+      }
+    }
+    m_cache_server->stop();
+    if (m_cache_server_thread->joinable()) {
+      m_cache_server_thread->join();
+    }
+    delete m_cache_server;
+    delete m_cache_server_thread;
+
+    std::remove(m_local_path.c_str());
+  }
+
+  CacheClient* create_session(uint64_t random_index) {
+     CacheClient* cache_client = new CacheClient(m_local_path, g_ceph_context);
+     cache_client->run();
+     while (true) {
+       if (0 == cache_client->connect()) {
+         break;
+       }
+     }
+    m_cache_client_vec[random_index] = cache_client;
+    return cache_client;
+  }
+
+  void server_handle_request(CacheSession* session_id, ObjectCacheRequest* req) {
+
+    switch (req->get_request_type()) {
+      case RBDSC_REGISTER: {
+        ObjectCacheRequest* reply = new ObjectCacheRegReplyData(RBDSC_REGISTER_REPLY,
+                                                                req->seq);
+        session_id->send(reply);
+        break;
+      }
+      case RBDSC_READ: {
+        ObjectCacheRequest* reply = new ObjectCacheReadReplyData(RBDSC_READ_REPLY,
+                                                                req->seq);
+        session_id->send(reply);
+        break;
+      }
+    }
+  }
+
+  void test_register_client(uint64_t random_index) {
+    ASSERT_TRUE(m_cache_client_vec[random_index] == nullptr);
+
+    auto ctx = new LambdaContext([](int ret){
+       ASSERT_TRUE(ret == 0);
+    });
+    auto session = create_session(random_index);
+    session->register_client(ctx);
+
+    ASSERT_TRUE(m_cache_client_vec[random_index] != nullptr);
+    ASSERT_TRUE(session->is_session_work());
+  }
+
+  void test_lookup_object(std::string pool_nspace, uint64_t index,
+                          uint64_t request_num, bool is_last) {
+
+    for (uint64_t i = 0; i < request_num; i++) {
+      auto ctx = make_gen_lambda_context<ObjectCacheRequest*,
+            std::function<void(ObjectCacheRequest*)>>([this](ObjectCacheRequest* ack) {
+        m_recv_ack_index++;
+      });
+      m_send_request_index++;
+      // here just for concurrently testing register + lookup, so fix object id.
+      m_cache_client_vec[index]->lookup_object(pool_nspace, 1, 2, 3, "1234", std::move(ctx));
+    }
+
+    if (is_last) {
+      while(m_send_request_index != m_recv_ack_index) {
+        usleep(1);
+      }
+      m_wait_event.signal();
+    }
+  }
+};
+
+// test concurrent : multi-session + register_client + lookup_request
+TEST_F(TestMultiSession, test_multi_session) {
+
+  uint64_t test_times = 1000;
+  uint64_t test_session_num = 100;
+
+  for (uint64_t i = 0; i <= test_times; i++) {
+    uint64_t random_index = random() % test_session_num;
+    if (m_cache_client_vec[random_index] == nullptr) {
+      test_register_client(random_index);
+    } else {
+      test_lookup_object(string("test_nspace") + std::to_string(random_index),
+                         random_index, 4, i == test_times ? true : false);
+    }
+  }
+
+  // make sure all ack will be received.
+  m_wait_event.wait();
+
+  ASSERT_TRUE(m_send_request_index == m_recv_ack_index);
+}