Allow to enable PG locks online when changing local_reads in pool configuration

docs/intro/author.en.md

@@ -90,7 +90,7 @@ need a commercial license which doesn't contain open-source requirements.
 
 ## Why?
 
-Because I believe into the spirit of copyleft (Linux wouldn't became so popular without GPL!)
+Because I believe into the spirit of copyleft (Linux wouldn't become so popular without GPL!)
 and, at the same time, I want to have a way to monetize the product.
 
 Existing licenses including AGPL are useless for it with an SDS - SDS is a very deeply

src/osd/osd.h

@@ -278,6 +278,7 @@ class osd_t
     void handle_peers();
     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 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_cluster.cpp

@@ -432,9 +432,16 @@ void osd_t::apply_pg_locks_localize_only()
         }
         auto & pool_cfg = pool_it->second;
         auto & pg = pp.second;
+        auto old_disable_pg_locks = pg.disable_pg_locks;
         pg.disable_pg_locks = pg_locks_localize_only &&
             pool_cfg.scheme == POOL_SCHEME_REPLICATED &&
             pool_cfg.local_reads == POOL_LOCAL_READ_PRIMARY;
+        if (!pg.disable_pg_locks && old_disable_pg_locks)
+        {
+            // Relock PG
+            printf("[PG %u/%u] Repeer to enable PG locks\n", pg.pool_id, pg.pg_num);
+            repeer_pg(pg);
+        }
     }
 }
 

src/osd/osd_peering.cpp

@@ -104,21 +104,26 @@ void osd_t::repeer_pgs(osd_num_t peer_osd)
             {
                 // Repeer this pg
                 printf("[PG %u/%u] Repeer because of OSD %ju\n", pg.pool_id, pg.pg_num, peer_osd);
-                if (!(pg.state & (PG_ACTIVE | PG_REPEERING)) || pg.can_repeer())
+                repeer_pg(pg);
-                {
-                    start_pg_peering(pg);
-                }
-                else
-                {
-                    // Stop accepting new operations, wait for current ones to finish or fail
-                    pg.state = pg.state & ~PG_ACTIVE | PG_REPEERING;
-                    report_pg_state(pg);
-                }
             }
         }
     }
 }
 
+void osd_t::repeer_pg(pg_t & pg)
+{
+    if (!(pg.state & (PG_ACTIVE | PG_REPEERING)) || pg.can_repeer())
+    {
+        start_pg_peering(pg);
+    }
+    else
+    {
+        // Stop accepting new operations, wait for current ones to finish or fail
+        pg.state = pg.state & ~PG_ACTIVE | PG_REPEERING;
+        report_pg_state(pg);
+    }
+}
+
 // Reset PG state (when peering or stopping)
 void osd_t::reset_pg(pg_t & pg)
 {