Merge pull request #10684 from dillaman/wip-16904-jewel

jewel: rbd-mirror: reduce memory footprint during journal replay

Reviewed-by: Loic Dachary <ldachary@redhat.com>
Reviewed-by: Jason Dillaman <dillaman@redhat.com>

qa/workunits/rbd/rbd_mirror.sh

@@ -186,6 +186,14 @@ for i in ${image2} ${image4}; do
   compare_images ${POOL} ${i}
 done
 
+testlog "TEST: snapshot rename"
+snap_name='snap_rename'
+create_snapshot ${CLUSTER2} ${POOL} ${image2} "${snap_name}_0"
+for i in `seq 1 20`; do
+  rename_snapshot ${CLUSTER2} ${POOL} ${image2} "${snap_name}_$(expr ${i} - 1)" "${snap_name}_${i}"
+done
+wait_for_snap_present ${CLUSTER1} ${POOL} ${image2} "${snap_name}_${i}"
+
 testlog "TEST: disable mirror while daemon is stopped"
 stop_mirror ${CLUSTER1}
 stop_mirror ${CLUSTER2}

qa/workunits/rbd/rbd_mirror_helpers.sh

@@ -182,6 +182,7 @@ start_mirror()
 	--pid-file=$(daemon_pid_file "${cluster}") \
 	--log-file=${TEMPDIR}/rbd-mirror.${cluster}_daemon.\$cluster.\$pid.log \
 	--admin-socket=${TEMPDIR}/rbd-mirror.${cluster}_daemon.\$cluster.asok \
+	--rbd-mirror-journal-poll-age=1 \
 	--debug-rbd=30 --debug-journaler=30 \
 	--debug-rbd_mirror=30 \
 	--daemonize=true
@@ -514,6 +515,17 @@ remove_snapshot()
     rbd --cluster ${cluster} -p ${pool} snap rm ${image}@${snap}
 }
 
