ReplicatedBackend: take read locks for clone sources during recovery

Otherwise, we run the risk of a clone source which hasn't actually
come into existence yet being used if we grab a clone which *just*
got added the the ssc, but has not yet actually had time to be
created (can't rely on message ordering here since recovery messages
don't necessarily order with client IO!).

Fixes: http://tracker.ceph.com/issues/17831
Signed-off-by: Samuel Just <sjust@redhat.com>

src/osd/PGBackend.h

@@ -186,6 +186,12 @@ typedef ceph::shared_ptr<const OSDMap> OSDMapRef;
        const hobject_t &hoid,
        map<string, bufferlist> &attrs) = 0;
 
+     virtual bool try_lock_for_read(
+       const hobject_t &hoid,
+       ObcLockManager &manager) = 0;
+
+     virtual void release_locks(ObcLockManager &manager) = 0;
+
      virtual void op_applied(
        const eversion_t &applied_version) = 0;
 

src/osd/PrimaryLogPG.h

@@ -335,11 +335,26 @@ class PrimaryLogPG : public PG, public PGBackend::Listener {
   const pg_pool_t &get_pool() const override {
     return pool.info;
   }
+
   ObjectContextRef get_obc(
     const hobject_t &hoid,
     map<string, bufferlist> &attrs) override {
     return get_object_context(hoid, true, &attrs);
   }
+
+  bool try_lock_for_read(
+    const hobject_t &hoid,
+    ObcLockManager &manager) override {
+    auto obc = get_object_context(hoid, false, nullptr);
+    if (!obc)
+      return false;
+    return manager.try_get_read_lock(hoid, obc);
+  }
+
+  void release_locks(ObcLockManager &manager) {
+    release_object_locks(manager);
+  }
+
   void pgb_set_object_snap_mapping(
     const hobject_t &soid,
     const set<snapid_t> &snaps,
@@ -352,7 +367,6 @@ class PrimaryLogPG : public PG, public PGBackend::Listener {
     return clear_object_snap_mapping(t, soid);
   }
 
-
   void log_operation(
     const vector<pg_log_entry_t> &logv,
     boost::optional<pg_hit_set_history_t> &hset_history,