+++ /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_PLAYER_H
-#define CEPH_JOURNAL_JOURNAL_PLAYER_H
-
-#include "include/int_types.h"
-#include "include/Context.h"
-#include "include/rados/librados.hpp"
-#include "common/AsyncOpTracker.h"
-#include "common/Mutex.h"
-#include "journal/JournalMetadata.h"
-#include "journal/ObjectPlayer.h"
-#include "cls/journal/cls_journal_types.h"
-#include <boost/none.hpp>
-#include <boost/optional.hpp>
-#include <map>
-
-class SafeTimer;
-
-namespace journal {
-
-class Entry;
-class ReplayHandler;
-
-class JournalPlayer {
-public:
- typedef cls::journal::ObjectPosition ObjectPosition;
- typedef cls::journal::ObjectPositions ObjectPositions;
- typedef cls::journal::ObjectSetPosition ObjectSetPosition;
-
- JournalPlayer(librados::IoCtx &ioctx, const std::string &object_oid_prefix,
- const JournalMetadataPtr& journal_metadata,
- ReplayHandler *replay_handler);
- ~JournalPlayer();
-
- void prefetch();
- void prefetch_and_watch(double interval);
- void shut_down(Context *on_finish);
-
- bool try_pop_front(Entry *entry, uint64_t *commit_tid);
-
-private:
- typedef std::set<uint8_t> PrefetchSplayOffsets;
- typedef std::map<uint8_t, ObjectPlayerPtr> SplayedObjectPlayers;
- typedef std::map<uint8_t, ObjectPosition> SplayedObjectPositions;
- typedef std::set<uint64_t> ObjectNumbers;
-
- enum State {
- STATE_INIT,
- STATE_PREFETCH,
- STATE_PLAYBACK,
- STATE_ERROR
- };
-
- enum WatchStep {
- WATCH_STEP_FETCH_CURRENT,
- WATCH_STEP_FETCH_FIRST,
- WATCH_STEP_ASSERT_ACTIVE
- };
-
- struct C_Fetch : public Context {
- JournalPlayer *player;
- uint64_t object_num;
- C_Fetch(JournalPlayer *p, uint64_t o) : player(p), object_num(o) {
- player->m_async_op_tracker.start_op();
- }
- ~C_Fetch() override {
- player->m_async_op_tracker.finish_op();
- }
- void finish(int r) override {
- player->handle_fetched(object_num, r);
- }
- };
-
- struct C_Watch : public Context {
- JournalPlayer *player;
- uint64_t object_num;
- C_Watch(JournalPlayer *player, uint64_t object_num)
- : player(player), object_num(object_num) {
- player->m_async_op_tracker.start_op();
- }
- ~C_Watch() override {
- player->m_async_op_tracker.finish_op();
- }
-
- void finish(int r) override {
- player->handle_watch(object_num, r);
- }
- };
-
- librados::IoCtx m_ioctx;
- CephContext *m_cct;
- std::string m_object_oid_prefix;
- JournalMetadataPtr m_journal_metadata;
-
- ReplayHandler *m_replay_handler;
-
- AsyncOpTracker m_async_op_tracker;
-
- mutable Mutex m_lock;
- State m_state;
- uint8_t m_splay_offset;
-
- bool m_watch_enabled;
- bool m_watch_scheduled;
- double m_watch_interval;
- WatchStep m_watch_step = WATCH_STEP_FETCH_CURRENT;
- bool m_watch_prune_active_tag = false;
-
- bool m_shut_down = false;
- bool m_handler_notified = false;
-
- ObjectNumbers m_fetch_object_numbers;
-
- PrefetchSplayOffsets m_prefetch_splay_offsets;
- SplayedObjectPlayers m_object_players;
-
- bool m_commit_position_valid = false;
- ObjectPosition m_commit_position;
- SplayedObjectPositions m_commit_positions;
- uint64_t m_active_set;
-
- boost::optional<uint64_t> m_active_tag_tid = boost::none;
- boost::optional<uint64_t> m_prune_tag_tid = boost::none;
-
- void advance_splay_object();
-
- bool is_object_set_ready() const;
- bool verify_playback_ready();
- void prune_tag(uint64_t tag_tid);
- void prune_active_tag(const boost::optional<uint64_t>& tag_tid);
-
- ObjectPlayerPtr get_object_player() const;
- ObjectPlayerPtr get_object_player(uint64_t object_number) const;
- bool remove_empty_object_player(const ObjectPlayerPtr &object_player);
-
- void process_state(uint64_t object_number, int r);
- int process_prefetch(uint64_t object_number);
- int process_playback(uint64_t object_number);
-
- void fetch(uint64_t object_num);
- void fetch(const ObjectPlayerPtr &object_player);
- void handle_fetched(uint64_t object_num, int r);
- void refetch(bool immediate);
-
- void schedule_watch(bool immediate);
- void handle_watch(uint64_t object_num, int r);
- void handle_watch_assert_active(int r);
-
- void notify_entries_available();
- void notify_complete(int r);
-};
-
-} // namespace journal
-
-#endif // CEPH_JOURNAL_JOURNAL_PLAYER_H