+rename_snapshot()
+{
+    local cluster=$1
+    local pool=$2
+    local image=$3
+    local snap=$4
+    local new_snap=$5
+
+    rbd --cluster ${cluster} -p ${pool} snap rename ${image}@${snap} ${image}@${new_snap}
+}
+
 purge_snapshots()
 {
     local cluster=$1

src/common/config_opts.h

@@ -1210,10 +1210,14 @@ OPTION(rbd_journal_object_flush_interval, OPT_INT, 0) // maximum number of pendi
 OPTION(rbd_journal_object_flush_bytes, OPT_INT, 0) // maximum number of pending bytes per journal object
 OPTION(rbd_journal_object_flush_age, OPT_DOUBLE, 0) // maximum age (in seconds) for pending commits
 OPTION(rbd_journal_pool, OPT_STR, "") // pool for journal objects
+OPTION(rbd_journal_max_payload_bytes, OPT_U32, 16384) // maximum journal payload size before splitting
 
 /**
  * RBD Mirror options
  */
+OPTION(rbd_mirror_journal_commit_age, OPT_DOUBLE, 5) // commit time interval, seconds
+OPTION(rbd_mirror_journal_poll_age, OPT_DOUBLE, 5) // maximum age (in seconds) between successive journal polls
+OPTION(rbd_mirror_journal_max_fetch_bytes, OPT_U32, 32768) // maximum bytes to read from each journal data object per fetch
 OPTION(rbd_mirror_sync_point_update_age, OPT_DOUBLE, 30) // number of seconds between each update of the image sync point object number
 OPTION(rbd_mirror_concurrent_image_syncs, OPT_U32, 5) // maximum number of image syncs in parallel
 

src/journal/Entry.cc

@@ -9,7 +9,7 @@
 
 #define dout_subsys ceph_subsys_journaler
 #undef dout_prefix
-#define dout_prefix *_dout << "Entry: "
+#define dout_prefix *_dout << "Entry: " << this << " "
 
 namespace journal {
 

src/journal/FutureImpl.cc

@@ -10,7 +10,7 @@ FutureImpl::FutureImpl(uint64_t tag_tid, uint64_t entry_tid,
                        uint64_t commit_tid)
   : RefCountedObject(NULL, 0), m_tag_tid(tag_tid), m_entry_tid(entry_tid),
     m_commit_tid(commit_tid),
-    m_lock(utils::unique_lock_name("FutureImpl::m_lock", this)), m_safe(false),
+    m_lock("FutureImpl::m_lock", false, false), m_safe(false),
     m_consistent(false), m_return_value(0), m_flush_state(FLUSH_STATE_NONE),
     m_consistent_ack(this) {
 }
@@ -27,36 +27,51 @@ void FutureImpl::init(const FutureImplPtr &prev_future) {
 }
 
 void FutureImpl::flush(Context *on_safe) {
+
   bool complete;
-  FlushHandlerPtr flush_handler;
+  FlushHandlers flush_handlers;
+  FutureImplPtr prev_future;
   {
     Mutex::Locker locker(m_lock);
     complete = (m_safe && m_consistent);
     if (!complete) {
-      if (on_safe != NULL) {
+      if (on_safe != nullptr) {
         m_contexts.push_back(on_safe);
       }
 
-      if (m_flush_state == FLUSH_STATE_NONE) {
-        m_flush_state = FLUSH_STATE_REQUESTED;
-        flush_handler = m_flush_handler;
-
-        // walk the chain backwards up to <splay width> futures
-        if (m_prev_future) {
-          m_prev_future->flush();
-        }
-      }
+      prev_future = prepare_flush(&flush_handlers);
     }
   }
 
+  // instruct prior futures to flush as well
+  while (prev_future) {
+    Mutex::Locker locker(prev_future->m_lock);
+    prev_future = prev_future->prepare_flush(&flush_handlers);
+  }
+
   if (complete && on_safe != NULL) {
     on_safe->complete(m_return_value);
-  } else if (flush_handler) {
+  } else if (!flush_handlers.empty()) {
     // attached to journal object -- instruct it to flush all entries through
     // this one.  possible to become detached while lock is released, so flush
     // will be re-requested by the object if it doesn't own the future
-    flush_handler->flush(this);
+    for (auto &pair : flush_handlers) {
+      pair.first->flush(pair.second);
+    }
+  }
+}
+
+FutureImplPtr FutureImpl::prepare_flush(FlushHandlers *flush_handlers) {
+  assert(m_lock.is_locked());
+
+  if (m_flush_state == FLUSH_STATE_NONE) {
+    m_flush_state = FLUSH_STATE_REQUESTED;
+
+    if (m_flush_handler && flush_handlers->count(m_flush_handler) == 0) {
+      flush_handlers->insert({m_flush_handler, this});
+    }
   }
+  return m_prev_future;
 }
 
 void FutureImpl::wait(Context *on_safe) {

src/journal/FutureImpl.h

@@ -9,6 +9,7 @@
 #include "common/RefCountedObj.h"
 #include "journal/Future.h"
 #include <list>
+#include <map>
 #include <boost/noncopyable.hpp>
 #include <boost/intrusive_ptr.hpp>
 #include "include/assert.h"
@@ -76,6 +77,7 @@ class FutureImpl : public RefCountedObject, boost::noncopyable {
 private:
   friend std::ostream &operator<<(std::ostream &, const FutureImpl &);
 
+  typedef std::map<FlushHandlerPtr, FutureImplPtr> FlushHandlers;
   typedef std::list<Context *> Contexts;
 
   enum FlushState {
@@ -110,6 +112,8 @@ class FutureImpl : public RefCountedObject, boost::noncopyable {
   C_ConsistentAck m_consistent_ack;
   Contexts m_contexts;
 
+  FutureImplPtr prepare_flush(FlushHandlers *flush_handlers);
+
   void consistent(int r);
   void finish_unlock();
 };