ceph/src/test/librados/watch_notify.cc

   1 #include "include/rados/librados.h"
   2 #include "include/rados/rados_types.h"
   3 #include "test/librados/test.h"
   4 #include "test/librados/TestCase.h"
   5
   6 #include <errno.h>
   7 #include <fcntl.h>
   8 #include <semaphore.h>
   9 #include "gtest/gtest.h"
  10 #include "include/encoding.h"
  11 #include <set>
  12 #include <map>
  13
  14 typedef RadosTestEC LibRadosWatchNotifyEC;
  15
  16 int notify_sleep = 0;
  17
  18 // notify
  19 static sem_t *sem;
  20
  21 static void watch_notify_test_cb(uint8_t opcode, uint64_t ver, void *arg)
  22 {
  23   std::cout << __func__ << std::endl;
  24   sem_post(sem);
  25 }
  26
  27 class LibRadosWatchNotify : public RadosTest
  28 {
  29 protected:
  30   // notify 2
  31   bufferlist notify_bl;
  32   std::set<uint64_t> notify_cookies;
  33   rados_ioctx_t notify_io;
  34   const char *notify_oid = nullptr;
  35   int notify_err = 0;
  36
  37   static void watch_notify2_test_cb(void *arg,
  38                                     uint64_t notify_id,
  39                                     uint64_t cookie,
  40                                     uint64_t notifier_gid,
  41                                     void *data,
  42                                     size_t data_len);
  43   static void watch_notify2_test_errcb(void *arg, uint64_t cookie, int err);
  44 };
  45
  46
  47 void LibRadosWatchNotify::watch_notify2_test_cb(void *arg,
  48                                   uint64_t notify_id,
  49                                   uint64_t cookie,
  50                                   uint64_t notifier_gid,
  51                                   void *data,
  52                                   size_t data_len)
  53 {
  54   std::cout << __func__ << " from " << notifier_gid << " notify_id " << notify_id
  55             << " cookie " << cookie << std::endl;
  56   ceph_assert(notifier_gid > 0);
  57   auto thiz = reinterpret_cast<LibRadosWatchNotify*>(arg);
  58   ceph_assert(thiz);
  59   thiz->notify_cookies.insert(cookie);
  60   thiz->notify_bl.clear();
  61   thiz->notify_bl.append((char*)data, data_len);
  62   if (notify_sleep)
  63     sleep(notify_sleep);
  64   rados_notify_ack(thiz->notify_io, thiz->notify_oid, notify_id, cookie,
  65                    "reply", 5);
  66 }
  67
  68 void LibRadosWatchNotify::watch_notify2_test_errcb(void *arg,
  69                                                    uint64_t cookie,
  70                                                    int err)
  71 {
  72   std::cout << __func__ << " cookie " << cookie << " err " << err << std::endl;
  73   ceph_assert(cookie > 1000);
  74   auto thiz = reinterpret_cast<LibRadosWatchNotify*>(arg);
  75   ceph_assert(thiz);
  76   thiz->notify_err = err;
  77 }
  78
  79 class WatchNotifyTestCtx2;
  80
  81 // --
  82
  83 #pragma GCC diagnostic ignored "-Wpragmas"
  84 #pragma GCC diagnostic push
  85 #pragma GCC diagnostic ignored "-Wdeprecated-declarations"
  86
  87 TEST_F(LibRadosWatchNotify, WatchNotify) {
  88   ASSERT_NE(SEM_FAILED, (sem = sem_open("/test_watch_notify_sem", O_CREAT, 0644, 0)));
  89   char buf[128];
  90   memset(buf, 0xcc, sizeof(buf));
  91   ASSERT_EQ(0, rados_write(ioctx, "foo", buf, sizeof(buf), 0));
  92   uint64_t handle;
  93   ASSERT_EQ(0,
  94       rados_watch(ioctx, "foo", 0, &handle, watch_notify_test_cb, NULL));
  95   ASSERT_EQ(0, rados_notify(ioctx, "foo", 0, NULL, 0));
  96   TestAlarm alarm;
  97   sem_wait(sem);
  98   rados_unwatch(ioctx, "foo", handle);
  99
 100   // when dne ...
 101   ASSERT_EQ(-ENOENT,
 102       rados_watch(ioctx, "dne", 0, &handle, watch_notify_test_cb, NULL));
 103
 104   sem_close(sem);
 105 }
 106
 107 TEST_F(LibRadosWatchNotifyEC, WatchNotify) {
 108   ASSERT_NE(SEM_FAILED, (sem = sem_open("/test_watch_notify_sem", O_CREAT, 0644, 0)));
 109   char buf[128];
 110   memset(buf, 0xcc, sizeof(buf));
 111   ASSERT_EQ(0, rados_write(ioctx, "foo", buf, sizeof(buf), 0));
 112   uint64_t handle;
 113   ASSERT_EQ(0,
 114       rados_watch(ioctx, "foo", 0, &handle, watch_notify_test_cb, NULL));
 115   ASSERT_EQ(0, rados_notify(ioctx, "foo", 0, NULL, 0));
 116   TestAlarm alarm;
 117   sem_wait(sem);
 118   rados_unwatch(ioctx, "foo", handle);
 119   sem_close(sem);
 120 }
 121
 122 #pragma GCC diagnostic pop
 123 #pragma GCC diagnostic warning "-Wpragmas"
 124
 125
 126 // --
 127
 128 TEST_F(LibRadosWatchNotify, Watch2Delete) {
 129   notify_io = ioctx;
 130   notify_oid = "foo";
 131   notify_err = 0;
 132   char buf[128];
 133   memset(buf, 0xcc, sizeof(buf));
 134   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 135   uint64_t handle;
 136   ASSERT_EQ(0,
 137             rados_watch2(ioctx, notify_oid, &handle,
 138                          watch_notify2_test_cb,
 139                          watch_notify2_test_errcb, this));
 140   ASSERT_EQ(0, rados_remove(ioctx, notify_oid));
 141   int left = 300;
 142   std::cout << "waiting up to " << left << " for disconnect notification ..."
 143             << std::endl;
 144   while (notify_err == 0 && --left) {
 145     sleep(1);
 146   }
 147   ASSERT_TRUE(left > 0);
 148   ASSERT_EQ(-ENOTCONN, notify_err);
 149   ASSERT_EQ(-ENOTCONN, rados_watch_check(ioctx, handle));
 150   rados_unwatch2(ioctx, handle);
 151   rados_watch_flush(cluster);
 152 }
 153
 154 TEST_F(LibRadosWatchNotify, AioWatchDelete) {
 155   notify_io = ioctx;
 156   notify_oid = "foo";
 157   notify_err = 0;
 158   char buf[128];
 159   memset(buf, 0xcc, sizeof(buf));
 160   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 161
 162
 163   rados_completion_t comp;
 164   uint64_t handle;
 165   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 166   rados_aio_watch(ioctx, notify_oid, comp, &handle,
 167                   watch_notify2_test_cb, watch_notify2_test_errcb, this);
 168   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 169   ASSERT_EQ(0, rados_aio_get_return_value(comp));
 170   rados_aio_release(comp);
 171   ASSERT_EQ(0, rados_remove(ioctx, notify_oid));
 172   int left = 300;
 173   std::cout << "waiting up to " << left << " for disconnect notification ..."
 174             << std::endl;
 175   while (notify_err == 0 && --left) {
 176     sleep(1);
 177   }
 178   ASSERT_TRUE(left > 0);
 179   ASSERT_EQ(-ENOTCONN, notify_err);
 180   ASSERT_EQ(-ENOTCONN, rados_watch_check(ioctx, handle));
 181   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 182   rados_aio_unwatch(ioctx, handle, comp);
 183   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 184   ASSERT_EQ(-ENOENT, rados_aio_get_return_value(comp));
 185   rados_aio_release(comp);
 186 }
 187
 188 // --
 189
 190 TEST_F(LibRadosWatchNotify, WatchNotify2) {
 191   notify_io = ioctx;
 192   notify_oid = "foo";
 193   notify_cookies.clear();
 194   char buf[128];
 195   memset(buf, 0xcc, sizeof(buf));
 196   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 197   uint64_t handle;
 198   ASSERT_EQ(0,
 199       rados_watch2(ioctx, notify_oid, &handle,
 200                    watch_notify2_test_cb,
 201                    watch_notify2_test_errcb, this));
 202   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 203   char *reply_buf = 0;
 204   size_t reply_buf_len;
 205   ASSERT_EQ(0, rados_notify2(ioctx, notify_oid,
 206                              "notify", 6, 300000,
 207                              &reply_buf, &reply_buf_len));
 208   bufferlist reply;
 209   reply.append(reply_buf, reply_buf_len);
 210   std::map<std::pair<uint64_t,uint64_t>, bufferlist> reply_map;
 211   std::set<std::pair<uint64_t,uint64_t> > missed_map;
 212   auto reply_p = reply.cbegin();
 213   decode(reply_map, reply_p);
 214   decode(missed_map, reply_p);
 215   ASSERT_EQ(1u, reply_map.size());
 216   ASSERT_EQ(0u, missed_map.size());
 217   ASSERT_EQ(1u, notify_cookies.size());
 218   ASSERT_EQ(1u, notify_cookies.count(handle));
 219   ASSERT_EQ(5u, reply_map.begin()->second.length());
 220   ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5));
 221   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 222   rados_buffer_free(reply_buf);
 223
 224   // try it on a non-existent object ... our buffer pointers
 225   // should get zeroed.
 226   ASSERT_EQ(-ENOENT, rados_notify2(ioctx, "doesnotexist",
 227                                    "notify", 6, 300000,
 228                                    &reply_buf, &reply_buf_len));
 229   ASSERT_EQ((char*)0, reply_buf);
 230   ASSERT_EQ(0u, reply_buf_len);
 231
 232   rados_unwatch2(ioctx, handle);
 233   rados_watch_flush(cluster);
 234 }
 235
 236 TEST_F(LibRadosWatchNotify, AioWatchNotify2) {
 237   notify_io = ioctx;
 238   notify_oid = "foo";
 239   notify_cookies.clear();
 240   char buf[128];
 241   memset(buf, 0xcc, sizeof(buf));
 242   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 243
 244   rados_completion_t comp;
 245   uint64_t handle;
 246   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 247   rados_aio_watch(ioctx, notify_oid, comp, &handle,
 248                   watch_notify2_test_cb, watch_notify2_test_errcb, this);
 249   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 250   ASSERT_EQ(0, rados_aio_get_return_value(comp));
 251   rados_aio_release(comp);
 252
 253   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 254   char *reply_buf = 0;
 255   size_t reply_buf_len;
 256   ASSERT_EQ(0, rados_notify2(ioctx, notify_oid,
 257                              "notify", 6, 300000,
 258                              &reply_buf, &reply_buf_len));
 259   bufferlist reply;
 260   reply.append(reply_buf, reply_buf_len);
 261   std::map<std::pair<uint64_t,uint64_t>, bufferlist> reply_map;
 262   std::set<std::pair<uint64_t,uint64_t> > missed_map;
 263   auto reply_p = reply.cbegin();
 264   decode(reply_map, reply_p);
 265   decode(missed_map, reply_p);
 266   ASSERT_EQ(1u, reply_map.size());
 267   ASSERT_EQ(0u, missed_map.size());
 268   ASSERT_EQ(1u, notify_cookies.size());
 269   ASSERT_EQ(1u, notify_cookies.count(handle));
 270   ASSERT_EQ(5u, reply_map.begin()->second.length());
 271   ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5));
 272   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 273   rados_buffer_free(reply_buf);
 274
 275   // try it on a non-existent object ... our buffer pointers
 276   // should get zeroed.
 277   ASSERT_EQ(-ENOENT, rados_notify2(ioctx, "doesnotexist",
 278                                    "notify", 6, 300000,
 279                                    &reply_buf, &reply_buf_len));
 280   ASSERT_EQ((char*)0, reply_buf);
 281   ASSERT_EQ(0u, reply_buf_len);
 282
 283   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 284   rados_aio_unwatch(ioctx, handle, comp);
 285   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 286   ASSERT_EQ(0, rados_aio_get_return_value(comp));
 287   rados_aio_release(comp);
 288 }
 289
 290 TEST_F(LibRadosWatchNotify, AioNotify) {
 291   notify_io = ioctx;
 292   notify_oid = "foo";
 293   notify_cookies.clear();
 294   char buf[128];
 295   memset(buf, 0xcc, sizeof(buf));
 296   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 297   uint64_t handle;
 298   ASSERT_EQ(0,
 299       rados_watch2(ioctx, notify_oid, &handle,
 300                    watch_notify2_test_cb,
 301                    watch_notify2_test_errcb, this));
 302   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 303   char *reply_buf = 0;
 304   size_t reply_buf_len;
 305   rados_completion_t comp;
 306   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 307   ASSERT_EQ(0, rados_aio_notify(ioctx, "foo", comp, "notify", 6, 300000,
 308                                 &reply_buf, &reply_buf_len));
 309   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 310   ASSERT_EQ(0, rados_aio_get_return_value(comp));
 311   rados_aio_release(comp);
 312
 313   bufferlist reply;
 314   reply.append(reply_buf, reply_buf_len);
 315   std::map<std::pair<uint64_t,uint64_t>, bufferlist> reply_map;
 316   std::set<std::pair<uint64_t,uint64_t> > missed_map;
 317   auto reply_p = reply.cbegin();
 318   decode(reply_map, reply_p);
 319   decode(missed_map, reply_p);
 320   ASSERT_EQ(1u, reply_map.size());
 321   ASSERT_EQ(0u, missed_map.size());
 322   ASSERT_EQ(1u, notify_cookies.size());
 323   ASSERT_EQ(1u, notify_cookies.count(handle));
 324   ASSERT_EQ(5u, reply_map.begin()->second.length());
 325   ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5));
 326   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 327   rados_buffer_free(reply_buf);
 328
 329   // try it on a non-existent object ... our buffer pointers
 330   // should get zeroed.
 331   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 332   ASSERT_EQ(0, rados_aio_notify(ioctx, "doesnotexist", comp, "notify", 6,
 333                                 300000, &reply_buf, &reply_buf_len));
 334   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 335   ASSERT_EQ(-ENOENT, rados_aio_get_return_value(comp));
 336   rados_aio_release(comp);
 337   ASSERT_EQ((char*)0, reply_buf);
 338   ASSERT_EQ(0u, reply_buf_len);
 339
 340   rados_unwatch2(ioctx, handle);
 341   rados_watch_flush(cluster);
 342 }
 343
 344 // --
 345
 346 TEST_F(LibRadosWatchNotify, WatchNotify2Multi) {
 347   notify_io = ioctx;
 348   notify_oid = "foo";
 349   notify_cookies.clear();
 350   char buf[128];
 351   memset(buf, 0xcc, sizeof(buf));
 352   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 353   uint64_t handle1, handle2;
 354   ASSERT_EQ(0,
 355       rados_watch2(ioctx, notify_oid, &handle1,
 356                    watch_notify2_test_cb,
 357                    watch_notify2_test_errcb, this));
 358   ASSERT_EQ(0,
 359       rados_watch2(ioctx, notify_oid, &handle2,
 360                    watch_notify2_test_cb,
 361                    watch_notify2_test_errcb, this));
 362   ASSERT_GT(rados_watch_check(ioctx, handle1), 0);
 363   ASSERT_GT(rados_watch_check(ioctx, handle2), 0);
 364   ASSERT_NE(handle1, handle2);
 365   char *reply_buf = 0;
 366   size_t reply_buf_len;
 367   ASSERT_EQ(0, rados_notify2(ioctx, notify_oid,
 368                              "notify", 6, 300000,
 369                              &reply_buf, &reply_buf_len));
 370   bufferlist reply;
 371   reply.append(reply_buf, reply_buf_len);
 372   std::map<std::pair<uint64_t,uint64_t>, bufferlist> reply_map;
 373   std::set<std::pair<uint64_t,uint64_t> > missed_map;
 374   auto reply_p = reply.cbegin();
 375   decode(reply_map, reply_p);
 376   decode(missed_map, reply_p);
 377   ASSERT_EQ(2u, reply_map.size());
 378   ASSERT_EQ(5u, reply_map.begin()->second.length());
 379   ASSERT_EQ(0u, missed_map.size());
 380   ASSERT_EQ(2u, notify_cookies.size());
 381   ASSERT_EQ(1u, notify_cookies.count(handle1));
 382   ASSERT_EQ(1u, notify_cookies.count(handle2));
 383   ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5));
 384   ASSERT_GT(rados_watch_check(ioctx, handle1), 0);
 385   ASSERT_GT(rados_watch_check(ioctx, handle2), 0);
 386   rados_buffer_free(reply_buf);
 387   rados_unwatch2(ioctx, handle1);
 388   rados_unwatch2(ioctx, handle2);
 389   rados_watch_flush(cluster);
 390 }
 391
 392 // --
 393
 394 TEST_F(LibRadosWatchNotify, WatchNotify2Timeout) {
 395   notify_io = ioctx;
 396   notify_oid = "foo";
 397   notify_sleep = 3; // 3s
 398   notify_cookies.clear();
 399   char buf[128];
 400   memset(buf, 0xcc, sizeof(buf));
 401   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 402   uint64_t handle;
 403   ASSERT_EQ(0,
 404       rados_watch2(ioctx, notify_oid, &handle,
 405                    watch_notify2_test_cb,
 406                    watch_notify2_test_errcb, this));
 407   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 408   char *reply_buf = 0;
 409   size_t reply_buf_len;
 410   ASSERT_EQ(-ETIMEDOUT, rados_notify2(ioctx, notify_oid,
 411                                       "notify", 6, 1000, // 1s
 412                                       &reply_buf, &reply_buf_len));
 413   ASSERT_EQ(1u, notify_cookies.size());
 414   {
 415     bufferlist reply;
 416     reply.append(reply_buf, reply_buf_len);
 417     std::map<std::pair<uint64_t,uint64_t>, bufferlist> reply_map;
 418     std::set<std::pair<uint64_t,uint64_t> > missed_map;
 419     auto reply_p = reply.cbegin();
 420     decode(reply_map, reply_p);
 421     decode(missed_map, reply_p);
 422     ASSERT_EQ(0u, reply_map.size());
 423     ASSERT_EQ(1u, missed_map.size());
 424   }
 425   rados_buffer_free(reply_buf);
 426
 427   // we should get the next notify, though!
 428   notify_sleep = 0;
 429   notify_cookies.clear();
 430   ASSERT_EQ(0, rados_notify2(ioctx, notify_oid,
 431                              "notify", 6, 300000, // 300s
 432                              &reply_buf, &reply_buf_len));
 433   ASSERT_EQ(1u, notify_cookies.size());
 434   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 435
 436   rados_unwatch2(ioctx, handle);
 437
 438   rados_completion_t comp;
 439   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 440   rados_aio_watch_flush(cluster, comp);
 441   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 442   ASSERT_EQ(0, rados_aio_get_return_value(comp));
 443   rados_aio_release(comp);
 444   rados_buffer_free(reply_buf);
 445
 446 }
 447
 448 TEST_F(LibRadosWatchNotify, Watch3Timeout) {
 449   notify_io = ioctx;
 450   notify_oid = "foo";
 451   notify_cookies.clear();
 452   notify_err = 0;
 453   char buf[128];
 454   memset(buf, 0xcc, sizeof(buf));
 455   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 456   uint64_t handle;
 457   time_t start = time(0);
 458   const uint32_t timeout = 4;
 459   {
 460     // make sure i timeout before the messenger reconnects to the OSD,
 461     // it will resend a watch request on behalf of the client, and the
 462     // timer of timeout on OSD side will be reset by the new request.
 463     char conf[128];
 464     ASSERT_EQ(0, rados_conf_get(cluster,
 465                                 "ms_tcp_read_timeout",
 466                                 conf, sizeof(conf)));
 467     auto tcp_read_timeout = std::stoll(conf);
 468     ASSERT_LT(timeout, tcp_read_timeout);
 469   }
 470   ASSERT_EQ(0,
 471             rados_watch3(ioctx, notify_oid, &handle,
 472                      watch_notify2_test_cb, watch_notify2_test_errcb,
 473                      timeout, this));
 474   int age = rados_watch_check(ioctx, handle);
 475   time_t age_bound = time(0) + 1 - start;
 476   ASSERT_LT(age, age_bound * 1000);
 477   ASSERT_GT(age, 0);
 478   rados_conf_set(cluster, "objecter_inject_no_watch_ping", "true");
 479   // allow a long time here since an osd peering event will renew our
 480   // watch.
 481   int left = 16 * timeout;
 482   std::cout << "waiting up to " << left << " for osd to time us out ..."
 483             << std::endl;
 484   while (notify_err == 0 && --left) {
 485     sleep(1);
 486   }
 487   ASSERT_GT(left, 0);
 488   rados_conf_set(cluster, "objecter_inject_no_watch_ping", "false");
 489   ASSERT_EQ(-ENOTCONN, notify_err);
 490   ASSERT_EQ(-ENOTCONN, rados_watch_check(ioctx, handle));
 491
 492   // a subsequent notify should not reach us
 493   char *reply_buf = nullptr;
 494   size_t reply_buf_len;
 495   ASSERT_EQ(0, rados_notify2(ioctx, notify_oid,
 496                              "notify", 6, 300000,
 497                              &reply_buf, &reply_buf_len));
 498   {
 499     bufferlist reply;
 500     reply.append(reply_buf, reply_buf_len);
 501     std::map<std::pair<uint64_t,uint64_t>, bufferlist> reply_map;
 502     std::set<std::pair<uint64_t,uint64_t> > missed_map;
 503     auto reply_p = reply.cbegin();
 504     decode(reply_map, reply_p);
 505     decode(missed_map, reply_p);
 506     ASSERT_EQ(0u, reply_map.size());
 507     ASSERT_EQ(0u, missed_map.size());
 508   }
 509   ASSERT_EQ(0u, notify_cookies.size());
 510   ASSERT_EQ(-ENOTCONN, rados_watch_check(ioctx, handle));
 511   rados_buffer_free(reply_buf);
 512
 513   // re-watch
 514   rados_unwatch2(ioctx, handle);
 515   rados_watch_flush(cluster);
 516
 517   handle = 0;
 518   ASSERT_EQ(0,
 519             rados_watch2(ioctx, notify_oid, &handle,
 520                          watch_notify2_test_cb,
 521                          watch_notify2_test_errcb, this));
 522   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 523
 524   // and now a notify will work.
 525   ASSERT_EQ(0, rados_notify2(ioctx, notify_oid,
 526                              "notify", 6, 300000,
 527                              &reply_buf, &reply_buf_len));
 528   {
 529     bufferlist reply;
 530     reply.append(reply_buf, reply_buf_len);
 531     std::map<std::pair<uint64_t,uint64_t>, bufferlist> reply_map;
 532     std::set<std::pair<uint64_t,uint64_t> > missed_map;
 533     auto reply_p = reply.cbegin();
 534     decode(reply_map, reply_p);
 535     decode(missed_map, reply_p);
 536     ASSERT_EQ(1u, reply_map.size());
 537     ASSERT_EQ(0u, missed_map.size());
 538     ASSERT_EQ(1u, notify_cookies.count(handle));
 539     ASSERT_EQ(5u, reply_map.begin()->second.length());
 540     ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5));
 541   }
 542   ASSERT_EQ(1u, notify_cookies.size());
 543   ASSERT_GT(rados_watch_check(ioctx, handle), 0);
 544
 545   rados_buffer_free(reply_buf);
 546   rados_unwatch2(ioctx, handle);
 547   rados_watch_flush(cluster);
 548 }
 549
 550 TEST_F(LibRadosWatchNotify, AioWatchDelete2) {
 551   notify_io = ioctx;
 552   notify_oid = "foo";
 553   notify_err = 0;
 554   char buf[128];
 555   uint32_t timeout = 3;
 556   memset(buf, 0xcc, sizeof(buf));
 557   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
 558
 559
 560   rados_completion_t comp;
 561   uint64_t handle;
 562   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 563   rados_aio_watch2(ioctx, notify_oid, comp, &handle,
 564                   watch_notify2_test_cb, watch_notify2_test_errcb, timeout, this);
 565   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 566   ASSERT_EQ(0, rados_aio_get_return_value(comp));
 567   rados_aio_release(comp);
 568   ASSERT_EQ(0, rados_remove(ioctx, notify_oid));
 569   int left = 30;
 570   std::cout << "waiting up to " << left << " for disconnect notification ..."
 571       << std::endl;
 572   while (notify_err == 0 && --left) {
 573     sleep(1);
 574   }
 575   ASSERT_TRUE(left > 0);
 576   ASSERT_EQ(-ENOTCONN, notify_err);
 577   ASSERT_EQ(-ENOTCONN, rados_watch_check(ioctx, handle));
 578   ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &comp));
 579   rados_aio_unwatch(ioctx, handle, comp);
 580   ASSERT_EQ(0, rados_aio_wait_for_complete(comp));
 581   ASSERT_EQ(-ENOENT, rados_aio_get_return_value(comp));
 582   rados_aio_release(comp);
 583 }