1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 #ifndef CEPH_RBD_MIRROR_POOL_REPLAYER_H
5 #define CEPH_RBD_MIRROR_POOL_REPLAYER_H
7 #include "common/AsyncOpTracker.h"
8 #include "common/Cond.h"
9 #include "common/Mutex.h"
10 #include "common/WorkQueue.h"
11 #include "include/rados/librados.hpp"
13 #include "ClusterWatcher.h"
14 #include "LeaderWatcher.h"
15 #include "PoolWatcher.h"
16 #include "ImageDeleter.h"
18 #include "tools/rbd_mirror/service_daemon/Types.h"
26 class AdminSocketHook;
28 namespace librbd { class ImageCtx; }
33 template <typename> class InstanceReplayer;
34 template <typename> class InstanceWatcher;
35 template <typename> class ServiceDaemon;
36 template <typename> struct Threads;
39 * Controls mirroring for a single remote cluster.
43 PoolReplayer(Threads<librbd::ImageCtx> *threads,
44 ServiceDaemon<librbd::ImageCtx>* service_daemon,
45 ImageDeleter<>* image_deleter,
46 int64_t local_pool_id, const peer_t &peer,
47 const std::vector<const char*> &args);
49 PoolReplayer(const PoolReplayer&) = delete;
50 PoolReplayer& operator=(const PoolReplayer&) = delete;
52 bool is_blacklisted() const;
53 bool is_leader() const;
54 bool is_running() const;
61 void print_status(Formatter *f, stringstream *ss);
63 void stop(bool manual);
66 void release_leader();
69 struct PoolWatcherListener : public PoolWatcher<>::Listener {
70 PoolReplayer *pool_replayer;
73 PoolWatcherListener(PoolReplayer *pool_replayer, bool local)
74 : pool_replayer(pool_replayer), local(local) {
77 void handle_update(const std::string &mirror_uuid,
78 ImageIds &&added_image_ids,
79 ImageIds &&removed_image_ids) override {
80 pool_replayer->handle_update((local ? "" : mirror_uuid),
81 std::move(added_image_ids),
82 std::move(removed_image_ids));
86 void handle_update(const std::string &mirror_uuid,
87 ImageIds &&added_image_ids,
88 ImageIds &&removed_image_ids);
90 int init_rados(const std::string &cluster_name,
91 const std::string &client_name,
92 const std::string &description, RadosRef *rados_ref,
93 bool strip_cluster_overrides);
95 void handle_post_acquire_leader(Context *on_finish);
96 void handle_pre_release_leader(Context *on_finish);
98 void init_local_pool_watcher(Context *on_finish);
99 void handle_init_local_pool_watcher(int r, Context *on_finish);
101 void init_remote_pool_watcher(Context *on_finish);
103 void shut_down_pool_watchers(Context *on_finish);
104 void handle_shut_down_pool_watchers(int r, Context *on_finish);
106 void wait_for_update_ops(Context *on_finish);
107 void handle_wait_for_update_ops(int r, Context *on_finish);
109 void handle_update_leader(const std::string &leader_instance_id);
111 Threads<librbd::ImageCtx> *m_threads;
112 ServiceDaemon<librbd::ImageCtx>* m_service_daemon;
113 ImageDeleter<>* m_image_deleter;
114 int64_t m_local_pool_id = -1;
116 std::vector<const char*> m_args;
118 mutable Mutex m_lock;
120 std::atomic<bool> m_stopping = { false };
121 bool m_manual_stop = false;
122 bool m_blacklisted = false;
124 RadosRef m_local_rados;
125 RadosRef m_remote_rados;
127 librados::IoCtx m_local_io_ctx;
128 librados::IoCtx m_remote_io_ctx;
130 PoolWatcherListener m_local_pool_watcher_listener;
131 std::unique_ptr<PoolWatcher<> > m_local_pool_watcher;
133 PoolWatcherListener m_remote_pool_watcher_listener;
134 std::unique_ptr<PoolWatcher<> > m_remote_pool_watcher;
136 std::unique_ptr<InstanceReplayer<librbd::ImageCtx>> m_instance_replayer;
138 std::string m_asok_hook_name;
139 AdminSocketHook *m_asok_hook = nullptr;
141 service_daemon::CalloutId m_callout_id = service_daemon::CALLOUT_ID_NONE;
143 class PoolReplayerThread : public Thread {
144 PoolReplayer *m_pool_replayer;
146 PoolReplayerThread(PoolReplayer *pool_replayer)
147 : m_pool_replayer(pool_replayer) {
149 void *entry() override {
150 m_pool_replayer->run();
153 } m_pool_replayer_thread;
155 class LeaderListener : public LeaderWatcher<>::Listener {
157 LeaderListener(PoolReplayer *pool_replayer)
158 : m_pool_replayer(pool_replayer) {
162 void post_acquire_handler(Context *on_finish) override {
163 m_pool_replayer->handle_post_acquire_leader(on_finish);
166 void pre_release_handler(Context *on_finish) override {
167 m_pool_replayer->handle_pre_release_leader(on_finish);
170 void update_leader_handler(
171 const std::string &leader_instance_id) override {
172 m_pool_replayer->handle_update_leader(leader_instance_id);
176 PoolReplayer *m_pool_replayer;
179 std::unique_ptr<LeaderWatcher<> > m_leader_watcher;
180 std::unique_ptr<InstanceWatcher<librbd::ImageCtx> > m_instance_watcher;
181 AsyncOpTracker m_update_op_tracker;
184 } // namespace mirror
187 #endif // CEPH_RBD_MIRROR_POOL_REPLAYER_H