1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 #ifndef CEPH_LIBRBD_JOURNAL_H
5 #define CEPH_LIBRBD_JOURNAL_H
7 #include "include/int_types.h"
8 #include "include/Context.h"
9 #include "include/interval_set.h"
10 #include "common/Cond.h"
11 #include "common/Mutex.h"
12 #include "common/Cond.h"
13 #include "common/WorkQueue.h"
14 #include "journal/Future.h"
15 #include "journal/JournalMetadataListener.h"
16 #include "journal/ReplayEntry.h"
17 #include "journal/ReplayHandler.h"
18 #include "librbd/Utils.h"
19 #include "librbd/journal/Types.h"
20 #include "librbd/journal/TypeTraits.h"
26 #include <unordered_map>
40 namespace io { struct ObjectRequestHandle; }
41 namespace journal { template <typename> class Replay; }
43 template <typename ImageCtxT = ImageCtx>
52 * UNINITIALIZED ---> INITIALIZING ---> REPLAYING ------> FLUSHING ---> READY
55 * | * . | (error) * . . . . . . . * |
58 * | * . | FLUSHING_RESTART . * |
62 * | * . | RESTARTING < * * * * * STOPPING
65 * | * * * * * * . \-------------/ . |
67 * | * . . . . . . . . . . . . . . . . |
70 * | CLOSED <----- CLOSING <---------------------------------------/
81 STATE_FLUSHING_RESTART,
82 STATE_RESTARTING_REPLAY,
83 STATE_FLUSHING_REPLAY,
90 static const std::string IMAGE_CLIENT_ID;
91 static const std::string LOCAL_MIRROR_UUID;
92 static const std::string ORPHAN_MIRROR_UUID;
94 typedef std::list<io::ObjectRequestHandle *> IOObjectRequests;
96 Journal(ImageCtxT &image_ctx);
99 static bool is_journal_supported(ImageCtxT &image_ctx);
100 static int create(librados::IoCtx &io_ctx, const std::string &image_id,
101 uint8_t order, uint8_t splay_width,
102 const std::string &object_pool);
103 static int remove(librados::IoCtx &io_ctx, const std::string &image_id);
104 static int reset(librados::IoCtx &io_ctx, const std::string &image_id);
106 static void is_tag_owner(ImageCtxT *image_ctx, bool *is_tag_owner,
108 static void is_tag_owner(librados::IoCtx& io_ctx, std::string& image_id,
109 bool *is_tag_owner, ContextWQ *op_work_queue,
111 static void get_tag_owner(librados::IoCtx& io_ctx, std::string& image_id,
112 std::string *mirror_uuid,
113 ContextWQ *op_work_queue, Context *on_finish);
114 static int request_resync(ImageCtxT *image_ctx);
115 static void promote(ImageCtxT *image_ctx, Context *on_finish);
116 static void demote(ImageCtxT *image_ctx, Context *on_finish);
118 bool is_journal_ready() const;
119 bool is_journal_replaying() const;
120 bool is_journal_appending() const;
122 void wait_for_journal_ready(Context *on_ready);
124 void open(Context *on_finish);
125 void close(Context *on_finish);
127 bool is_tag_owner() const;
128 uint64_t get_tag_tid() const;
129 journal::TagData get_tag_data() const;
131 void allocate_local_tag(Context *on_finish);
132 void allocate_tag(const std::string &mirror_uuid,
133 const journal::TagPredecessor &predecessor,
136 void flush_commit_position(Context *on_finish);
138 uint64_t append_write_event(uint64_t offset, size_t length,
139 const bufferlist &bl,
140 const IOObjectRequests &requests,
142 uint64_t append_io_event(journal::EventEntry &&event_entry,
143 const IOObjectRequests &requests,
144 uint64_t offset, size_t length,
146 void commit_io_event(uint64_t tid, int r);
147 void commit_io_event_extent(uint64_t tid, uint64_t offset, uint64_t length,
150 void append_op_event(uint64_t op_tid, journal::EventEntry &&event_entry,
152 void commit_op_event(uint64_t tid, int r, Context *on_safe);
153 void replay_op_ready(uint64_t op_tid, Context *on_resume);
155 void flush_event(uint64_t tid, Context *on_safe);
156 void wait_event(uint64_t tid, Context *on_safe);
158 uint64_t allocate_op_tid() {
159 uint64_t op_tid = ++m_op_tid;
164 void start_external_replay(journal::Replay<ImageCtxT> **journal_replay,
166 void stop_external_replay();
168 void add_listener(journal::Listener *listener);
169 void remove_listener(journal::Listener *listener);
171 int is_resync_requested(bool *do_resync);
173 inline ContextWQ *get_work_queue() {
178 ImageCtxT &m_image_ctx;
180 // mock unit testing support
181 typedef journal::TypeTraits<ImageCtxT> TypeTraits;
182 typedef typename TypeTraits::Journaler Journaler;
183 typedef typename TypeTraits::Future Future;
184 typedef typename TypeTraits::ReplayEntry ReplayEntry;
186 typedef std::list<bufferlist> Bufferlists;
187 typedef std::list<Context *> Contexts;
188 typedef std::list<Future> Futures;
189 typedef interval_set<uint64_t> ExtentInterval;
193 IOObjectRequests aio_object_requests;
194 Contexts on_safe_contexts;
195 ExtentInterval pending_extents;
196 bool committed_io = false;
202 Event(const Futures &_futures, const IOObjectRequests &_requests,
203 uint64_t offset, size_t length)
204 : futures(_futures), aio_object_requests(_requests) {
206 pending_extents.insert(offset, length);
211 typedef std::unordered_map<uint64_t, Event> Events;
212 typedef std::unordered_map<uint64_t, Future> TidToFutures;
214 struct C_IOEventSafe : public Context {
218 C_IOEventSafe(Journal *_journal, uint64_t _tid)
219 : journal(_journal), tid(_tid) {
222 void finish(int r) override {
223 journal->handle_io_event_safe(r, tid);
227 struct C_OpEventSafe : public Context {
230 Future op_start_future;
231 Future op_finish_future;
234 C_OpEventSafe(Journal *journal, uint64_t tid, const Future &op_start_future,
235 const Future &op_finish_future, Context *on_safe)
236 : journal(journal), tid(tid), op_start_future(op_start_future),
237 op_finish_future(op_finish_future), on_safe(on_safe) {
240 void finish(int r) override {
241 journal->handle_op_event_safe(r, tid, op_start_future, op_finish_future,
246 struct C_ReplayProcessSafe : public Context {
248 ReplayEntry replay_entry;
250 C_ReplayProcessSafe(Journal *journal, ReplayEntry &&replay_entry) :
251 journal(journal), replay_entry(std::move(replay_entry)) {
253 void finish(int r) override {
254 journal->handle_replay_process_safe(replay_entry, r);
258 struct ReplayHandler : public ::journal::ReplayHandler {
260 ReplayHandler(Journal *_journal) : journal(_journal) {
263 void get() override {
266 void put() override {
270 void handle_entries_available() override {
271 journal->handle_replay_ready();
273 void handle_complete(int r) override {
274 journal->handle_replay_complete(r);
278 ContextWQ *m_work_queue = nullptr;
279 SafeTimer *m_timer = nullptr;
280 Mutex *m_timer_lock = nullptr;
282 Journaler *m_journaler;
283 mutable Mutex m_lock;
285 uint64_t m_max_append_size = 0;
286 uint64_t m_tag_class = 0;
287 uint64_t m_tag_tid = 0;
288 journal::ImageClientMeta m_client_meta;
289 journal::TagData m_tag_data;
292 Contexts m_wait_for_state_contexts;
294 ReplayHandler m_replay_handler;
295 bool m_close_pending;
298 uint64_t m_event_tid;
301 std::atomic<uint64_t> m_op_tid = { 0 };
302 TidToFutures m_op_futures;
304 bool m_processing_entry = false;
305 bool m_blocking_writes;
307 journal::Replay<ImageCtxT> *m_journal_replay;
309 util::AsyncOpTracker m_async_journal_op_tracker;
311 struct MetadataListener : public ::journal::JournalMetadataListener {
312 Journal<ImageCtxT> *journal;
314 MetadataListener(Journal<ImageCtxT> *journal) : journal(journal) { }
316 void handle_update(::journal::JournalMetadata *) override {
317 FunctionContext *ctx = new FunctionContext([this](int r) {
318 journal->handle_metadata_updated();
320 journal->m_work_queue->queue(ctx, 0);
322 } m_metadata_listener;
324 typedef std::set<journal::Listener *> Listeners;
325 Listeners m_listeners;
326 Cond m_listener_cond;
327 bool m_listener_notify = false;
329 uint64_t m_refresh_sequence = 0;
331 bool is_journal_replaying(const Mutex &) const;
332 bool is_tag_owner(const Mutex &) const;
334 uint64_t append_io_events(journal::EventType event_type,
335 const Bufferlists &bufferlists,
336 const IOObjectRequests &requests,
337 uint64_t offset, size_t length, bool flush_entry);
338 Future wait_event(Mutex &lock, uint64_t tid, Context *on_safe);
340 void create_journaler();
341 void destroy_journaler(int r);
342 void recreate_journaler(int r);
344 void complete_event(typename Events::iterator it, int r);
348 void handle_open(int r);
350 void handle_replay_ready();
351 void handle_replay_complete(int r);
352 void handle_replay_process_ready(int r);
353 void handle_replay_process_safe(ReplayEntry replay_entry, int r);
355 void handle_start_external_replay(int r,
356 journal::Replay<ImageCtxT> **journal_replay,
359 void handle_flushing_restart(int r);
360 void handle_flushing_replay();
362 void handle_recording_stopped(int r);
364 void handle_journal_destroyed(int r);
366 void handle_io_event_safe(int r, uint64_t tid);
367 void handle_op_event_safe(int r, uint64_t tid, const Future &op_start_future,
368 const Future &op_finish_future, Context *on_safe);
370 void stop_recording();
372 void transition_state(State state, int r);
374 bool is_steady_state() const;
375 void wait_for_steady_state(Context *on_state);
377 int check_resync_requested(bool *do_resync);
379 void handle_metadata_updated();
380 void handle_refresh_metadata(uint64_t refresh_sequence, uint64_t tag_tid,
381 journal::TagData tag_data, int r);
385 } // namespace librbd
387 extern template class librbd::Journal<librbd::ImageCtx>;
389 #endif // CEPH_LIBRBD_JOURNAL_H