test_stress_watch: exercise watch/unwatch/close from second client

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

diff --git a/src/test/rados-api/test.cc b/src/test/rados-api/test.cc
index d741ffac5ef903a195a4a540a39f075ae4a364b1..9ba9641c2b81fa7b7b8b0ac09f76f73b2eb16d16 100644 (file)
--- a/src/test/rados-api/test.cc
+++ b/src/test/rados-api/test.cc
@@ -96,6 +96,36 @@ std::string create_one_pool_pp(const std::string &pool_name, Rados &cluster)
   return "";
 }
 
+std::string connect_cluster_pp(Rados &cluster)
+{
+  char *id = getenv("CEPH_CLIENT_ID");
+  if (id) std::cerr << "Client id is: " << id << std::endl;
+
+  int ret;
+  ret = cluster.init(id);
+  if (ret) {
+    std::ostringstream oss;
+    oss << "cluster.init failed with error " << ret;
+    return oss.str();
+  }
+  ret = cluster.conf_read_file(NULL);
+  if (ret) {
+    cluster.shutdown();
+    std::ostringstream oss;
+    oss << "cluster.conf_read_file failed with error " << ret;
+    return oss.str();
+  }
+  cluster.conf_parse_env(NULL);
+  ret = cluster.connect();
+  if (ret) {
+    cluster.shutdown();
+    std::ostringstream oss;
+    oss << "cluster.connect failed with error " << ret;
+    return oss.str();
+  }
+  return "";
+}
+
 int destroy_one_pool(const std::string &pool_name, rados_t *cluster)
 {
   int ret = rados_pool_delete(*cluster, pool_name.c_str());

diff --git a/src/test/rados-api/test.h b/src/test/rados-api/test.h
index 8e22d87105bc2838f04cae6f3a85f6113b338b5e..df27ba0a687329da9fc08ee669b29331cb4c84d2 100644 (file)
--- a/src/test/rados-api/test.h
+++ b/src/test/rados-api/test.h
@@ -26,6 +26,7 @@ std::string get_temp_pool_name();
 std::string create_one_pool(const std::string &pool_name, rados_t *cluster);
 std::string create_one_pool_pp(const std::string &pool_name,
                            librados::Rados &cluster);
+std::string connect_cluster_pp(librados::Rados &cluster);
 int destroy_one_pool(const std::string &pool_name, rados_t *cluster);
 int destroy_one_pool_pp(const std::string &pool_name, librados::Rados &cluster);
 

diff --git a/src/test/test_stress_watch.cc b/src/test/test_stress_watch.cc
index 8679c920d4efd4e8acdf2009039701b8d8b4b708..d5cfc9e21ff9bad905be0f22d8422e26799eb2a4 100644 (file)
--- a/src/test/test_stress_watch.cc
+++ b/src/test/test_stress_watch.cc
@@ -1,5 +1,9 @@
 #include "include/rados/librados.h"
 #include "include/rados/librados.hpp"
+#include "include/atomic.h"
+#include "include/utime.h"
+#include "common/Thread.h"
+#include "common/Clock.h"
 #include "test/rados-api/test.h"
 
 #include "gtest/gtest.h"
@@ -10,6 +14,7 @@
 #include <iostream>
 #include <string>
 
+
 using namespace librados;
 using ceph::buffer;
 using std::map;
@@ -17,6 +22,7 @@ using std::ostringstream;
 using std::string;
 
 static sem_t sem;
+static atomic_t stop_flag;
 
 class WatchNotifyTestCtx : public WatchCtx
 {
@@ -27,6 +33,27 @@ public:
     }
 };
 
+struct WatcherUnwatcher : public Thread {
+  string pool;
+  WatcherUnwatcher(string& _pool) : pool(_pool) {}
+
+  void *entry() {
+    while (!stop_flag.read()) {
+      Rados cluster;
+      IoCtx ioctx;
+      connect_cluster_pp(cluster);
+      cluster.ioctx_create(pool.c_str(), ioctx);
+
+      uint64_t handle;
+      WatchNotifyTestCtx watch_ctx;
+      ioctx.watch("foo", 0, &handle, &watch_ctx);
+      bufferlist bl;
+      ioctx.unwatch("foo", handle);
+      ioctx.close();
+    }
+    return NULL;
+  }
+};
 TEST(WatchStress, Stress1) {
   ASSERT_EQ(0, sem_init(&sem, 0, 0));
   Rados cluster;
@@ -35,7 +62,9 @@ TEST(WatchStress, Stress1) {
   IoCtx ioctx;
   cluster.ioctx_create(pool_name.c_str(), ioctx);
   WatchNotifyTestCtx ctx;
-  
+
+  WatcherUnwatcher *thr = new WatcherUnwatcher(pool_name);
+  thr->create();
   ASSERT_EQ(0, ioctx.create("foo", false));
 
   for (int i = 0; i < 10000; ++i) {
@@ -44,12 +73,16 @@ TEST(WatchStress, Stress1) {
     WatchNotifyTestCtx ctx;
     ASSERT_EQ(0, ioctx.watch("foo", 0, &handle, &ctx));
     bufferlist bl2;
+    utime_t timestamp = ceph_clock_now(NULL);
     ASSERT_EQ(0, ioctx.notify("foo", 0, bl2));
+    timestamp = ceph_clock_now(NULL) - timestamp;
+    ASSERT_LT(timestamp.sec(), 5);
     TestAlarm alarm;
     sem_wait(&sem);
     ioctx.unwatch("foo", handle);
   }
-
+  stop_flag.set(1);
+  thr->join();
   ioctx.close();
   ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
   sem_destroy(&sem);