SimpleMessenger: Don't grab the lock when sending messages if we don't have to

We'd like it if sending a message didn't require any global locks, but the
submit_message() function conditionally needs it in order to create new
Pipes. So:
1) When failing on a dud Pipe, verify that it's still the Pipe the Connection
is linked to; if not, try sending along the newly-linked Pipe.
2) Add an "already_locked" param to submit_message
3) Have the Connection-based interface set this param to false, and
the addr-based interface set it to true, reflecting whether they have
taken the SimpleMessenger::lock.
4) If we discover we need to reference global data structures in
submit_mesage:
  4a) if locked, do as we previously have
  4b) if not locked, take the lock and call into submit_message again.

The net effect of this is that in the typical case, the Connection-based
_send_message() function no longer acquires global locks, only per-Connection
ones. In the case where the Connection must recreate a Pipe, it falls back to
performing like the addr-based _send_message() does. In the case where
we are racing with somebody else recreating a Pipe(either us or the other
end), we may try twice but we will still only take per-Connection/Pipe locks,
which is a fair tradeoff for not taking the global lock.

Signed-off-by: Greg Farnum <greg@inktank.com>
Reviewed-by: Sage Weil <sage@inktank.com>

diff --git a/src/msg/SimpleMessenger.cc b/src/msg/SimpleMessenger.cc
index b612fcfe0e6285445cd645f91ffc5bab1c94ff24..ddc0dcaffcc75e1e9f5d253091bd67f39be238e6 100644 (file)
--- a/src/msg/SimpleMessenger.cc
+++ b/src/msg/SimpleMessenger.cc
@@ -113,7 +113,7 @@ int SimpleMessenger::_send_message(Message *m, const entity_inst_t& dest,
   lock.Lock();
   Pipe *pipe = _lookup_pipe(dest.addr);
   submit_message(m, (pipe ? pipe->connection_state.get() : NULL),
-                 dest.addr, dest.name.type(), lazy);
+                 dest.addr, dest.name.type(), lazy, true);
   lock.Unlock();
   return 0;
 }
@@ -131,9 +131,7 @@ int SimpleMessenger::_send_message(Message *m, Connection *con, bool lazy)
       << " " << m << " con " << con
       << dendl;
 
-  lock.Lock();
-  submit_message(m, con, con->get_peer_addr(), con->get_peer_type(), lazy);
-  lock.Unlock();
+  submit_message(m, con, con->get_peer_addr(), con->get_peer_type(), lazy, false);
   return 0;
 }
 
@@ -397,9 +395,9 @@ ConnectionRef SimpleMessenger::get_loopback_connection()
 }
 
 void SimpleMessenger::submit_message(Message *m, Connection *con,
-                                    const entity_addr_t& dest_addr, int dest_type, bool lazy)
+                                    const entity_addr_t& dest_addr, int dest_type,
+                                    bool lazy, bool already_locked)
 {
-
   if (cct->_conf->ms_dump_on_send) {
     m->encode(-1, true);
     ldout(cct, 0) << "submit_message " << *m << "\n";
@@ -422,8 +420,9 @@ void SimpleMessenger::submit_message(Message *m, Connection *con,
       m->put();
       return;
     }
-    if (pipe) {
-      pipe->pipe_lock.Lock();
+    while (pipe && ok) {
+      // we loop in case of a racing reconnect, either from us or them
+      pipe->pipe_lock.Lock(); // can't use a Locker because of the Pipe ref
       if (pipe->state != Pipe::STATE_CLOSED) {
        ldout(cct,20) << "submit_message " << *m << " remote, " << dest_addr << ", have pipe." << dendl;
        pipe->_send(m);
@@ -431,12 +430,20 @@ void SimpleMessenger::submit_message(Message *m, Connection *con,
        pipe->put();
        return;
       }
+      Pipe *current_pipe;
+      ok = con->try_get_pipe((RefCountedObject**)&current_pipe);
       pipe->pipe_lock.Unlock();
-      pipe->put();
-      ldout(cct,20) << "submit_message " << *m << " remote, " << dest_addr
-                   << ", had pipe " << pipe << ", but it closed." << dendl;
-      m->put();
-      return;
+      if (current_pipe == pipe) {
+       ldout(cct,20) << "submit_message " << *m << " remote, " << dest_addr
+                     << ", had pipe " << pipe << ", but it closed." << dendl;
+       pipe->put();
+       current_pipe->put();
+       m->put();
+       return;
+      } else {
+       pipe->put();
+       pipe = current_pipe;
+      }
     }
   }
 
@@ -459,7 +466,15 @@ void SimpleMessenger::submit_message(Message *m, Connection *con,
     m->put();
   } else {
     ldout(cct,20) << "submit_message " << *m << " remote, " << dest_addr << ", new pipe." << dendl;
-    connect_rank(dest_addr, dest_type, con, m);
+    if (!already_locked) {
+      /** We couldn't handle the Message without reference to global data, so
+       *  grab the lock and do it again. If we got here, we know it's a non-lossy
+       *  Connection, so we can use our existing pointer without doing another lookup. */
+      Mutex::Locker l(lock);
+      submit_message(m, con, dest_addr, dest_type, lazy, true);
+    } else {
+      connect_rank(dest_addr, dest_type, con, m);
+    }
   }
 }
 

diff --git a/src/msg/SimpleMessenger.h b/src/msg/SimpleMessenger.h
index e6e1fb1d0cb86d2e89cf6b9e1ebe298f6361acdf..4cb0ad56b8e4dd7eed8cff6d3c9187d42699f434 100644 (file)
--- a/src/msg/SimpleMessenger.h
+++ b/src/msg/SimpleMessenger.h
@@ -273,9 +273,14 @@ private:
    * @param dest_type The peer type of the address we're sending to
    * @param lazy If true, do not establish or fix a Connection to send the Message;
    * just drop silently under failure.
+   * @param already_locked If false, submit_message() will acquire the
+   * SimpleMessenger lock before accessing shared data structures; otherwise
+   * it will assume the lock is held. NOTE: if you are making a request
+   * without locking, you MUST have filled in the con with a valid pointer.
    */
   void submit_message(Message *m, Connection *con,
-                      const entity_addr_t& addr, int dest_type, bool lazy);
+                     const entity_addr_t& addr, int dest_type,
+                     bool lazy, bool already_locked);
   /**
    * Look through the pipes in the pipe_reap_queue and tear them down.
    */