Decouple break_pg_locks from outbound OSD disconnections

src/client/messenger.cpp

@@ -199,7 +199,7 @@ void osd_messenger_t::init()
                 if (!cl->ping_time_remaining)
                 {
                     // Ping timed out, stop the client
-                    fprintf(stderr, "Ping timed out for OSD %ju (client %d), disconnecting peer\n", cl->osd_num, cl->peer_fd);
+                    fprintf(stderr, "Ping timed out for OSD %ju (client %d), disconnecting peer\n", cl->in_osd_num ? cl->in_osd_num : cl->osd_num, cl->peer_fd);
                     stop_client(peer_fd, true);
                     // Restart iterator because it may be invalidated
                     cl_it = clients.upper_bound(peer_fd);
@@ -230,7 +230,7 @@ void osd_messenger_t::init()
                             return;
                         }
                         int fail_fd = (op->reply.hdr.retval != 0 ? op->peer_fd : -1);
-                        auto fail_osd_num = cl->osd_num;
+                        auto fail_osd_num = cl->in_osd_num ? cl->in_osd_num : cl->osd_num;
                         cl->ping_time_remaining = 0;
                         delete op;
                         if (fail_fd >= 0)

src/client/messenger.h

@@ -237,6 +237,7 @@ public:
     void outbox_push(osd_op_t *cur_op);
     std::function<void(osd_op_t*)> exec_op;
     std::function<void(osd_num_t)> repeer_pgs;
+    std::function<void(osd_num_t)> break_pg_locks;
     std::function<bool(osd_client_t*, json11::Json)> check_config_hook;
     void read_requests();
     void send_replies();

src/client/msgr_stop.cpp

@@ -62,6 +62,10 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
         {
             fprintf(stderr, "[OSD %ju] Stopping client %d (OSD peer %ju)\n", osd_num, peer_fd, cl->osd_num);
         }
+        else if (cl->in_osd_num)
+        {
+            fprintf(stderr, "[OSD %ju] Stopping client %d (incoming OSD peer %ju)\n", osd_num, peer_fd, cl->in_osd_num);
+        }
         else
         {
             fprintf(stderr, "[OSD %ju] Stopping client %d (regular client)\n", osd_num, peer_fd);
@@ -104,6 +108,11 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
         }
     }
 #endif
+    if (cl->in_osd_num && break_pg_locks)
+    {
+        // Break PG locks
+        break_pg_locks(cl->in_osd_num);
+    }
     if (cl->osd_num)
     {
         // Then repeer PGs because cancel_op() callbacks can try to perform

src/osd/osd.cpp

@@ -85,6 +85,7 @@ osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop)
     msgr.ringloop = this->ringloop;
     msgr.exec_op = [this](osd_op_t *op) { exec_op(op); };
     msgr.repeer_pgs = [this](osd_num_t peer_osd) { repeer_pgs(peer_osd); };
+    msgr.break_pg_locks = [this](osd_num_t peer_osd) { break_pg_locks(peer_osd); };
     msgr.check_config_hook = [this](osd_client_t *cl, json11::Json conf) { return check_peer_config(cl, conf); };
     msgr.init();
 

src/osd/osd.h

@@ -279,6 +279,7 @@ class osd_t
     bool check_peer_config(osd_client_t *cl, json11::Json conf);
     void repeer_pgs(osd_num_t osd_num);
     void repeer_pg(pg_t & pg);
+    void break_pg_locks(osd_num_t osd_num);
     void start_pg_peering(pg_t & pg);
     void drop_dirty_pg_connections(pool_pg_num_t pg);
     void record_pg_lock(pg_t & pg, osd_num_t peer_osd, uint64_t pg_state);

src/osd/osd_peering.cpp

@@ -73,18 +73,25 @@ void osd_t::handle_peers()
     }
 }
 
+void osd_t::break_pg_locks(osd_num_t peer_osd)
+{
+    for (auto lock_it = pg_locks.begin(); lock_it != pg_locks.end(); )
+    {
+        if (lock_it->second.primary_osd == peer_osd)
+        {
+            if (log_level > 3)
+            {
+                printf("Break PG %u/%u lock on disconnection of OSD %ju\n", lock_it->first.pool_id, lock_it->first.pg_num, peer_osd);
+            }
+            pg_locks.erase(lock_it++);
+        }
+        else
+            lock_it++;
+    }
+}
+
 void osd_t::repeer_pgs(osd_num_t peer_osd)
 {
-    if (msgr.osd_peer_fds.find(peer_osd) == msgr.osd_peer_fds.end())
-    {
-        for (auto lock_it = pg_locks.begin(); lock_it != pg_locks.end(); )
-        {
-            if (lock_it->second.primary_osd == peer_osd)
-                pg_locks.erase(lock_it++);
-            else
-                lock_it++;
-        }
-    }
     // Re-peer affected PGs
     for (auto & p: pgs)
     {
@@ -471,6 +478,7 @@ void osd_t::relock_pg(pg_t & pg)
             auto pg_it = pgs.find(pg_id);
             if (pg_it == pgs.end())
             {
+                printf("Warning: PG %u/%u is gone during lock attempt\n", pg_id.pool_id, pg_id.pg_num);
                 return;
             }
             auto & pg = pg_it->second;

src/osd/osd_secondary.cpp

@@ -304,6 +304,10 @@ void osd_t::exec_sec_lock(osd_op_t *cur_op)
             finish_op(cur_op, -EBUSY);
             return;
         }
+        if (log_level > 3)
+        {
+            printf("Lock PG %u/%u for OSD %ju\n", ppg.pool_id, ppg.pg_num, cl->in_osd_num);
+        }
         pg_locks[ppg] = (osd_pg_lock_t){
             .primary_osd = cl->in_osd_num,
             .state = cur_op->req.sec_lock.pg_state,
@@ -311,6 +315,10 @@ void osd_t::exec_sec_lock(osd_op_t *cur_op)
     }
     else if (lock_it != pg_locks.end() && lock_it->second.primary_osd == cl->in_osd_num)
     {
+        if (log_level > 3)
+        {
+            printf("Unlock PG %u/%u by OSD %ju\n", ppg.pool_id, ppg.pg_num, cl->in_osd_num);
+        }
         pg_locks.erase(lock_it);
     }
     finish_op(cur_op, 0);