filestore: guard replay of collection_add

- set guard on apply
- check guard on replay

Signed-off-by: Sage Weil <sage.weil@dreamhost.com>

diff --git a/src/os/FileStore.cc b/src/os/FileStore.cc
index 0c00b23d7f59b7bb24aae7c3cd22024ad5b6a07f..5089e70994f4d474750ea9cc4f7575beb8d3cc75 100644 (file)
--- a/src/os/FileStore.cc
+++ b/src/os/FileStore.cc
@@ -2617,7 +2617,7 @@ unsigned FileStore::_do_transaction(Transaction& t, uint64_t op_seq, int trans_n
        coll_t ocid = i.get_cid();
        coll_t ncid = i.get_cid();
        hobject_t oid = i.get_oid();
-       r = _collection_add(ocid, ncid, oid);
+       r = _collection_add(ocid, ncid, oid, spos);
       }
       break;
 
@@ -4239,10 +4239,24 @@ int FileStore::_destroy_collection(coll_t c)
 }
 
 
-int FileStore::_collection_add(coll_t c, coll_t cid, const hobject_t& o) 
+int FileStore::_collection_add(coll_t c, coll_t cid, const hobject_t& o,
+                              const SequencerPosition& spos) 
 {
   dout(15) << "collection_add " << c << "/" << o << " " << cid << "/" << o << dendl;
+  
+  if (!_check_replay_guard(c, o, spos))
+    return 0;
+
   int r = lfn_link(cid, c, o);
+
+  // set guard on object so we don't do this again
+  if (r >= 0) {
+    int fd = lfn_open(c, o, 0);
+    assert(fd >= 0);
+    _set_replay_guard(fd, spos);
+    TEMP_FAILURE_RETRY(::close(fd));
+  }
+
   dout(10) << "collection_add " << c << "/" << o << " " << cid << "/" << o << " = " << r << dendl;
   return r;
 }

diff --git a/src/os/FileStore.h b/src/os/FileStore.h
index a633c31d8b414f8b6087b7bcfa3078191f5d9900..e29419af112da1341255f762a7854e6fb13e0901 100644 (file)
--- a/src/os/FileStore.h
+++ b/src/os/FileStore.h
@@ -418,7 +418,8 @@ public:
 
   int _create_collection(coll_t c);
   int _destroy_collection(coll_t c);
-  int _collection_add(coll_t c, coll_t ocid, const hobject_t& o);
+  int _collection_add(coll_t c, coll_t ocid, const hobject_t& o,
+                     const SequencerPosition& spos);
   int _collection_remove(coll_t c, const hobject_t& o);
   int _collection_move(coll_t c, coll_t ocid, const hobject_t& o);