FileStore: _split_collection should not create the collection

This will simplify adding a replay guard to create_collection.

Backport: bobtail
Signed-off-by: Samuel Just <sam.just@inktank.com>
(cherry picked from commit b184ff581a08c9e6ce5b858f06ccbe9d0e2a170b)

5 files changed, 69 insertions, 1 deletions

diff --git a/src/os/FileStore.cc b/src/os/FileStore.cc
index ac4d9b3afc8..6e002d39468 100644
--- a/src/os/FileStore.cc
+++ b/src/os/FileStore.cc
@@ -2596,6 +2596,15 @@ unsigned FileStore::_do_transaction(Transaction& t, uint64_t op_seq, int trans_n
 	uint32_t bits(i.get_u32());
 	uint32_t rem(i.get_u32());
 	coll_t dest(i.get_cid());
+	r = _split_collection_create(cid, bits, rem, dest, spos);
+      }
+      break;
+    case Transaction::OP_SPLIT_COLLECTION2:
+      {
+	coll_t cid(i.get_cid());
+	uint32_t bits(i.get_u32());
+	uint32_t rem(i.get_u32());
+	coll_t dest(i.get_cid());
 	r = _split_collection(cid, bits, rem, dest, spos);
       }
       break;
@@ -4616,6 +4625,43 @@ int FileStore::_split_collection(coll_t cid,
 				 const SequencerPosition &spos)
 {
   dout(15) << __func__ << " " << cid << " bits: " << bits << dendl;
+  int dstcmp = _check_replay_guard(dest, spos);
+  if (dstcmp < 0)
+    return 0;
+  if (dstcmp > 0 && !collection_empty(dest))
+    return -ENOTEMPTY;
+
+  int srccmp = _check_replay_guard(cid, spos);
+  if (srccmp < 0)
+    return 0;
+
+  _set_replay_guard(cid, spos, true);
+  _set_replay_guard(dest, spos, true);
+
+  Index from;
+  int r = get_index(cid, &from);
+
+  Index to;
+  if (!r)
+    r = get_index(dest, &to);
+
+  if (!r)
+    r = from->split(rem, bits, to);
+
+  _close_replay_guard(cid, spos);
+  _close_replay_guard(dest, spos);
+  return r;
+}
+
+// DEPRECATED: remove once we are sure there won't be any such transactions
+// replayed
+int FileStore::_split_collection_create(coll_t cid,
+					uint32_t bits,
+					uint32_t rem,
+					coll_t dest,
+					const SequencerPosition &spos)
+{
+  dout(15) << __func__ << " " << cid << " bits: " << bits << dendl;
   int r = _create_collection(dest);
   if (r < 0 && !(r == -EEXIST && replaying))
     return r;
diff --git a/src/os/FileStore.h b/src/os/FileStore.h
index b781de2b432..7e450ceb82c 100644
--- a/src/os/FileStore.h
+++ b/src/os/FileStore.h
@@ -475,6 +475,9 @@ private:
 		      const SequencerPosition &spos);
   int _split_collection(coll_t cid, uint32_t bits, uint32_t rem, coll_t dest,
                         const SequencerPosition &spos);
+  int _split_collection_create(coll_t cid, uint32_t bits, uint32_t rem,
+			       coll_t dest,
+			       const SequencerPosition &spos);
 
   virtual const char** get_tracked_conf_keys() const;
   virtual void handle_conf_change(const struct md_config_t *conf,
diff --git a/src/os/ObjectStore.cc b/src/os/ObjectStore.cc
index 70e8b6ed19e..813356f33ed 100644
--- a/src/os/ObjectStore.cc
+++ b/src/os/ObjectStore.cc
@@ -419,6 +419,19 @@ void ObjectStore::Transaction::dump(ceph::Formatter *f)
 	uint32_t bits(i.get_u32());
 	uint32_t rem(i.get_u32());
 	coll_t dest(i.get_cid());
+	f->dump_string("op_name", "op_split_collection_create");
+	f->dump_stream("collection") << cid;
+	f->dump_stream("bits") << bits;
+	f->dump_stream("rem") << rem;
+	f->dump_stream("dest") << dest;
+      }
+
+    case Transaction::OP_SPLIT_COLLECTION2:
+      {
+	coll_t cid(i.get_cid());
+	uint32_t bits(i.get_u32());
+	uint32_t rem(i.get_u32());
+	coll_t dest(i.get_cid());
 	f->dump_string("op_name", "op_split_collection");
 	f->dump_stream("collection") << cid;
 	f->dump_stream("bits") << bits;
diff --git a/src/os/ObjectStore.h b/src/os/ObjectStore.h
index 504422981f4..e88a67fe66b 100644
--- a/src/os/ObjectStore.h
+++ b/src/os/ObjectStore.h
@@ -153,6 +153,8 @@ public:
       OP_OMAP_RMKEYS = 33,  // cid, keyset
       OP_OMAP_SETHEADER = 34, // cid, header
       OP_SPLIT_COLLECTION = 35, // cid, bits, destination
+      OP_SPLIT_COLLECTION2 = 36, /* cid, bits, destination
+				    doesn't create the destination */
     };
 
   private:
@@ -555,7 +557,7 @@ public:
       uint32_t bits,
       uint32_t rem,
       coll_t destination) {
-      __u32 op = OP_SPLIT_COLLECTION;
+      __u32 op = OP_SPLIT_COLLECTION2;
       ::encode(op, tbl);
       ::encode(cid, tbl);
       ::encode(bits, tbl);
diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index b54fce36762..07fef776f80 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -4588,6 +4588,8 @@ void OSD::split_pgs(
     dout(10) << "m_seed " << i->ps() << dendl;
     dout(10) << "split_bits is " << split_bits << dendl;
 
+    rctx->transaction->create_collection(
+      coll_t(*i));
     rctx->transaction->split_collection(
       coll_t(parent->info.pgid),
       split_bits,
@@ -4607,6 +4609,8 @@ void OSD::split_pgs(
 	 ++k) {
       for (snapid_t j = k.get_start(); j < k.get_start() + k.get_len();
 	   ++j) {
+	rctx->transaction->create_collection(
+	  coll_t(*i, j));
 	rctx->transaction->split_collection(
 	  coll_t(parent->info.pgid, j),
 	  split_bits,