From: Sage Weil Date: Wed, 1 Feb 2017 22:35:36 +0000 (-0500) Subject: osdc/Objecter: respect backoffs X-Git-Tag: v12.0.1~441^2~3 X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=b8e1c05ed9a009ac33cd3e0b368935b6facb548a;p=ceph.git osdc/Objecter: respect backoffs Signed-off-by: Sage Weil --- diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc index 7e87f4ab05d..f592811b6fc 100644 --- a/src/osdc/Objecter.cc +++ b/src/osdc/Objecter.cc @@ -24,6 +24,7 @@ #include "messages/MPing.h" #include "messages/MOSDOp.h" #include "messages/MOSDOpReply.h" +#include "messages/MOSDBackoff.h" #include "messages/MOSDMap.h" #include "messages/MPoolOp.h" @@ -969,6 +970,10 @@ bool Objecter::ms_dispatch(Message *m) handle_osd_op_reply(static_cast(m)); return true; + case CEPH_MSG_OSD_BACKOFF: + handle_osd_backoff(static_cast(m)); + return true; + case CEPH_MSG_WATCH_NOTIFY: handle_watch_notify(static_cast(m)); m->put(); @@ -1948,6 +1953,10 @@ void Objecter::_kick_requests(OSDSession *session, { // rwlock is locked unique + // clear backoffs + session->backoffs.clear(); + session->backoffs_by_id.clear(); + // resend ops map resend; // resend in tid order for (map::iterator p = session->ops.begin(); @@ -3061,6 +3070,26 @@ void Objecter::_send_op(Op *op, MOSDOp *m) // rwlock is locked // op->session->lock is locked + // backoff? + hobject_t hoid = op->target.get_hobj(); + auto q = op->session->backoffs.lower_bound(hoid); + if (q != op->session->backoffs.begin()) { + --q; + if (cmp_bitwise(hoid, q->second.end) >= 0) { + ++q; + } + } + if (q != op->session->backoffs.end()) { + ldout(cct, 20) << __func__ << " ? " << q->first << " [" << q->second.begin + << "," << q->second.end << ")" << dendl; + int r = cmp_bitwise(hoid, q->second.begin); + if (r == 0 || (r > 0 && cmp_bitwise(hoid, q->second.end) < 0)) { + ldout(cct, 10) << __func__ << " backoff on " << hoid << ", queuing " + << op << " tid " << op->tid << dendl; + return; + } + } + if (!m) { assert(op->tid > 0); m = _prepare_osd_op(op); @@ -3357,6 +3386,90 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m) put_session(s); } +void Objecter::handle_osd_backoff(MOSDBackoff *m) +{ + ldout(cct, 10) << __func__ << " " << *m << dendl; + shunique_lock sul(rwlock, ceph::acquire_shared); + if (!initialized.read()) { + m->put(); + return; + } + + ConnectionRef con = m->get_connection(); + OSDSession *s = static_cast(con->get_priv()); + if (!s || s->con != con) { + ldout(cct, 7) << __func__ << " no session on con " << con << dendl; + m->put(); + return; + } + + get_session(s); + s->put(); // from get_priv() above + + OSDSession::unique_lock sl(s->lock); + + switch (m->op) { + case CEPH_OSD_BACKOFF_OP_BLOCK: + { + // register + OSDBackoff& b = s->backoffs[m->begin]; + s->backoffs_by_id.insert(make_pair(m->id, &b)); + b.id = m->id; + b.begin = m->begin; + b.end = m->end; + + // ack + Message *r = new MOSDBackoff(CEPH_OSD_BACKOFF_OP_ACK_BLOCK, + m->id, m->begin, m->end, + osdmap->get_epoch()); + // this priority must match the MOSDOps from _prepare_osd_op + r->set_priority(cct->_conf->osd_client_op_priority); + con->send_message(r); + } + break; + + case CEPH_OSD_BACKOFF_OP_UNBLOCK: + { + auto p = s->backoffs_by_id.find(m->id); + while (p != s->backoffs_by_id.end() && + p->second->id == m->id) { + OSDBackoff *b = p->second; + if (b->begin != m->begin && + b->end != m->end) { + lderr(cct) << __func__ << " got id " << m->id << " unblock on [" + << m->begin << "," << m->end << ") but backoff is [" + << b->begin << "," << b->end << ")" << dendl; + // hrmpf, unblock it anyway. + } + ldout(cct, 10) << __func__ << " unblock backoff " << b->id + << " [" << b->begin << "," << b->end + << ")" << dendl; + s->backoffs.erase(b->begin); + p = s->backoffs_by_id.erase(p); + + // check for any ops to resend + for (auto& q : s->ops) { + int r = q.second->target.contained_by(m->begin, m->end); + ldout(cct, 20) << __func__ << " contained_by " << r << " on " + << q.second->target.get_hobj() << dendl; + if (r) { + _send_op(q.second); + } + } + } + } + break; + + default: + ldout(cct, 10) << __func__ << " unrecognized op " << (int)m->op << dendl; + } + + sul.unlock(); + sl.unlock(); + + m->put(); + put_session(s); +} uint32_t Objecter::list_nobjects_seek(NListContext *list_context, uint32_t pos) diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h index 0f34458e7cc..8682a61d7ad 100644 --- a/src/osdc/Objecter.h +++ b/src/osdc/Objecter.h @@ -1225,6 +1225,21 @@ public: op_target_t() = default; + hobject_t get_hobj() { + return hobject_t(target_oid, + target_oloc.key, + CEPH_NOSNAP, + target_oloc.hash >= 0 ? target_oloc.hash : pgid.ps(), + target_oloc.pool, + target_oloc.nspace); + } + + bool contained_by(const hobject_t& begin, const hobject_t& end) { + hobject_t h = get_hobj(); + int r = cmp_bitwise(h, begin); + return r == 0 || (r > 0 && cmp_bitwise(h, end) < 0); + } + void dump(Formatter *f) const; }; @@ -1774,6 +1789,11 @@ public: }; // -- osd sessions -- + struct OSDBackoff { + uint64_t id; + hobject_t begin, end; + }; + struct OSDSession : public RefCountedObject { boost::shared_mutex lock; using lock_guard = std::lock_guard; @@ -1786,6 +1806,10 @@ public: map linger_ops; map command_ops; + // backoffs + map backoffs; + multimap backoffs_by_id; + int osd; int incarnation; ConnectionRef con; @@ -2074,6 +2098,7 @@ private: } void handle_osd_op_reply(class MOSDOpReply *m); + void handle_osd_backoff(class MOSDBackoff *m); void handle_watch_notify(class MWatchNotify *m); void handle_osd_map(class MOSDMap *m); void wait_for_osd_map();