struct CB_DoWatchNotify {
Objecter *objecter;
boost::intrusive_ptr<Objecter::LingerOp> info;
- boost::intrusive_ptr<MWatchNotify> msg;
+ cref_t<MWatchNotify> msg;
CB_DoWatchNotify(Objecter *o,
boost::intrusive_ptr<Objecter::LingerOp> i,
- MWatchNotify *m)
- : objecter(o), info(std::move(i)), msg(m) {
+ cref_t<MWatchNotify> m)
+ : objecter(o), info(std::move(i)), msg(std::move(m)) {
info->_queued_async();
}
void operator()() {
}
};
-void Objecter::handle_watch_notify(MWatchNotify *m)
+void Objecter::handle_watch_notify(cref_t<MWatchNotify> m)
{
shared_lock l(rwlock);
if (!initialized) {
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<LingerOp> info,
- boost::intrusive_ptr<MWatchNotify> m)
+ cref_t<MWatchNotify> m)
{
ldout(cct, 10) << __func__ << " " << *m << dendl;
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;
}
{
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<MOSDOpReply>(m).get());
+ case CEPH_MSG_OSD_OPREPLY: {
+ cref_t<MOSDOpReply> msg = ref_cast<MOSDOpReply>(m);
+ auto priv = msg->get_connection()->get_priv();
+ auto s = static_cast<OSDSession*>(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<MOSDBackoff>(m).get());
+ case CEPH_MSG_OSD_BACKOFF: {
+ cref_t<MOSDBackoff> msg = ref_cast<MOSDBackoff>(m);
+ auto priv = msg->get_connection()->get_priv();
+ auto s = static_cast<OSDSession*>(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<MWatchNotify>(m).get());
+ case CEPH_MSG_WATCH_NOTIFY: {
+ cref_t<MWatchNotify> msg = ref_cast<MWatchNotify>(m);
+ auto priv = msg->get_connection()->get_priv();
+ auto s = static_cast<OSDSession*>(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<MCommandReply>(m).get());
+ cref_t<MCommandReply> msg = ref_cast<MCommandReply>(m);
+ auto priv = msg->get_connection()->get_priv();
+ auto s = static_cast<OSDSession*>(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();
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});
return 1;
}
-bs::error_code Objecter::handle_osd_op_reply2(Op *op, vector<OSDOp> &out_ops) {
+bs::error_code Objecter::handle_osd_op_reply2(Op *op, const vector<OSDOp>& out_ops) {
ceph_assert(op->ops.size() == op->out_bl.size());
ceph_assert(op->ops.size() == op->out_rval.size());
}
-/* 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<MOSDOpReply> m)
{
ldout(cct, 10) << "in handle_osd_op_reply" << dendl;
shunique_lock sul(rwlock, ceph::acquire_shared);
if (!initialized) {
- m->put();
return;
}
auto s = static_cast<OSDSession*>(priv.get());
if (!s || s->con != con) {
ldout(cct, 7) << __func__ << " no session on con " << con << dendl;
- m->put();
return;
}
" onnvram" : " ack"))
<< " ... stray" << dendl;
sl.unlock();
- m->put();
return;
}
sl.unlock();
_op_submit(op, sul, NULL);
- m->put();
return;
}
<< " from " << m->get_source_inst()
<< "; last attempt " << (op->attempts - 1) << " sent to "
<< op->session->con->get_peer_addr() << dendl;
- m->put();
sl.unlock();
return;
}
CEPH_OSD_FLAG_IGNORE_CACHE |
CEPH_OSD_FLAG_IGNORE_OVERLAY);
_op_submit(op, sul, NULL);
- m->put();
return;
}
CEPH_OSD_FLAG_LOCALIZE_READS);
op->target.pgid = pg_t();
_op_submit(op, sul, NULL);
- m->put();
return;
}
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<OSDOp> 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
if (completion_lock.mutex()) {
completion_lock.unlock();
}
-
- m->put();
}
-void Objecter::handle_osd_backoff(MOSDBackoff *m)
+void Objecter::handle_osd_backoff(cref_t<MOSDBackoff> m)
{
ldout(cct, 10) << __func__ << " " << *m << dendl;
shunique_lock sul(rwlock, ceph::acquire_shared);
if (!initialized) {
- m->put();
return;
}
auto s = static_cast<OSDSession*>(priv.get());
if (!s || s->con != con) {
ldout(cct, 7) << __func__ << " no session on con " << con << dendl;
- m->put();
return;
}
sul.unlock();
sl.unlock();
- m->put();
put_session(s);
}
// commands
-void Objecter::handle_command_reply(MCommandReply *m)
+void Objecter::handle_command_reply(cref_t<MCommandReply> m)
{
unique_lock wl(rwlock);
if (!initialized) {
- m->put();
return;
}
auto s = static_cast<OSDSession*>(priv.get());
if (!s || s->con != con) {
ldout(cct, 7) << __func__ << " no session on con " << con << dendl;
- m->put();
return;
}
if (p == s->command_ops.end()) {
ldout(cct, 10) << "handle_command_reply tid " << m->get_tid()
<< " not found" << dendl;
- m->put();
sl.unlock();
return;
}
<< " got reply from wrong connection "
<< m->get_connection() << " " << m->get_source_inst()
<< dendl;
- m->put();
sl.unlock();
return;
}
// we get an updated osdmap and the PG is found to have moved.
_maybe_request_map();
_send_command(c);
- m->put();
sl.unlock();
return;
}
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)
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<std::chrono::seconds>("rados_mon_op_timeout");
osd_timeout = cct->_conf.get_val<std::chrono::seconds>("rados_osd_op_timeout");
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<MCommandReply> m);
// -- lingering ops --
// lockdep (using std::sharedMutex) because lockdep doesn't know
// that.
std::shared_mutex lock;
+ boost::asio::strand<boost::asio::io_context::executor_type> strand;
int incarnation;
ConnectionRef con;
int num_locks;
std::unique_ptr<std::mutex[]> 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]) {}
std::map<ceph_tid_t,PoolOp*> pool_ops;
std::atomic<unsigned> 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
[[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<OSDOp> &out_ops);
- void handle_osd_backoff(class MOSDBackoff *m);
- void handle_watch_notify(class MWatchNotify *m);
+ void handle_osd_op_reply(cref_t<MOSDOpReply> m);
+ boost::system::error_code handle_osd_op_reply2(Op *op, const std::vector<OSDOp> &out_ops);
+ void handle_osd_backoff(cref_t<MOSDBackoff> m);
+ void handle_watch_notify(cref_t<MWatchNotify> m);
void handle_osd_map(class MOSDMap *m);
void wait_for_osd_map(epoch_t e=0);
public:
void _do_watch_notify(boost::intrusive_ptr<LingerOp> info,
- boost::intrusive_ptr<MWatchNotify> m);
+ cref_t<MWatchNotify> m);
/**
* set up initial ops in the op std::vector, and allocate a final op slot.
projects / ceph-ci.git / commitdiff