Merge remote-tracking branch 'gh/hammer'

Conflicts:
	src/gmock
	src/msg/xio/XioPortal.h

qa/workunits/cephtool/test.sh

@@ -319,6 +319,9 @@ function test_tiering()
   ceph osd pool create cache3 2
   ceph osd tier add-cache slow cache3 1024000
   ceph osd dump | grep cache3 | grep bloom | grep 'false_positive_probability: 0.05' | grep 'target_bytes 1024000' | grep '1200s x4'
+  ceph osd tier remove slow cache3 2> $TMPFILE || true
+  check_response "EBUSY: tier pool 'cache3' is the overlay for 'slow'; please remove-overlay first"
+  ceph osd tier remove-overlay slow
   ceph osd tier remove slow cache3
   ceph osd pool ls | grep cache3
   ceph osd pool delete cache3 cache3 --yes-i-really-really-mean-it
@@ -327,6 +330,22 @@ function test_tiering()
   ceph osd pool delete slow2 slow2 --yes-i-really-really-mean-it
   ceph osd pool delete slow slow --yes-i-really-really-mean-it
 
+  # check add-cache whether work
+  ceph osd pool create datapool 2
+  ceph osd pool create cachepool 2
+  ceph osd tier add-cache datapool cachepool 1024000
+  ceph osd tier cache-mode cachepool writeback
+  dd if=/dev/zero of=/tmp/add-cache bs=4K count=1
+  rados -p datapool put object /tmp/add-cache
+  rados -p cachepool stat object
+  rados -p cachepool cache-flush object
+  rados -p datapool stat object
+  ceph osd tier remove-overlay datapool
+  ceph osd tier remove datapool cachepool
+  ceph osd pool delete cachepool cachepool --yes-i-really-really-mean-it
+  ceph osd pool delete datapool datapool --yes-i-really-really-mean-it
+  rm -rf /tmp/add-cache
+
   # protection against pool removal when used as tiers
   ceph osd pool create datapool 2
   ceph osd pool create cachepool 2
@@ -335,6 +354,7 @@ function test_tiering()
   check_response "EBUSY: pool 'cachepool' is a tier of 'datapool'"
   ceph osd pool delete datapool datapool --yes-i-really-really-mean-it 2> $TMPFILE || true
   check_response "EBUSY: pool 'datapool' has tiers cachepool"
+  ceph osd tier remove-overlay datapool
   ceph osd tier remove datapool cachepool
   ceph osd pool delete cachepool cachepool --yes-i-really-really-mean-it
   ceph osd pool delete datapool datapool --yes-i-really-really-mean-it

src/cls/rbd/cls_rbd.cc

