--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_JOURNAL_JOURNAL_METADATA_H
+#define CEPH_JOURNAL_JOURNAL_METADATA_H
+
+#include "include/int_types.h"
+#include "include/Context.h"
+#include "include/rados/librados.hpp"
+#include "common/AsyncOpTracker.h"
+#include "common/Cond.h"
+#include "common/Mutex.h"
+#include "common/RefCountedObj.h"
+#include "common/WorkQueue.h"
+#include "cls/journal/cls_journal_types.h"
+#include "journal/JournalMetadataListener.h"
+#include "journal/Settings.h"
+#include <boost/intrusive_ptr.hpp>
+#include <boost/noncopyable.hpp>
+#include <boost/optional.hpp>
+#include <functional>
+#include <list>
+#include <map>
+#include <string>
+#include "include/assert.h"
+
+class SafeTimer;
+
+namespace journal {
+
+class JournalMetadata;
+typedef boost::intrusive_ptr<JournalMetadata> JournalMetadataPtr;
+
+class JournalMetadata : public RefCountedObject, boost::noncopyable {
+public:
+ typedef std::function<Context*()> CreateContext;
+ typedef cls::journal::ObjectPosition ObjectPosition;
+ typedef cls::journal::ObjectPositions ObjectPositions;
+ typedef cls::journal::ObjectSetPosition ObjectSetPosition;
+ typedef cls::journal::Client Client;
+ typedef cls::journal::Tag Tag;
+
+ typedef std::set<Client> RegisteredClients;
+ typedef std::list<Tag> Tags;
+
+ JournalMetadata(ContextWQ *work_queue, SafeTimer *timer, Mutex *timer_lock,
+ librados::IoCtx &ioctx, const std::string &oid,
+ const std::string &client_id, const Settings &settings);
+ ~JournalMetadata() override;
+
+ void init(Context *on_init);
+ void shut_down(Context *on_finish);
+
+ bool is_initialized() const { return m_initialized; }
+
+ void get_immutable_metadata(uint8_t *order, uint8_t *splay_width,
+ int64_t *pool_id, Context *on_finish);
+
+ void get_mutable_metadata(uint64_t *minimum_set, uint64_t *active_set,
+ RegisteredClients *clients, Context *on_finish);
+
+ void add_listener(JournalMetadataListener *listener);
+ void remove_listener(JournalMetadataListener *listener);
+
+ void register_client(const bufferlist &data, Context *on_finish);
+ void update_client(const bufferlist &data, Context *on_finish);
+ void unregister_client(Context *on_finish);
+ void get_client(const std::string &client_id, cls::journal::Client *client,
+ Context *on_finish);
+
+ void allocate_tag(uint64_t tag_class, const bufferlist &data,
+ Tag *tag, Context *on_finish);
+ void get_tag(uint64_t tag_tid, Tag *tag, Context *on_finish);
+ void get_tags(uint64_t start_after_tag_tid,
+ const boost::optional<uint64_t> &tag_class, Tags *tags,
+ Context *on_finish);
+
+ inline const Settings &get_settings() const {
+ return m_settings;
+ }
+ inline const std::string &get_client_id() const {
+ return m_client_id;
+ }
+ inline uint8_t get_order() const {
+ return m_order;
+ }
+ inline uint64_t get_object_size() const {
+ return 1 << m_order;
+ }
+ inline uint8_t get_splay_width() const {
+ return m_splay_width;
+ }
+ inline int64_t get_pool_id() const {
+ return m_pool_id;
+ }
+
+ inline void queue(Context *on_finish, int r) {
+ m_work_queue->queue(on_finish, r);
+ }
+
+ inline ContextWQ *get_work_queue() {
+ return m_work_queue;
+ }
+
+ inline SafeTimer &get_timer() {
+ return *m_timer;
+ }
+ inline Mutex &get_timer_lock() {
+ return *m_timer_lock;
+ }
+
+ void set_minimum_set(uint64_t object_set);
+ inline uint64_t get_minimum_set() const {
+ Mutex::Locker locker(m_lock);
+ return m_minimum_set;
+ }
+
+ int set_active_set(uint64_t object_set);
+ void set_active_set(uint64_t object_set, Context *on_finish);
+ inline uint64_t get_active_set() const {
+ Mutex::Locker locker(m_lock);
+ return m_active_set;
+ }
+
+ void assert_active_tag(uint64_t tag_tid, Context *on_finish);
+
+ void flush_commit_position();
+ void flush_commit_position(Context *on_safe);
+ void get_commit_position(ObjectSetPosition *commit_position) const {
+ Mutex::Locker locker(m_lock);
+ *commit_position = m_client.commit_position;
+ }
+
+ void get_registered_clients(RegisteredClients *registered_clients) {
+ Mutex::Locker locker(m_lock);
+ *registered_clients = m_registered_clients;
+ }
+
+ inline uint64_t allocate_entry_tid(uint64_t tag_tid) {
+ Mutex::Locker locker(m_lock);
+ return m_allocated_entry_tids[tag_tid]++;
+ }
+ void reserve_entry_tid(uint64_t tag_tid, uint64_t entry_tid);
+ bool get_last_allocated_entry_tid(uint64_t tag_tid, uint64_t *entry_tid) const;
+
+ uint64_t allocate_commit_tid(uint64_t object_num, uint64_t tag_tid,
+ uint64_t entry_tid);
+ void overflow_commit_tid(uint64_t commit_tid, uint64_t object_num);
+ void get_commit_entry(uint64_t commit_tid, uint64_t *object_num,
+ uint64_t *tag_tid, uint64_t *entry_tid);
+ void committed(uint64_t commit_tid, const CreateContext &create_context);
+
+ void notify_update();
+ void async_notify_update(Context *on_safe);
+
+ void wait_for_ops();
+
+private:
+ typedef std::map<uint64_t, uint64_t> AllocatedEntryTids;
+ typedef std::list<JournalMetadataListener*> Listeners;
+
+ struct CommitEntry {
+ uint64_t object_num;
+ uint64_t tag_tid;
+ uint64_t entry_tid;
+ bool committed;
+
+ CommitEntry() : object_num(0), tag_tid(0), entry_tid(0), committed(false) {
+ }
+ CommitEntry(uint64_t _object_num, uint64_t _tag_tid, uint64_t _entry_tid)
+ : object_num(_object_num), tag_tid(_tag_tid), entry_tid(_entry_tid),
+ committed(false) {
+ }
+ };
+ typedef std::map<uint64_t, CommitEntry> CommitTids;
+
+ struct C_WatchCtx : public librados::WatchCtx2 {
+ JournalMetadata *journal_metadata;
+
+ C_WatchCtx(JournalMetadata *_journal_metadata)
+ : journal_metadata(_journal_metadata) {}
+
+ void handle_notify(uint64_t notify_id, uint64_t cookie,
+ uint64_t notifier_id, bufferlist& bl) override {
+ journal_metadata->handle_watch_notify(notify_id, cookie);
+ }
+ void handle_error(uint64_t cookie, int err) override {
+ journal_metadata->handle_watch_error(err);
+ }
+ };
+
+ struct C_WatchReset : public Context {
+ JournalMetadata *journal_metadata;
+
+ C_WatchReset(JournalMetadata *_journal_metadata)
+ : journal_metadata(_journal_metadata) {
+ journal_metadata->m_async_op_tracker.start_op();
+ }
+ ~C_WatchReset() override {
+ journal_metadata->m_async_op_tracker.finish_op();
+ }
+ void finish(int r) override {
+ journal_metadata->handle_watch_reset();
+ }
+ };
+
+ struct C_CommitPositionTask : public Context {
+ JournalMetadata *journal_metadata;
+
+ C_CommitPositionTask(JournalMetadata *_journal_metadata)
+ : journal_metadata(_journal_metadata) {
+ journal_metadata->m_async_op_tracker.start_op();
+ }
+ ~C_CommitPositionTask() override {
+ journal_metadata->m_async_op_tracker.finish_op();
+ }
+ void finish(int r) override {
+ Mutex::Locker locker(journal_metadata->m_lock);
+ journal_metadata->handle_commit_position_task();
+ };
+ };
+
+ struct C_AioNotify : public Context {
+ JournalMetadata* journal_metadata;
+ Context *on_safe;
+
+ C_AioNotify(JournalMetadata *_journal_metadata, Context *_on_safe)
+ : journal_metadata(_journal_metadata), on_safe(_on_safe) {
+ journal_metadata->m_async_op_tracker.start_op();
+ }
+ ~C_AioNotify() override {
+ journal_metadata->m_async_op_tracker.finish_op();
+ }
+ void finish(int r) override {
+ journal_metadata->handle_notified(r);
+ if (on_safe != nullptr) {
+ on_safe->complete(0);
+ }
+ }
+ };
+
+ struct C_NotifyUpdate : public Context {
+ JournalMetadata* journal_metadata;
+ Context *on_safe;
+
+ C_NotifyUpdate(JournalMetadata *_journal_metadata, Context *_on_safe = NULL)
+ : journal_metadata(_journal_metadata), on_safe(_on_safe) {
+ journal_metadata->m_async_op_tracker.start_op();
+ }
+ ~C_NotifyUpdate() override {
+ journal_metadata->m_async_op_tracker.finish_op();
+ }
+ void finish(int r) override {
+ if (r == 0) {
+ journal_metadata->async_notify_update(on_safe);
+ return;
+ }
+ if (on_safe != NULL) {
+ on_safe->complete(r);
+ }
+ }
+ };
+
+ struct C_ImmutableMetadata : public Context {
+ JournalMetadata* journal_metadata;
+ Context *on_finish;
+
+ C_ImmutableMetadata(JournalMetadata *_journal_metadata, Context *_on_finish)
+ : journal_metadata(_journal_metadata), on_finish(_on_finish) {
+ Mutex::Locker locker(journal_metadata->m_lock);
+ journal_metadata->m_async_op_tracker.start_op();
+ }
+ ~C_ImmutableMetadata() override {
+ journal_metadata->m_async_op_tracker.finish_op();
+ }
+ void finish(int r) override {
+ journal_metadata->handle_immutable_metadata(r, on_finish);
+ }
+ };
+
+ struct C_Refresh : public Context {
+ JournalMetadata* journal_metadata;
+ uint64_t minimum_set;
+ uint64_t active_set;
+ RegisteredClients registered_clients;
+ Context *on_finish;
+
+ C_Refresh(JournalMetadata *_journal_metadata, Context *_on_finish)
+ : journal_metadata(_journal_metadata), minimum_set(0), active_set(0),
+ on_finish(_on_finish) {
+ Mutex::Locker locker(journal_metadata->m_lock);
+ journal_metadata->m_async_op_tracker.start_op();
+ }
+ ~C_Refresh() override {
+ journal_metadata->m_async_op_tracker.finish_op();
+ }
+ void finish(int r) override {
+ journal_metadata->handle_refresh_complete(this, r);
+ }
+ };
+
+ librados::IoCtx m_ioctx;
+ CephContext *m_cct;
+ std::string m_oid;
+ std::string m_client_id;
+ Settings m_settings;
+
+ uint8_t m_order;
+ uint8_t m_splay_width;
+ int64_t m_pool_id;
+ bool m_initialized;
+
+ ContextWQ *m_work_queue;
+ SafeTimer *m_timer;
+ Mutex *m_timer_lock;
+
+ mutable Mutex m_lock;
+
+ uint64_t m_commit_tid;
+ CommitTids m_pending_commit_tids;
+
+ Listeners m_listeners;
+
+ C_WatchCtx m_watch_ctx;
+ uint64_t m_watch_handle;
+
+ uint64_t m_minimum_set;
+ uint64_t m_active_set;
+ RegisteredClients m_registered_clients;
+ Client m_client;
+
+ AllocatedEntryTids m_allocated_entry_tids;
+
+ size_t m_update_notifications;
+ Cond m_update_cond;
+
+ uint64_t m_commit_position_tid = 0;
+ ObjectSetPosition m_commit_position;
+ Context *m_commit_position_ctx;
+ Context *m_commit_position_task_ctx;
+
+ AsyncOpTracker m_async_op_tracker;
+
+ void handle_immutable_metadata(int r, Context *on_init);
+
+ void refresh(Context *on_finish);
+ void handle_refresh_complete(C_Refresh *refresh, int r);
+
+ void cancel_commit_task();
+ void schedule_commit_task();
+ void handle_commit_position_task();
+
+ void schedule_watch_reset();
+ void handle_watch_reset();
+ void handle_watch_notify(uint64_t notify_id, uint64_t cookie);
+ void handle_watch_error(int err);
+ void handle_notified(int r);
+
+ Context *schedule_laggy_clients_disconnect(Context *on_finish);
+
+ friend std::ostream &operator<<(std::ostream &os,
+ const JournalMetadata &journal_metadata);
+};
+
+std::ostream &operator<<(std::ostream &os,
+ const JournalMetadata::RegisteredClients &clients);
+
+std::ostream &operator<<(std::ostream &os,
+ const JournalMetadata &journal_metadata);
+
+} // namespace journal
+
+#endif // CEPH_JOURNAL_JOURNAL_METADATA_H
Code Review / stor4nfv.git / blobdiff