From: Patrick Donnelly Date: Wed, 11 Mar 2026 06:52:15 +0000 (+0530) Subject: osdc: dispatch messenger handlers to per-session ASIO strands X-Git-Tag: testing/wip-pdonnell-testing-20260520.191703-main^2~5 X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=2a5aa2774325fa8f66d6ee2032cfb5ea1b2de47c;p=ceph-ci.git osdc: dispatch messenger handlers to per-session ASIO strands Currently, RADOS messenger handler functions (such as handle_osd_op_reply, handle_osd_backoff, handle_watch_notify, and handle_command_reply) execute directly on the messenger dispatcher thread, which can block further message dispatch. This patch mitigates blocking by offloading the processing of these messages to an I/O thread pool context. Specifically: - Introduces a boost::asio::strand to OSDSession and updates the constructor to take an executor. - Updates ms_dispatch2 to dispatch incoming OSD messages to the respective OSDSession's strand if the session exists, ensuring sequential execution per session without blocking the main dispatcher. - Refactors message passing across Objecter handlers to use cref_t (smart pointer references) instead of raw pointers. This architectural shift improves memory safety and eliminates scattered manual reference counting (e.g., removing m->put() calls). - Adds get_ops() to MOSDOpReply to expose the operations vector via a const reference , avoiding the need to consume and swap out the operations destructively. Simple test with `main` against the LRC: [pdonnell@sockeni03 build]$ env CEPH_ARGS='--admin_socket=foo.socket --conf ceph.conf --keyring=keyring --id pdonnell-test' /usr/bin/rados -p pdonnell-test bench 30 write --concurrent-ios=128 hints = 1 Maintaining 128 concurrent writes of 4194304 bytes to objects of size 4194304 for up to 30 seconds or 0 objects Object prefix: benchmark_data_sockeni03.front.sepia.ceph.co_1548405 sec Cur ops started finished avg MB/s cur MB/s last lat(s) avg lat(s) 0 0 0 0 0 0 - 0 1 127 789 662 2647.46 2648 0.151097 0.162754 2 127 1532 1405 2809.52 2972 0.156297 0.167421 3 127 2262 2135 2846.19 2920 0.210822 0.169546 4 127 3057 2930 2929.46 3180 0.156612 0.16757 5 127 3836 3709 2966.67 3116 0.168111 0.166617 6 127 4593 4466 2976.77 3028 0.166451 0.167173 7 127 5337 5210 2976.6 2976 0.209467 0.167153 8 127 6122 5995 2996.97 3140 0.162973 0.167285 9 127 6875 6748 2998.59 3012 0.192893 0.167413 10 127 7612 7485 2993.49 2948 0.183582 0.167984 11 127 8409 8282 3011.13 3188 0.158497 0.167468 12 127 9172 9045 3014.5 3052 0.161077 0.167487 13 127 9972 9845 3028.73 3200 0.16204 0.166873 14 127 10723 10596 3026.93 3004 0.156975 0.167192 15 127 11517 11390 3036.84 3176 0.165272 0.166697 16 127 12235 12108 3026.51 2872 0.17727 0.167362 17 127 13031 12904 3035.75 3184 0.167001 0.166974 18 127 13834 13707 3045.51 3212 0.161747 0.166566 19 127 14588 14461 3043.94 3016 0.185274 0.166648 2026-05-12T16:55:23.431497+0000 min lat: 0.140888 max lat: 0.254646 avg lat: 0.166449 sec Cur ops started finished avg MB/s cur MB/s last lat(s) avg lat(s) 20 127 15374 15247 3048.92 3144 0.173577 0.166449 21 127 16170 16043 3055.33 3184 0.156316 0.166276 22 127 16944 16817 3057.16 3096 0.16904 0.166148 23 127 17738 17611 3062.3 3176 0.155423 0.166018 24 127 18524 18397 3065.69 3144 0.173122 0.165771 25 127 19322 19195 3070.72 3192 0.16625 0.165593 26 127 20128 20001 3076.6 3224 0.16389 0.165339 27 127 20919 20792 3079.82 3164 0.161066 0.165208 28 127 21657 21530 3075.24 2952 0.192096 0.165479 29 127 22454 22327 3079.11 3188 0.159583 0.165332 30 25 23209 23184 3090.72 3428 0.0253493 0.165156 Total time run: 30.0432 Total writes made: 23209 Write size: 4194304 Object size: 4194304 Bandwidth (MB/sec): 3090.08 Stddev Bandwidth: 144.556 Max bandwidth (MB/sec): 3428 Min bandwidth (MB/sec): 2648 Average IOPS: 772 Stddev IOPS: 36.139 Max IOPS: 857 Min IOPS: 662 Average Latency(s): 0.165018 Stddev Latency(s): 0.0152955 Max latency(s): 0.254646 Min latency(s): 0.0199798 Cleaning up (deleting benchmark objects) Removed 23209 objects Clean up completed and total clean up time :0.587544 with this branch: [pdonnell@sockeni03 build]$ env CEPH_ARGS='--admin_socket=foo.socket --conf ceph.conf --keyring=keyring --id pdonnell-test' ./bin/rados -p pdonnell-test bench 30 write --concurrent-ios=128 hints = 1 Maintaining 128 concurrent writes of 4194304 bytes to objects of size 4194304 for up to 30 seconds or 0 objects Object prefix: benchmark_data_sockeni03.front.sepia.ceph.co_1548463 sec Cur ops started finished avg MB/s cur MB/s last lat(s) avg lat(s) 0 0 0 0 0 0 - 0 1 127 959 832 3327.19 3328 0.12148 0.132731 2 127 1854 1727 3453.26 3580 0.206007 0.132826 3 127 2815 2688 3583.26 3844 0.131366 0.135925 4 127 3669 3542 3541.28 3416 0.131225 0.139544 5 127 4632 4505 3603.27 3852 0.128576 0.138044 6 127 5604 5477 3650.59 3888 0.123859 0.136985 7 127 6576 6449 3684.4 3888 0.129267 0.136138 8 127 7543 7416 3707.21 3868 0.126072 0.135687 9 127 8484 8357 3713.44 3764 0.158376 0.13528 10 127 9371 9244 3696.84 3548 0.128821 0.136534 11 127 10338 10211 3712.33 3868 0.1325 0.136114 12 127 11170 11043 3680.25 3328 0.275972 0.136479 13 127 12135 12008 3694.02 3860 0.134695 0.137055 14 127 13100 12973 3705.82 3860 0.133235 0.136726 15 127 14073 13946 3718.18 3892 0.13525 0.13638 16 127 15052 14925 3730.5 3916 0.129054 0.136028 17 127 15939 15812 3719.72 3548 0.159312 0.136467 18 127 16926 16799 3732.36 3948 0.128867 0.136087 19 127 17903 17776 3741.56 3908 0.129389 0.135815 2026-05-12T17:02:01.598292+0000 min lat: 0.119566 max lat: 0.277312 avg lat: 0.136375 sec Cur ops started finished avg MB/s cur MB/s last lat(s) avg lat(s) 20 127 18769 18642 3727.65 3464 0.13077 0.136375 21 127 19727 19600 3732.56 3832 0.130508 0.136253 22 127 20686 20559 3737.23 3836 0.140792 0.136089 23 127 21668 21541 3745.49 3928 0.127547 0.135867 24 127 22633 22506 3750.23 3860 0.129095 0.135714 25 127 23630 23503 3759.71 3988 0.130776 0.135397 26 127 24616 24489 3766.77 3944 0.132067 0.13517 27 127 25588 25461 3771.23 3888 0.130989 0.135056 28 127 26573 26446 3777.23 3940 0.127752 0.134855 29 127 27550 27423 3781.71 3908 0.128681 0.13473 30 16 28534 28518 3801.62 4380 0.0313388 0.134301 Total time run: 30.0252 Total writes made: 28534 Write size: 4194304 Object size: 4194304 Bandwidth (MB/sec): 3801.34 Stddev Bandwidth: 221.928 Max bandwidth (MB/sec): 4380 Min bandwidth (MB/sec): 3328 Average IOPS: 950 Stddev IOPS: 55.482 Max IOPS: 1095 Min IOPS: 832 Average Latency(s): 0.134239 Stddev Latency(s): 0.0177345 Max latency(s): 0.277312 Min latency(s): 0.0185666 Cleaning up (deleting benchmark objects) Removed 28534 objects Clean up completed and total clean up time :0.450369 This isn't exhaustively studied but it does seem to help even for trivial completions. I expect the difference will be much more significant when the completions cross multiple component boundaries e.g. objecter->filer->Client->nfs-ganesha. Signed-off-by: Patrick Donnelly --- diff --git a/src/messages/MOSDOpReply.h b/src/messages/MOSDOpReply.h index 7b7ad9608dc..3b260431f1e 100644 --- a/src/messages/MOSDOpReply.h +++ b/src/messages/MOSDOpReply.h @@ -106,6 +106,9 @@ public: ops[i].outdata = std::move(o[i].outdata); } } + auto const& get_ops() const { + return ops; + } void claim_ops(std::vector& o) { o.swap(ops); bdata_encode = false; diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc index 18d76a5cd48..3c61fb555be 100644 --- a/src/osdc/Objecter.cc +++ b/src/osdc/Objecter.cc @@ -936,11 +936,11 @@ void Objecter::_linger_submit(LingerOp *info, struct CB_DoWatchNotify { Objecter *objecter; boost::intrusive_ptr info; - boost::intrusive_ptr msg; + cref_t msg; CB_DoWatchNotify(Objecter *o, boost::intrusive_ptr i, - MWatchNotify *m) - : objecter(o), info(std::move(i)), msg(m) { + cref_t m) + : objecter(o), info(std::move(i)), msg(std::move(m)) { info->_queued_async(); } void operator()() { @@ -948,7 +948,7 @@ struct CB_DoWatchNotify { } }; -void Objecter::handle_watch_notify(MWatchNotify *m) +void Objecter::handle_watch_notify(cref_t m) { shared_lock l(rwlock); if (!initialized) { @@ -981,18 +981,18 @@ void Objecter::handle_watch_notify(MWatchNotify *m) asio::defer(service.get_executor(), asio::append(std::move(info->on_notify_finish), osdcode(m->return_code), - std::move(m->get_data()))); + bufferlist(m->get_data()))); // if we race with reconnect we might get a second notify; only // notify the caller once! info->on_notify_finish = nullptr; } } else { - asio::defer(finish_strand, CB_DoWatchNotify(this, std::move(info), m)); + asio::defer(finish_strand, CB_DoWatchNotify(this, std::move(info), std::move(m))); } } void Objecter::_do_watch_notify(boost::intrusive_ptr info, - boost::intrusive_ptr m) + cref_t m) { ldout(cct, 10) << __func__ << " " << *m << dendl; @@ -1013,7 +1013,7 @@ void Objecter::_do_watch_notify(boost::intrusive_ptr info, switch (m->opcode) { case CEPH_WATCH_EVENT_NOTIFY: - info->handle({}, m->notify_id, m->cookie, m->notifier_gid, std::move(m->bl)); + info->handle({}, m->notify_id, m->cookie, m->notifier_gid, bufferlist{m->bl}); break; } @@ -1025,26 +1025,60 @@ Dispatcher::dispatch_result_t Objecter::ms_dispatch2(const MessageRef &m) { ldout(cct, 10) << __func__ << " " << cct << " " << *m << dendl; switch (m->get_type()) { - // these we exlusively handle - case CEPH_MSG_OSD_OPREPLY: - m->get(); /* ref to be consumed */ - handle_osd_op_reply(ref_cast(m).get()); + case CEPH_MSG_OSD_OPREPLY: { + cref_t msg = ref_cast(m); + auto priv = msg->get_connection()->get_priv(); + auto s = static_cast(priv.get()); + if (s) { + boost::asio::dispatch(s->strand, [this, msg = std::move(msg)]() { + handle_osd_op_reply(std::move(msg)); + }); + } else { + handle_osd_op_reply(std::move(msg)); + } return Dispatcher::HANDLED(); + } - case CEPH_MSG_OSD_BACKOFF: - m->get(); /* ref to be consumed */ - handle_osd_backoff(ref_cast(m).get()); + case CEPH_MSG_OSD_BACKOFF: { + cref_t msg = ref_cast(m); + auto priv = msg->get_connection()->get_priv(); + auto s = static_cast(priv.get()); + if (s) { + boost::asio::dispatch(s->strand, [this, msg = std::move(msg)]() { + handle_osd_backoff(std::move(msg)); + }); + } else { + handle_osd_backoff(std::move(msg)); + } return Dispatcher::HANDLED(); + } - case CEPH_MSG_WATCH_NOTIFY: - /* ref not consumed! */ - handle_watch_notify(ref_cast(m).get()); + case CEPH_MSG_WATCH_NOTIFY: { + cref_t msg = ref_cast(m); + auto priv = msg->get_connection()->get_priv(); + auto s = static_cast(priv.get()); + if (s) { + boost::asio::dispatch(s->strand, [this, msg = std::move(msg)]() { + handle_watch_notify(std::move(msg)); + }); + } else { + handle_watch_notify(std::move(msg)); + } return Dispatcher::HANDLED(); + } case MSG_COMMAND_REPLY: if (m->get_source().type() == CEPH_ENTITY_TYPE_OSD) { - m->get(); /* ref to be consumed */ - handle_command_reply(ref_cast(m).get()); + cref_t msg = ref_cast(m); + auto priv = msg->get_connection()->get_priv(); + auto s = static_cast(priv.get()); + if (s) { + boost::asio::dispatch(s->strand, [this, msg = std::move(msg)]() { + handle_command_reply(std::move(msg)); + }); + } else { + handle_command_reply(std::move(msg)); + } return Dispatcher::HANDLED(); } else { return Dispatcher::UNHANDLED(); @@ -1917,7 +1951,7 @@ int Objecter::_get_session(int osd, OSDSession **session, if (!sul.owns_lock()) { return -EAGAIN; } - auto s = new OSDSession(cct, osd); + auto s = new OSDSession(cct, osd, service.get_executor()); osd_sessions[osd] = s; s->con = messenger->connect_to_osd(osdmap->get_addrs(osd)); s->con->set_priv(RefCountedPtr{s}); @@ -3574,7 +3608,7 @@ int Objecter::take_linger_budget(LingerOp *info) return 1; } -bs::error_code Objecter::handle_osd_op_reply2(Op *op, vector &out_ops) { +bs::error_code Objecter::handle_osd_op_reply2(Op *op, const vector& out_ops) { ceph_assert(op->ops.size() == op->out_bl.size()); ceph_assert(op->ops.size() == op->out_rval.size()); @@ -3650,8 +3684,7 @@ bs::error_code Objecter::handle_osd_op_reply2(Op *op, vector &out_ops) { } -/* This function DOES put the passed message before returning */ -void Objecter::handle_osd_op_reply(MOSDOpReply *m) +void Objecter::handle_osd_op_reply(cref_t m) { ldout(cct, 10) << "in handle_osd_op_reply" << dendl; @@ -3660,7 +3693,6 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) shunique_lock sul(rwlock, ceph::acquire_shared); if (!initialized) { - m->put(); return; } @@ -3669,7 +3701,6 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) auto s = static_cast(priv.get()); if (!s || s->con != con) { ldout(cct, 7) << __func__ << " no session on con " << con << dendl; - m->put(); return; } @@ -3682,7 +3713,6 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) " onnvram" : " ack")) << " ... stray" << dendl; sl.unlock(); - m->put(); return; } @@ -3706,7 +3736,6 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) sl.unlock(); _op_submit(op, sul, NULL); - m->put(); return; } @@ -3717,7 +3746,6 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) << " from " << m->get_source_inst() << "; last attempt " << (op->attempts - 1) << " sent to " << op->session->con->get_peer_addr() << dendl; - m->put(); sl.unlock(); return; } @@ -3747,7 +3775,6 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) CEPH_OSD_FLAG_IGNORE_CACHE | CEPH_OSD_FLAG_IGNORE_OVERLAY); _op_submit(op, sul, NULL); - m->put(); return; } @@ -3772,7 +3799,6 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) CEPH_OSD_FLAG_LOCALIZE_READS); op->target.pgid = pg_t(); _op_submit(op, sul, NULL); - m->put(); return; } @@ -3806,14 +3832,13 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) bl.begin().copy(bl.length(), t.c_str()); op->outbl->substr_of(t, 0, bl.length()); } else { - m->claim_data(*op->outbl); + *op->outbl = m->get_data(); } op->outbl = 0; } // per-op result demuxing - vector out_ops; - m->claim_ops(out_ops); + auto& out_ops = m->get_ops(); if (out_ops.size() != op->ops.size()) ldout(cct, 0) << "WARNING: tid " << op->tid << " reply ops " << out_ops @@ -3861,16 +3886,13 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) if (completion_lock.mutex()) { completion_lock.unlock(); } - - m->put(); } -void Objecter::handle_osd_backoff(MOSDBackoff *m) +void Objecter::handle_osd_backoff(cref_t m) { ldout(cct, 10) << __func__ << " " << *m << dendl; shunique_lock sul(rwlock, ceph::acquire_shared); if (!initialized) { - m->put(); return; } @@ -3879,7 +3901,6 @@ void Objecter::handle_osd_backoff(MOSDBackoff *m) auto s = static_cast(priv.get()); if (!s || s->con != con) { ldout(cct, 7) << __func__ << " no session on con " << con << dendl; - m->put(); return; } @@ -3960,7 +3981,6 @@ void Objecter::handle_osd_backoff(MOSDBackoff *m) sul.unlock(); sl.unlock(); - m->put(); put_session(s); } @@ -5049,11 +5069,10 @@ void Objecter::blocklist_self(bool set) // commands -void Objecter::handle_command_reply(MCommandReply *m) +void Objecter::handle_command_reply(cref_t m) { unique_lock wl(rwlock); if (!initialized) { - m->put(); return; } @@ -5062,7 +5081,6 @@ void Objecter::handle_command_reply(MCommandReply *m) auto s = static_cast(priv.get()); if (!s || s->con != con) { ldout(cct, 7) << __func__ << " no session on con " << con << dendl; - m->put(); return; } @@ -5071,7 +5089,6 @@ void Objecter::handle_command_reply(MCommandReply *m) if (p == s->command_ops.end()) { ldout(cct, 10) << "handle_command_reply tid " << m->get_tid() << " not found" << dendl; - m->put(); sl.unlock(); return; } @@ -5083,7 +5100,6 @@ void Objecter::handle_command_reply(MCommandReply *m) << " got reply from wrong connection " << m->get_connection() << " " << m->get_source_inst() << dendl; - m->put(); sl.unlock(); return; } @@ -5095,7 +5111,6 @@ void Objecter::handle_command_reply(MCommandReply *m) // we get an updated osdmap and the PG is found to have moved. _maybe_request_map(); _send_command(c); - m->put(); sl.unlock(); return; } @@ -5104,11 +5119,9 @@ void Objecter::handle_command_reply(MCommandReply *m) unique_lock sul(s->lock); _finish_command(c, m->r < 0 ? bs::error_code(-m->r, osd_category()) : - bs::error_code(), std::move(m->rs), - std::move(m->get_data())); + bs::error_code(), std::string(m->rs), + bufferlist(m->get_data())); sul.unlock(); - - m->put(); } Objecter::LingerOp::LingerOp(Objecter *o, uint64_t linger_id) @@ -5305,7 +5318,8 @@ Objecter::OSDSession::~OSDSession() Objecter::Objecter(CephContext *cct, Messenger *m, MonClient *mc, asio::io_context& service) : - Dispatcher(cct), messenger(m), monc(mc), service(service) + Dispatcher(cct), messenger(m), monc(mc), service(service), + homeless_session(new OSDSession(cct, -1, service.get_executor())) { mon_timeout = cct->_conf.get_val("rados_mon_op_timeout"); osd_timeout = cct->_conf.get_val("rados_osd_op_timeout"); diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h index a40ba465fcc..30930bda982 100644 --- a/src/osdc/Objecter.h +++ b/src/osdc/Objecter.h @@ -2360,7 +2360,7 @@ public: boost::system::error_code ec); void _finish_command(CommandOp *c, boost::system::error_code ec, std::string&& rs, ceph::buffer::list&& bl); - void handle_command_reply(MCommandReply *m); + void handle_command_reply(cref_t m); // -- lingering ops -- @@ -2498,14 +2498,15 @@ public: // lockdep (using std::sharedMutex) because lockdep doesn't know // that. std::shared_mutex lock; + boost::asio::strand strand; int incarnation; ConnectionRef con; int num_locks; std::unique_ptr completion_locks; - OSDSession(CephContext *cct, int o) : - osd(o), incarnation(0), con(NULL), + OSDSession(CephContext *cct, int o, boost::asio::io_context::executor_type ex) : + osd(o), strand(ex), incarnation(0), con(NULL), num_locks(cct->_conf->objecter_completion_locks_per_session), completion_locks(new std::mutex[num_locks]) {} @@ -2544,7 +2545,7 @@ public: std::map pool_ops; std::atomic num_homeless_ops{0}; - OSDSession* homeless_session = new OSDSession(cct, -1); + OSDSession* homeless_session; // ops waiting for an osdmap with a new pool or confirmation that @@ -2772,10 +2773,10 @@ private: [[maybe_unused]] auto s = ms_dispatch2(m); } - void handle_osd_op_reply(class MOSDOpReply *m); - boost::system::error_code handle_osd_op_reply2(Op *op, std::vector &out_ops); - void handle_osd_backoff(class MOSDBackoff *m); - void handle_watch_notify(class MWatchNotify *m); + void handle_osd_op_reply(cref_t m); + boost::system::error_code handle_osd_op_reply2(Op *op, const std::vector &out_ops); + void handle_osd_backoff(cref_t m); + void handle_watch_notify(cref_t m); void handle_osd_map(class MOSDMap *m); void wait_for_osd_map(epoch_t e=0); @@ -3291,7 +3292,7 @@ public: public: void _do_watch_notify(boost::intrusive_ptr info, - boost::intrusive_ptr m); + cref_t m); /** * set up initial ops in the op std::vector, and allocate a final op slot.