Merge pull request #35440 from smithfarm/wip-45885-octopus

octopus: rbd-mirror: stop local journal replayer first during shut down

Reviewed-by: Jason Dillaman <dillaman@redhat.com>
Reviewed-by: Mykola Golub <mgolub@mirantis.com>

diff --cc src/test/rbd_mirror/image_replayer/journal/test_mock_Replayer.cc
Simple merge

diff --cc src/tools/rbd_mirror/image_replayer/journal/Replayer.cc
index af6802194424e9025a0b8a09c49b2c5ec4bf2a6c,3af886fdf3abd9f21acac3676746509e7dd67b30..60b9a7e59319ad6399adcceb675348aff519375c
--- 1/src/tools/rbd_mirror/image_replayer/journal/Replayer.cc
--- 2/src/tools/rbd_mirror/image_replayer/journal/Replayer.cc
+++ b/src/tools/rbd_mirror/image_replayer/journal/Replayer.cc
@@@ -869,14 -854,15 +893,16 @@@ void Replayer<I>::handle_replay_flush_s
  
  template <typename I>
  void Replayer<I>::handle_replay_flush(int r) {
 +  std::unique_lock locker{m_lock};
    dout(10) << "r=" << r << dendl;
+   m_flush_tracker.finish_op();
+ 
    if (r < 0) {
      derr << "replay flush encountered an error: " << cpp_strerror(r) << dendl;
 -    handle_replay_complete(r, "replay flush encountered an error");
 +    handle_replay_complete(locker, r, "replay flush encountered an error");
      m_event_replay_tracker.finish_op();
      return;
 -  } else if (is_replay_complete()) {
 +  } else if (is_replay_complete(locker)) {
      m_event_replay_tracker.finish_op();
      return;
    }

diff --cc src/tools/rbd_mirror/image_replayer/journal/Replayer.h
index ec6e44f1ae6495d92208713756c22c4964a1f2aa,b279aa9a297e19308cb2bfef72f04cb96313ad78..6b1f36d9c7fe9d9f699667fb1e13d1a6041aba78
--- 1/src/tools/rbd_mirror/image_replayer/journal/Replayer.h
--- 2/src/tools/rbd_mirror/image_replayer/journal/Replayer.h
+++ b/src/tools/rbd_mirror/image_replayer/journal/Replayer.h
@@@ -235,13 -240,14 +240,16 @@@ private
    void init_remote_journaler();
    void handle_init_remote_journaler(int r);
  
 -  void start_external_replay();
 +  void start_external_replay(std::unique_lock<ceph::mutex>& locker);
    void handle_start_external_replay(int r);
  
 +  bool add_local_journal_listener(std::unique_lock<ceph::mutex>& locker);
 +
    bool notify_init_complete(std::unique_lock<ceph::mutex>& locker);
  
+   void wait_for_flush();
+   void handle_wait_for_flush(int r);
+ 
    void shut_down_local_journal_replay();
    void handle_shut_down_local_journal_replay(int r);