@@ -768,6 +768,7 @@ int get_flags(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
  * Input:
  * @params flags image flags
  * @params mask image flag mask
+ * @params snap_id which snapshot to update, or CEPH_NOSNAP (uint64_t)
  *
  * Output:
  * none
@@ -778,33 +779,61 @@ int set_flags(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 {
   uint64_t flags;
   uint64_t mask;
+  uint64_t snap_id = CEPH_NOSNAP;
   bufferlist::iterator iter = in->begin();
   try {
     ::decode(flags, iter);
     ::decode(mask, iter);
+    if (!iter.end()) {
+      ::decode(snap_id, iter);
+    }
   } catch (const buffer::error &err) {
     return -EINVAL;
   }
 
   // check that size exists to make sure this is a header object
   // that was created correctly
+  int r;
   uint64_t orig_flags = 0;
-  int r = read_key(hctx, "flags", &orig_flags);
-  if (r < 0 && r != -ENOENT) {
-    CLS_ERR("Could not read image's flags off disk: %s",
-            cpp_strerror(r).c_str());
-    return r;
+  cls_rbd_snap snap_meta;
+  string snap_meta_key;
+  if (snap_id == CEPH_NOSNAP) {
+    r = read_key(hctx, "flags", &orig_flags);
+    if (r < 0 && r != -ENOENT) {
+      CLS_ERR("Could not read image's flags off disk: %s",
+              cpp_strerror(r).c_str());
+      return r;
+    }
+  } else {
+    key_from_snap_id(snap_id, &snap_meta_key);
+    r = read_key(hctx, snap_meta_key, &snap_meta);
+    if (r < 0) {
+      CLS_ERR("Could not read snapshot: snap_id=%" PRIu64 ": %s",
+              snap_id, cpp_strerror(r).c_str());
+      return r;
+    }
+    orig_flags = snap_meta.flags;
   }
 
   flags = (orig_flags & ~mask) | (flags & mask);
-  CLS_LOG(20, "set_flags flags=%llu orig_flags=%llu", (unsigned long long)flags,
-          (unsigned long long)orig_flags);
+  CLS_LOG(20, "set_flags snap_id=%" PRIu64 ", orig_flags=%" PRIu64 ", "
+              "new_flags=%" PRIu64 ", mask=%" PRIu64, snap_id, orig_flags,
+              flags, mask);
+
+  if (snap_id == CEPH_NOSNAP) {
+    bufferlist bl;
+    ::encode(flags, bl);
+    r = cls_cxx_map_set_val(hctx, "flags", &bl);
+  } else {
+    snap_meta.flags = flags;
+
+    bufferlist bl;
+    ::encode(snap_meta, bl);
+    r = cls_cxx_map_set_val(hctx, snap_meta_key, &bl);
+  }
 
-  bufferlist flagsbl;
-  ::encode(flags, flagsbl);
-  r = cls_cxx_map_set_val(hctx, "flags", &flagsbl);
   if (r < 0) {
-    CLS_ERR("error updating flags: %d", r);
+    CLS_ERR("error updating flags: %s", cpp_strerror(r).c_str());
     return r;
   }
   return 0;

src/cls/rbd/cls_rbd_client.cc

@@ -280,12 +280,13 @@ namespace librbd {
       return 0;
     }
 
-    void set_flags(librados::ObjectWriteOperation *op, uint64_t flags,
-		   uint64_t mask)
+    void set_flags(librados::ObjectWriteOperation *op, snapid_t snap_id,
+                   uint64_t flags, uint64_t mask)
     {
       bufferlist inbl;
       ::encode(flags, inbl);
       ::encode(mask, inbl);
+      ::encode(snap_id, inbl);
       op->exec("rbd", "set_flags", inbl);
     }
 

src/cls/rbd/cls_rbd_client.h

@@ -50,7 +50,7 @@ namespace librbd {
     int get_flags(librados::IoCtx *ioctx, const std::string &oid,
 		  uint64_t *flags, const std::vector<snapid_t> &snap_ids,
 		  vector<uint64_t> *snap_flags);
-    void set_flags(librados::ObjectWriteOperation *op,
+    void set_flags(librados::ObjectWriteOperation *op, snapid_t snap_id,
                    uint64_t flags, uint64_t mask);
     int remove_parent(librados::IoCtx *ioctx, const std::string &oid);
     void remove_parent(librados::ObjectWriteOperation *op);

src/common/config_opts.h

@@ -640,6 +640,7 @@ OPTION(osd_debug_drop_pg_create_probability, OPT_DOUBLE, 0)
 OPTION(osd_debug_drop_pg_create_duration, OPT_INT, 1)
 OPTION(osd_debug_drop_op_probability, OPT_DOUBLE, 0)   // probability of stalling/dropping a client op
 OPTION(osd_debug_op_order, OPT_BOOL, false)
+OPTION(osd_debug_scrub_chance_rewrite_digest, OPT_U64, 0)
 OPTION(osd_debug_verify_snaps_on_info, OPT_BOOL, false)
 OPTION(osd_debug_verify_stray_on_activate, OPT_BOOL, false)
 OPTION(osd_debug_skip_full_check_in_backfill_reservation, OPT_BOOL, false)

src/crush/crush.c

@@ -42,6 +42,8 @@ int crush_get_bucket_item_weight(const struct crush_bucket *b, int p)
 		return ((struct crush_bucket_tree *)b)->node_weights[crush_calc_tree_node(p)];
 	case CRUSH_BUCKET_STRAW:
 		return ((struct crush_bucket_straw *)b)->item_weights[p];
+	case CRUSH_BUCKET_STRAW2:
+		return ((struct crush_bucket_straw2 *)b)->item_weights[p];
 	}
 	return 0;
 }
@@ -79,6 +81,14 @@ void crush_destroy_bucket_straw(struct crush_bucket_straw *b)
 	kfree(b);
 }
 
+void crush_destroy_bucket_straw2(struct crush_bucket_straw2 *b)
+{
+	kfree(b->item_weights);
+	kfree(b->h.perm);
+	kfree(b->h.items);
+	kfree(b);
+}
+
 void crush_destroy_bucket(struct crush_bucket *b)
 {
 	switch (b->alg) {
@@ -94,6 +104,9 @@ void crush_destroy_bucket(struct crush_bucket *b)
 	case CRUSH_BUCKET_STRAW:
 		crush_destroy_bucket_straw((struct crush_bucket_straw *)b);
 		break;
+	case CRUSH_BUCKET_STRAW2:
+		crush_destroy_bucket_straw2((struct crush_bucket_straw2 *)b);
+		break;
 	}
 }
 

src/include/xlist.h

@@ -137,7 +137,10 @@ class xlist {
   }
 
   T front() { return static_cast<T>(_front->_item); }
+  const T front() const { return static_cast<const T>(_front->_item); }
+
   T back() { return static_cast<T>(_back->_item); }
+  const T back() const { return static_cast<const T>(_back->_item); }
 
   void pop_front() {
     assert(!empty());

src/librbd/AioRequest.cc

@@ -209,8 +209,12 @@ namespace librbd {
           }
 
           // If parent still exists, overlap might also have changed.
+          uint64_t parent_overlap;
+          r = m_ictx->get_parent_overlap(CEPH_NOSNAP, &parent_overlap);
+          assert(r == 0);
+
           uint64_t newlen = m_ictx->prune_parent_extents(
-            m_image_extents, m_ictx->parent_md.overlap);
+            m_image_extents, parent_overlap);
           if (newlen != 0) {
             // create and kick off a CopyupRequest
             CopyupRequest *new_req = new CopyupRequest(m_ictx, m_oid,
@@ -340,13 +344,17 @@ namespace librbd {
 	}
 
 	// If parent still exists, overlap might also have changed.
+	uint64_t parent_overlap;
+        r = m_ictx->get_parent_overlap(CEPH_NOSNAP, &parent_overlap);
+        assert(r == 0);
+
 	uint64_t newlen = m_ictx->prune_parent_extents(
-	  m_object_image_extents, m_ictx->parent_md.overlap);
+	  m_object_image_extents, parent_overlap);
 
 	// copyup the entire object up to the overlap point, if any
 	if (newlen != 0) {
 	  ldout(m_ictx->cct, 20) << "should_complete(" << this << ") overlap "
-				 << m_ictx->parent_md.overlap << " newlen "
+				 << parent_overlap << " newlen "
 				 << newlen << " image_extents"
 				 << m_object_image_extents << dendl;
 

src/librbd/AsyncFlattenRequest.cc

@@ -49,7 +49,10 @@ class AsyncFlattenObjectContext : public C_AsyncObjectThrottle {
       }
 
       // resize might have occurred while flatten is running
-      overlap = min(m_image_ctx.size, m_image_ctx.parent_md.overlap);
+      uint64_t parent_overlap;
+      int r = m_image_ctx.get_parent_overlap(CEPH_NOSNAP, &parent_overlap);
+      assert(r == 0);
+      overlap = min(m_image_ctx.size, parent_overlap);
     }
 
     // map child object onto the parent

src/librbd/AsyncRequest.h

@@ -19,7 +19,7 @@ class AsyncRequest
   virtual ~AsyncRequest();
 
   void complete(int r) {
-    if (m_canceled) {
+    if (m_canceled && safely_cancel(r)) {
       m_on_finish->complete(-ERESTART);
       delete this;
     } else if (should_complete(r)) {
@@ -44,6 +44,9 @@ class AsyncRequest
   librados::AioCompletion *create_callback_completion();
   Context *create_callback_context();
 
+  virtual bool safely_cancel(int r) {
+    return true;
+  }
   virtual bool should_complete(int r) = 0;
 
 private: