Merge branch 'wip-4064-sam'

Reviewed-by: Sage Weil <sage@inktank.com>

9 files changed, 119 insertions, 11 deletions

diff --git a/src/os/DBObjectMap.cc b/src/os/DBObjectMap.cc
index 10b7b705a4b..f884266ff75 100644
--- a/src/os/DBObjectMap.cc
+++ b/src/os/DBObjectMap.cc
@@ -242,8 +242,7 @@ bool DBObjectMap::parse_hobject_key_v0(const string &in, coll_t *c,
   *c = coll_t(coll);
   int64_t pool = -1;
   pg_t pg;
-  snapid_t pg_snap;
-  if (c->is_pg(pg, pg_snap))
+  if (c->is_pg_prefix(pg))
     pool = (int64_t)pg.pool();
   (*hoid) = hobject_t(name, key, snap, hash, pool);
   return true;
diff --git a/src/os/FileStore.cc b/src/os/FileStore.cc
index 44f3b571960..9ce7550d9a5 100644
--- a/src/os/FileStore.cc
+++ b/src/os/FileStore.cc
@@ -2481,7 +2481,7 @@ unsigned FileStore::_do_transaction(Transaction& t, uint64_t op_seq, int trans_n
       {
 	coll_t cid = i.get_cid();
 	if (_check_replay_guard(cid, spos) > 0)
-	  r = _create_collection(cid);
+	  r = _create_collection(cid, spos);
       }
       break;
 
@@ -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;
@@ -2610,7 +2619,8 @@ unsigned FileStore::_do_transaction(Transaction& t, uint64_t op_seq, int trans_n
 
       if (r == -ENOENT && !(op == Transaction::OP_CLONERANGE ||
 			    op == Transaction::OP_CLONE ||
-			    op == Transaction::OP_CLONERANGE2))
+			    op == Transaction::OP_CLONERANGE2 ||
+			    op == Transaction::OP_COLL_ADD))
 	// -ENOENT is normally okay
 	// ...including on a replayed OP_RMCOLL with !stable_commits
 	ok = true;
@@ -4457,6 +4467,30 @@ ObjectMap::ObjectMapIterator FileStore::get_omap_iterator(coll_t c,
   return object_map->get_iterator(hoid);
 }
 
+int FileStore::_create_collection(
+  coll_t c,
+  const SequencerPosition &spos)
+{
+  char fn[PATH_MAX];
+  get_cdir(c, fn, sizeof(fn));
+  dout(15) << "create_collection " << fn << dendl;
+  int r = ::mkdir(fn, 0755);
+  if (r < 0)
+    r = -errno;
+  if (r == -EEXIST && replaying)
+    r = 0;
+  dout(10) << "create_collection " << fn << " = " << r << dendl;
+
+  if (r < 0)
+    return r;
+  r = init_index(c);
+  if (r < 0)
+    return r;
+  _set_replay_guard(c, spos);
+  return 0;
+}
+
+// DEPRECATED -- remove with _split_collection_create
 int FileStore::_create_collection(coll_t c) 
 {
   char fn[PATH_MAX];
@@ -4612,6 +4646,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..3336e59378e 100644
--- a/src/os/FileStore.h
+++ b/src/os/FileStore.h
@@ -453,6 +453,7 @@ public:
   ObjectMap::ObjectMapIterator get_omap_iterator(coll_t c, const hobject_t &hoid);
 
   int _create_collection(coll_t c);
+  int _create_collection(coll_t c, const SequencerPosition &spos);
   int _destroy_collection(coll_t c);
   int _collection_add(coll_t c, coll_t ocid, const hobject_t& o,
 		      const SequencerPosition& spos);
@@ -475,6 +476,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/LFNIndex.cc b/src/os/LFNIndex.cc
index 5e505638d15..1aae19b3a15 100644
--- a/src/os/LFNIndex.cc
+++ b/src/os/LFNIndex.cc
@@ -893,8 +893,7 @@ bool LFNIndex::lfn_parse_object_name_keyless(const string &long_name, hobject_t
   bool r = parse_object(long_name.c_str(), *out);
   int64_t pool = -1;
   pg_t pg;
-  snapid_t snap;
-  if (coll().is_pg(pg, snap))
+  if (coll().is_pg_prefix(pg))
     pool = (int64_t)pg.pool();
   out->pool = pool;
   if (!r) return r;
@@ -985,8 +984,7 @@ bool LFNIndex::lfn_parse_object_name_poolless(const string &long_name,
 
   int64_t pool = -1;
   pg_t pg;
-  snapid_t pg_snap;
-  if (coll().is_pg(pg, pg_snap))
+  if (coll().is_pg_prefix(pg))
     pool = (int64_t)pg.pool();
   (*out) = hobject_t(name, key, snap, hash, pool);
   return true;
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 1785678ced4..fe85532db0b 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -4732,6 +4732,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,
@@ -4742,6 +4744,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,
diff --git a/src/osd/osd_types.cc b/src/osd/osd_types.cc
index 786d0e876b4..d6faa273c47 100644
--- a/src/osd/osd_types.cc
+++ b/src/osd/osd_types.cc
@@ -294,10 +294,26 @@ bool coll_t::is_pg(pg_t& pgid, snapid_t& snap) const
   const char *snap_start = strchr(cstr, '_');
   if (!snap_start)
     return false;
-  if (strncmp(snap_start, "_head", 5) == 0)
+  if (strncmp(snap_start, "_head", 5) == 0) {
     snap = CEPH_NOSNAP;
-  else
+  } else {
+    errno = 0;
     snap = strtoull(snap_start+1, 0, 16);
+    if (errno)
+      return false;
+  }
+  return true;
+}
+
+bool coll_t::is_pg_prefix(pg_t& pgid) const
+{
+  const char *cstr(str.c_str());
+
+  if (!pgid.parse(cstr))
+    return false;
+  const char *snap_start = strchr(cstr, '_');
+  if (!snap_start)
+    return false;
   return true;
 }
 
diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index e0680574057..4d8789755a8 100644
--- a/src/osd/osd_types.h
+++ b/src/osd/osd_types.h
@@ -355,6 +355,7 @@ public:
     return str < rhs.str;
   }
 
+  bool is_pg_prefix(pg_t& pgid) const;
   bool is_pg(pg_t& pgid, snapid_t& snap) const;
   bool is_temp(pg_t& pgid) const;
   bool is_removal(uint64_t *seq, pg_t *pgid) const;