From: Sage Weil Date: Mon, 17 Nov 2014 16:02:55 +0000 (-0800) Subject: librados: include missed watchers in notify reply data X-Git-Tag: v0.91~89 X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=5e994508580ba720dbc7b6578f792b1b5e6bbab1;p=ceph.git librados: include missed watchers in notify reply data Include a list of who missed their notify in the notify reply. Signed-off-by: Sage Weil --- diff --git a/src/include/rados/librados.h b/src/include/rados/librados.h index 528adddf91a..0e1fed4c462 100644 --- a/src/include/rados/librados.h +++ b/src/include/rados/librados.h @@ -2064,6 +2064,10 @@ CEPH_RADOS_API int rados_notify(rados_ioctx_t io, const char *o, uint64_t ver, * le32 buflen length of reply message buffer * u8 * buflen payload * } * num_acks + * le32 num_timeouts + * { + * le64 gid global id for the client + * } * num_timeouts * * Note: There may be multiple instances of the same gid if there are * multiple watchers registered via the same client. @@ -2071,6 +2075,10 @@ CEPH_RADOS_API int rados_notify(rados_ioctx_t io, const char *o, uint64_t ver, * Note: The buffer must be released with rados_buffer_free() when the * user is done with it. * + * Note: Since the result buffer includes clients that time out, it + * will be set even when rados_notify() returns an error code (like + * -ETIMEDOUT). + * * @param io the pool the object is in * @param o the name of the object * @param buf data to send to watchers diff --git a/src/librados/IoCtxImpl.cc b/src/librados/IoCtxImpl.cc index f821c2115e1..957e8a2e238 100644 --- a/src/librados/IoCtxImpl.cc +++ b/src/librados/IoCtxImpl.cc @@ -1237,21 +1237,29 @@ int librados::IoCtxImpl::notify(const object_t& oid, bufferlist& bl, << linger_op << dendl; notify_private.wait(); - // pass result back to user - if (notify_private.result >= 0) { - if (preply_buf) { - *preply_buf = (char*)malloc(notify_private.reply_bl.length()); - memcpy(*preply_buf, notify_private.reply_bl.c_str(), - notify_private.reply_bl.length()); - } - if (preply_buf_len) - *preply_buf_len = notify_private.reply_bl.length(); - if (preply_bl) - preply_bl->claim(notify_private.reply_bl); + ldout(client->cct, 10) << __func__ << " completed notify (linger op " + << linger_op << "), r = " << notify_private.result + << dendl; + } else { + ldout(client->cct, 10) << __func__ << " failed to initiate notify, r = " + << r_issue << dendl; + } + + // pass result back to user + // NOTE: we do this regardless of what error code we return + if (preply_buf) { + if (notify_private.reply_bl.length()) { + *preply_buf = (char*)malloc(notify_private.reply_bl.length()); + memcpy(*preply_buf, notify_private.reply_bl.c_str(), + notify_private.reply_bl.length()); + } else { + *preply_buf = NULL; } } - ldout(client->cct, 10) << __func__ << " completed notify (linger op " - << linger_op << "), unregistering" << dendl; + if (preply_buf_len) + *preply_buf_len = notify_private.reply_bl.length(); + if (preply_bl) + preply_bl->claim(notify_private.reply_bl); lock->Lock(); objecter->linger_cancel(linger_op); diff --git a/src/osd/Watch.cc b/src/osd/Watch.cc index bccd0f70161..4c21c83c212 100644 --- a/src/osd/Watch.cc +++ b/src/osd/Watch.cc @@ -179,8 +179,16 @@ void Notify::maybe_complete_notify() << watchers.size() << " in progress watchers " << dendl; if (watchers.empty() || timed_out) { + // prepare reply bufferlist bl; ::encode(notify_replies, bl); + list missed; + for (set::iterator p = watchers.begin(); p != watchers.end(); ++p) { + (*p)->send_failed_notify(this); + missed.push_back((*p)->get_entity().num()); + } + ::encode(missed, bl); + bufferlist empty; MWatchNotify *reply(new MWatchNotify(cookie, version, notify_id, CEPH_WATCH_EVENT_NOTIFY_COMPLETE, empty)); @@ -191,9 +199,6 @@ void Notify::maybe_complete_notify() client->send_message(reply); unregister_cb(); - for (set::iterator p = watchers.begin(); p != watchers.end(); ++p) { - (*p)->send_failed_notify(this); - } complete = true; } } diff --git a/src/test/librados/watch_notify.cc b/src/test/librados/watch_notify.cc index aade49c5253..7dfabe399d3 100644 --- a/src/test/librados/watch_notify.cc +++ b/src/test/librados/watch_notify.cc @@ -8,6 +8,7 @@ #include #include "gtest/gtest.h" #include "include/encoding.h" +#include using namespace librados; @@ -258,7 +259,7 @@ TEST_F(LibRadosWatchNotify, Watch2Timeout) { ASSERT_EQ(-ENOTCONN, rados_watch_check(ioctx, handle)); // a subsequent notify should not reach us - char *reply_buf; + char *reply_buf = 0; size_t reply_buf_len; ASSERT_EQ(0, rados_notify2(ioctx, notify_oid, "notify", 6, 0, @@ -267,12 +268,16 @@ TEST_F(LibRadosWatchNotify, Watch2Timeout) { bufferlist reply; reply.append(reply_buf, reply_buf_len); std::multimap reply_map; + std::multiset missed_map; bufferlist::iterator reply_p = reply.begin(); ::decode(reply_map, reply_p); + ::decode(missed_map, reply_p); ASSERT_EQ(0u, reply_map.size()); + ASSERT_EQ(0u, missed_map.size()); } ASSERT_EQ(0u, notify_cookies.size()); ASSERT_EQ(-ENOTCONN, rados_watch_check(ioctx, handle)); + rados_buffer_free(reply_buf); // re-watch rados_unwatch2(ioctx, handle); @@ -292,9 +297,12 @@ TEST_F(LibRadosWatchNotify, Watch2Timeout) { bufferlist reply; reply.append(reply_buf, reply_buf_len); std::multimap reply_map; + std::multiset missed_map; bufferlist::iterator reply_p = reply.begin(); ::decode(reply_map, reply_p); + ::decode(missed_map, reply_p); ASSERT_EQ(1u, reply_map.size()); + ASSERT_EQ(0, missed_map.size()); ASSERT_EQ(1, notify_cookies.count(handle)); ASSERT_EQ(5, reply_map.begin()->second.length()); ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5)); @@ -321,7 +329,7 @@ TEST_F(LibRadosWatchNotify, WatchNotify2) { watch_notify2_test_failcb, watch_notify2_test_errcb, NULL)); ASSERT_TRUE(rados_watch_check(ioctx, handle) > 0); - char *reply_buf; + char *reply_buf = 0; size_t reply_buf_len; ASSERT_EQ(0, rados_notify2(ioctx, notify_oid, "notify", 6, 0, @@ -329,14 +337,27 @@ TEST_F(LibRadosWatchNotify, WatchNotify2) { bufferlist reply; reply.append(reply_buf, reply_buf_len); std::multimap reply_map; + std::multiset missed_map; bufferlist::iterator reply_p = reply.begin(); ::decode(reply_map, reply_p); + ::decode(missed_map, reply_p); ASSERT_EQ(1u, reply_map.size()); + ASSERT_EQ(0, missed_map.size()); ASSERT_EQ(1u, notify_cookies.size()); ASSERT_EQ(1, notify_cookies.count(handle)); ASSERT_EQ(5, reply_map.begin()->second.length()); ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5)); ASSERT_TRUE(rados_watch_check(ioctx, handle) > 0); + rados_buffer_free(reply_buf); + + // try it on a non-existent object ... our buffer pointers + // should get zeroed. + ASSERT_EQ(-ENOENT, rados_notify2(ioctx, "doesnotexist", + "notify", 6, 0, + &reply_buf, &reply_buf_len)); + ASSERT_EQ(NULL, reply_buf); + ASSERT_EQ(0, reply_buf_len); + rados_unwatch2(ioctx, handle); } @@ -360,12 +381,15 @@ TEST_P(LibRadosWatchNotifyPP, WatchNotify2) { ASSERT_EQ(0, ioctx.notify(notify_oid, bl2, 0, &bl_reply)); bufferlist::iterator p = bl_reply.begin(); std::multimap reply_map; + std::multiset missed_map; ::decode(reply_map, p); + ::decode(missed_map, p); ASSERT_EQ(1u, notify_cookies.size()); ASSERT_EQ(1, notify_cookies.count(handle)); ASSERT_EQ(1u, reply_map.size()); ASSERT_EQ(5, reply_map.begin()->second.length()); ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5)); + ASSERT_EQ(0, missed_map.size()); ASSERT_TRUE(ioctx.watch_check(handle) > 0); ioctx.unwatch(handle); } @@ -393,7 +417,7 @@ TEST_F(LibRadosWatchNotify, WatchNotify2Multi) { ASSERT_TRUE(rados_watch_check(ioctx, handle1) > 0); ASSERT_TRUE(rados_watch_check(ioctx, handle2) > 0); ASSERT_NE(handle1, handle2); - char *reply_buf; + char *reply_buf = 0; size_t reply_buf_len; ASSERT_EQ(0, rados_notify2(ioctx, notify_oid, "notify", 6, 0, @@ -401,16 +425,20 @@ TEST_F(LibRadosWatchNotify, WatchNotify2Multi) { bufferlist reply; reply.append(reply_buf, reply_buf_len); std::multimap reply_map; + std::multiset missed_map; bufferlist::iterator reply_p = reply.begin(); ::decode(reply_map, reply_p); + ::decode(missed_map, reply_p); ASSERT_EQ(2u, reply_map.size()); ASSERT_EQ(5, reply_map.begin()->second.length()); + ASSERT_EQ(0, missed_map.size()); ASSERT_EQ(2u, notify_cookies.size()); ASSERT_EQ(1, notify_cookies.count(handle1)); ASSERT_EQ(1, notify_cookies.count(handle2)); ASSERT_EQ(0, strncmp("reply", reply_map.begin()->second.c_str(), 5)); ASSERT_TRUE(rados_watch_check(ioctx, handle1) > 0); ASSERT_TRUE(rados_watch_check(ioctx, handle2) > 0); + rados_buffer_free(reply_buf); rados_unwatch2(ioctx, handle1); rados_unwatch2(ioctx, handle2); } @@ -433,7 +461,7 @@ TEST_F(LibRadosWatchNotify, WatchNotify2Timeout) { watch_notify2_test_failcb, watch_notify2_test_errcb, NULL)); ASSERT_TRUE(rados_watch_check(ioctx, handle) > 0); - char *reply_buf; + char *reply_buf = 0; size_t reply_buf_len; ASSERT_EQ(-ETIMEDOUT, rados_notify2(ioctx, notify_oid, "notify", 6, 1000, // 1s @@ -443,6 +471,18 @@ TEST_F(LibRadosWatchNotify, WatchNotify2Timeout) { while (!notify_failed && --wait) sleep(1); ASSERT_TRUE(notify_failed); + { + bufferlist reply; + reply.append(reply_buf, reply_buf_len); + std::multimap reply_map; + std::multiset missed_map; + bufferlist::iterator reply_p = reply.begin(); + ::decode(reply_map, reply_p); + ::decode(missed_map, reply_p); + ASSERT_EQ(0, reply_map.size()); + ASSERT_EQ(1, missed_map.size()); + } + rados_buffer_free(reply_buf); // we should get the next notify, though! notify_failed = false; diff --git a/src/tools/rados/rados.cc b/src/tools/rados/rados.cc index af29f51c083..2138aadb8e6 100644 --- a/src/tools/rados/rados.cc +++ b/src/tools/rados/rados.cc @@ -2389,11 +2389,18 @@ static int rados_tool_common(const std::map < std::string, std::string > &opts, ::decode(rm, p); for (multimap::iterator p = rm.begin(); p != rm.end(); ++p) { - cout << "client." << p->first + cout << "reply client." << p->first << " : " << p->second.length() << " bytes" << std::endl; if (p->second.length()) p->second.hexdump(cout); } + if (!p.end()) { + list missed; + ::decode(missed, p); + for (list::iterator p = missed.begin(); p != missed.end(); ++p) { + cout << "timeout client." << *p << std::endl; + } + } } } else if (strcmp(nargs[0], "set-alloc-hint") == 0) { if (!pool_name || nargs.size() < 4)