Objecter: don't throttle resent linger ops

Throttling is intended to stop the caller from submitting too many
requests, not blocking requests that are being resent internally. This
prevents a deadlock when handling an osdmap - previously
handle_osd_map could block when resending linger ops due to the
throttling. This would stop the messenger's dispatch thread from
delivering any subsequest messages, so the throttle budget would never
be replenished.

Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
Reviewed-by: Sage Weil <sage@inktank.com>

2 files changed, 26 insertions, 8 deletions

diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc
index e81bfff43fb..047e1be0f4f 100644
--- a/src/osdc/Objecter.cc
+++ b/src/osdc/Objecter.cc
@@ -246,7 +246,7 @@ void Objecter::shutdown()
   }
 }
 
-void Objecter::send_linger(LingerOp *info)
+void Objecter::send_linger(LingerOp *info, bool first_send)
 {
   if (!info->registering) {
     ldout(cct, 15) << "send_linger " << info->linger_id << dendl;
@@ -264,7 +264,13 @@ void Objecter::send_linger(LingerOp *info)
 	linger_check_for_latest_map(info);
       }
     }
-    op_submit(o, info->session);
+
+    if (first_send) {
+      op_submit(o, info->session);
+    } else {
+      _op_submit(o, info->session);
+    }
+
     OSDSession *s = o->session;
     if (info->session != s) {
       info->session_item.remove_myself();
@@ -342,7 +348,7 @@ tid_t Objecter::linger(const object_t& oid, const object_locator_t& oloc,
 
   logger->set(l_osdc_linger_active, linger_ops.size());
 
-  send_linger(info);
+  send_linger(info, true);
 
   return info->linger_id;
 }
@@ -544,7 +550,7 @@ void Objecter::handle_osd_map(MOSDMap *m)
     LingerOp *op = *p;
     if (op->session) {
       logger->inc(l_osdc_linger_resend);
-      send_linger(op);
+      send_linger(op, false);
     }
   }
 
@@ -749,7 +755,7 @@ void Objecter::kick_requests(OSDSession *session)
   // resend lingers
   for (xlist<LingerOp*>::iterator j = session->linger_ops.begin(); !j.end(); ++j) {
     logger->inc(l_osdc_linger_resend);
-    send_linger(*j);
+    send_linger(*j, false);
   }
 }
 
@@ -863,6 +869,11 @@ tid_t Objecter::op_submit(Op *op, OSDSession *s)
   // take_op_budget() may drop our lock while it blocks.
   take_op_budget(op);
 
+  return _op_submit(op, s);
+}
+
+tid_t Objecter::_op_submit(Op *op, OSDSession *s)
+{
   // pick tid
   tid_t mytid = ++last_tid;
   op->tid = mytid;
@@ -1288,7 +1299,8 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m)
   if (!op->onack && !op->oncommit) {
     op->session_item.remove_myself();
     ldout(cct, 15) << "handle_osd_op_reply completed tid " << tid << dendl;
-    put_op_budget(op);
+    if (op->budgeted)
+      put_op_budget(op);
     ops.erase(tid);
     logger->set(l_osdc_op_active, ops.size());
     if (op->con)
diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index d90e314d1e0..ff858773066 100644
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -600,6 +600,8 @@ public:
 
     bool precalc_pgid;
 
+    bool budgeted;
+
     Op(const object_t& o, const object_locator_t& ol, vector<OSDOp>& op,
        int f, Context *ac, Context *co, eversion_t *ov) :
       session(NULL), session_item(this), incarnation(0),
@@ -609,7 +611,8 @@ public:
       outbl(NULL),
       flags(f), priority(0), onack(ac), oncommit(co),
       tid(0), attempts(0),
-      paused(false), objver(ov), reply_epoch(NULL), precalc_pgid(false) {
+      paused(false), objver(ov), reply_epoch(NULL), precalc_pgid(false),
+      budgeted(false) {
       ops.swap(op);
       
       /* initialize out_* to match op vector */
@@ -863,7 +866,7 @@ public:
   int recalc_op_target(Op *op);
   bool recalc_linger_op_target(LingerOp *op);
 
-  void send_linger(LingerOp *info);
+  void send_linger(LingerOp *info, bool first_send);
   void _linger_ack(LingerOp *info, int r);
   void _linger_commit(LingerOp *info, int r);
 
@@ -899,8 +902,10 @@ public:
       op_throttle_bytes.take(op_budget);
       op_throttle_ops.take(1);
     }
+    op->budgeted = true;
   }
   void put_op_budget(Op *op) {
+    assert(op->budgeted);
     int op_budget = calc_op_budget(op);
     op_throttle_bytes.put(op_budget);
     op_throttle_ops.put(1);
@@ -957,6 +962,7 @@ public:
 private:
   // low-level
   tid_t op_submit(Op *op, OSDSession *s = NULL);
+  tid_t _op_submit(Op *op, OSDSession *s);
 
   // public interface
  public: