+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-/*
- * Ceph - scalable distributed file system
- *
- * Copyright (C) 2013 Inktank Storage, Inc.
- *
- * This is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License version 2.1, as published by the Free Software
- * Foundation. See file COPYING.
- *
- */
-
-#ifndef REPBACKEND_H
-#define REPBACKEND_H
-
-#include "OSD.h"
-#include "PGBackend.h"
-#include "include/memory.h"
-
-struct C_ReplicatedBackend_OnPullComplete;
-class ReplicatedBackend : public PGBackend {
- struct RPGHandle : public PGBackend::RecoveryHandle {
- map<pg_shard_t, vector<PushOp> > pushes;
- map<pg_shard_t, vector<PullOp> > pulls;
- };
- friend struct C_ReplicatedBackend_OnPullComplete;
-public:
- ReplicatedBackend(
- PGBackend::Listener *pg,
- coll_t coll,
- ObjectStore::CollectionHandle &ch,
- ObjectStore *store,
- CephContext *cct);
-
- /// @see PGBackend::open_recovery_op
- RPGHandle *_open_recovery_op() {
- return new RPGHandle();
- }
- PGBackend::RecoveryHandle *open_recovery_op() override {
- return _open_recovery_op();
- }
-
- /// @see PGBackend::run_recovery_op
- void run_recovery_op(
- PGBackend::RecoveryHandle *h,
- int priority) override;
-
- /// @see PGBackend::recover_object
- int recover_object(
- const hobject_t &hoid,
- eversion_t v,
- ObjectContextRef head,
- ObjectContextRef obc,
- RecoveryHandle *h
- ) override;
-
- void check_recovery_sources(const OSDMapRef& osdmap) override;
-
- bool can_handle_while_inactive(OpRequestRef op) override;
-
- /// @see PGBackend::handle_message
- bool _handle_message(
- OpRequestRef op
- ) override;
-
- void on_change() override;
- void clear_recovery_state() override;
- void on_flushed() override;
-
- class RPCRecPred : public IsPGRecoverablePredicate {
- public:
- bool operator()(const set<pg_shard_t> &have) const override {
- return !have.empty();
- }
- };
- IsPGRecoverablePredicate *get_is_recoverable_predicate() override {
- return new RPCRecPred;
- }
-
- class RPCReadPred : public IsPGReadablePredicate {
- pg_shard_t whoami;
- public:
- explicit RPCReadPred(pg_shard_t whoami) : whoami(whoami) {}
- bool operator()(const set<pg_shard_t> &have) const override {
- return have.count(whoami);
- }
- };
- IsPGReadablePredicate *get_is_readable_predicate() override {
- return new RPCReadPred(get_parent()->whoami_shard());
- }
-
- void dump_recovery_info(Formatter *f) const override {
- {
- f->open_array_section("pull_from_peer");
- for (map<pg_shard_t, set<hobject_t> >::const_iterator i = pull_from_peer.begin();
- i != pull_from_peer.end();
- ++i) {
- f->open_object_section("pulling_from");
- f->dump_stream("pull_from") << i->first;
- {
- f->open_array_section("pulls");
- for (set<hobject_t>::const_iterator j = i->second.begin();
- j != i->second.end();
- ++j) {
- f->open_object_section("pull_info");
- assert(pulling.count(*j));
- pulling.find(*j)->second.dump(f);
- f->close_section();
- }
- f->close_section();
- }
- f->close_section();
- }
- f->close_section();
- }
- {
- f->open_array_section("pushing");
- for (map<hobject_t, map<pg_shard_t, PushInfo>>::const_iterator i =
- pushing.begin();
- i != pushing.end();
- ++i) {
- f->open_object_section("object");
- f->dump_stream("pushing") << i->first;
- {
- f->open_array_section("pushing_to");
- for (map<pg_shard_t, PushInfo>::const_iterator j = i->second.begin();
- j != i->second.end();
- ++j) {
- f->open_object_section("push_progress");
- f->dump_stream("pushing_to") << j->first;
- {
- f->open_object_section("push_info");
- j->second.dump(f);
- f->close_section();
- }
- f->close_section();
- }
- f->close_section();
- }
- f->close_section();
- }
- f->close_section();
- }
- }
-
- int objects_read_sync(
- const hobject_t &hoid,
- uint64_t off,
- uint64_t len,
- uint32_t op_flags,
- bufferlist *bl) override;
-
- void objects_read_async(
- const hobject_t &hoid,
- const list<pair<boost::tuple<uint64_t, uint64_t, uint32_t>,
- pair<bufferlist*, Context*> > > &to_read,
- Context *on_complete,
- bool fast_read = false) override;
-
-private:
- // push
- struct PushInfo {
- ObjectRecoveryProgress recovery_progress;
- ObjectRecoveryInfo recovery_info;
- ObjectContextRef obc;
- object_stat_sum_t stat;
- ObcLockManager lock_manager;
-
- void dump(Formatter *f) const {
- {
- f->open_object_section("recovery_progress");
- recovery_progress.dump(f);
- f->close_section();
- }
- {
- f->open_object_section("recovery_info");
- recovery_info.dump(f);
- f->close_section();
- }
- }
- };
- map<hobject_t, map<pg_shard_t, PushInfo>> pushing;
-
- // pull
- struct PullInfo {
- pg_shard_t from;
- hobject_t soid;
- ObjectRecoveryProgress recovery_progress;
- ObjectRecoveryInfo recovery_info;
- ObjectContextRef head_ctx;
- ObjectContextRef obc;
- object_stat_sum_t stat;
- bool cache_dont_need;
- ObcLockManager lock_manager;
-
- void dump(Formatter *f) const {
- {
- f->open_object_section("recovery_progress");
- recovery_progress.dump(f);
- f->close_section();
- }
- {
- f->open_object_section("recovery_info");
- recovery_info.dump(f);
- f->close_section();
- }
- }
-
- bool is_complete() const {
- return recovery_progress.is_complete(recovery_info);
- }
- };
-
- map<hobject_t, PullInfo> pulling;
-
- // Reverse mapping from osd peer to objects beging pulled from that peer
- map<pg_shard_t, set<hobject_t> > pull_from_peer;
- void clear_pull(
- map<hobject_t, PullInfo>::iterator piter,
- bool clear_pull_from_peer = true);
- void clear_pull_from(
- map<hobject_t, PullInfo>::iterator piter);
-
- void _do_push(OpRequestRef op);
- void _do_pull_response(OpRequestRef op);
- void do_push(OpRequestRef op) {
- if (is_primary()) {
- _do_pull_response(op);
- } else {
- _do_push(op);
- }
- }
- void do_pull(OpRequestRef op);
- void do_push_reply(OpRequestRef op);
-
- bool handle_push_reply(pg_shard_t peer, const PushReplyOp &op, PushOp *reply);
- void handle_pull(pg_shard_t peer, PullOp &op, PushOp *reply);
-
- struct pull_complete_info {
- hobject_t hoid;
- object_stat_sum_t stat;
- };
- bool handle_pull_response(
- pg_shard_t from, const PushOp &op, PullOp *response,
- list<pull_complete_info> *to_continue,
- ObjectStore::Transaction *t);
- void handle_push(pg_shard_t from, const PushOp &op, PushReplyOp *response,
- ObjectStore::Transaction *t);
-
- static void trim_pushed_data(const interval_set<uint64_t> ©_subset,
- const interval_set<uint64_t> &intervals_received,
- bufferlist data_received,
- interval_set<uint64_t> *intervals_usable,
- bufferlist *data_usable);
- void _failed_pull(pg_shard_t from, const hobject_t &soid);
-
- void send_pushes(int prio, map<pg_shard_t, vector<PushOp> > &pushes);
- void prep_push_op_blank(const hobject_t& soid, PushOp *op);
- void send_pulls(
- int priority,
- map<pg_shard_t, vector<PullOp> > &pulls);
-
- int build_push_op(const ObjectRecoveryInfo &recovery_info,
- const ObjectRecoveryProgress &progress,
- ObjectRecoveryProgress *out_progress,
- PushOp *out_op,
- object_stat_sum_t *stat = 0,
- bool cache_dont_need = true);
- void submit_push_data(const ObjectRecoveryInfo &recovery_info,
- bool first,
- bool complete,
- bool cache_dont_need,
- const interval_set<uint64_t> &intervals_included,
- bufferlist data_included,
- bufferlist omap_header,
- const map<string, bufferlist> &attrs,
- const map<string, bufferlist> &omap_entries,
- ObjectStore::Transaction *t);
- void submit_push_complete(const ObjectRecoveryInfo &recovery_info,
- ObjectStore::Transaction *t);
-
- void calc_clone_subsets(
- SnapSet& snapset, const hobject_t& poid, const pg_missing_t& missing,
- const hobject_t &last_backfill,
- interval_set<uint64_t>& data_subset,
- map<hobject_t, interval_set<uint64_t>>& clone_subsets,
- ObcLockManager &lock_manager);
- void prepare_pull(
- eversion_t v,
- const hobject_t& soid,
- ObjectContextRef headctx,
- RPGHandle *h);
- int start_pushes(
- const hobject_t &soid,
- ObjectContextRef obj,
- RPGHandle *h);
- int prep_push_to_replica(
- ObjectContextRef obc, const hobject_t& soid, pg_shard_t peer,
- PushOp *pop, bool cache_dont_need = true);
- int prep_push(
- ObjectContextRef obc,
- const hobject_t& oid, pg_shard_t dest,
- PushOp *op,
- bool cache_dont_need);
- int prep_push(
- ObjectContextRef obc,
- const hobject_t& soid, pg_shard_t peer,
- eversion_t version,
- interval_set<uint64_t> &data_subset,
- map<hobject_t, interval_set<uint64_t>>& clone_subsets,
- PushOp *op,
- bool cache,
- ObcLockManager &&lock_manager);
- void calc_head_subsets(
- ObjectContextRef obc, SnapSet& snapset, const hobject_t& head,
- const pg_missing_t& missing,
- const hobject_t &last_backfill,
- interval_set<uint64_t>& data_subset,
- map<hobject_t, interval_set<uint64_t>>& clone_subsets,
- ObcLockManager &lock_manager);
- ObjectRecoveryInfo recalc_subsets(
- const ObjectRecoveryInfo& recovery_info,
- SnapSetContext *ssc,
- ObcLockManager &lock_manager);
-
- /**
- * Client IO
- */
- struct InProgressOp {
- ceph_tid_t tid;
- set<pg_shard_t> waiting_for_commit;
- set<pg_shard_t> waiting_for_applied;
- Context *on_commit;
- Context *on_applied;
- OpRequestRef op;
- eversion_t v;
- InProgressOp(
- ceph_tid_t tid, Context *on_commit, Context *on_applied,
- OpRequestRef op, eversion_t v)
- : tid(tid), on_commit(on_commit), on_applied(on_applied),
- op(op), v(v) {}
- bool done() const {
- return waiting_for_commit.empty() &&
- waiting_for_applied.empty();
- }
- };
- map<ceph_tid_t, InProgressOp> in_progress_ops;
-public:
- friend class C_OSD_OnOpCommit;
- friend class C_OSD_OnOpApplied;
-
- void call_write_ordered(std::function<void(void)> &&cb) override {
- // ReplicatedBackend submits writes inline in submit_transaction, so
- // we can just call the callback.
- cb();
- }
-
- void submit_transaction(
- const hobject_t &hoid,
- const object_stat_sum_t &delta_stats,
- const eversion_t &at_version,
- PGTransactionUPtr &&t,
- const eversion_t &trim_to,
- const eversion_t &roll_forward_to,
- const vector<pg_log_entry_t> &log_entries,
- boost::optional<pg_hit_set_history_t> &hset_history,
- Context *on_local_applied_sync,
- Context *on_all_applied,
- Context *on_all_commit,
- ceph_tid_t tid,
- osd_reqid_t reqid,
- OpRequestRef op
- ) override;
-
-private:
- Message * generate_subop(
- const hobject_t &soid,
- const eversion_t &at_version,
- ceph_tid_t tid,
- osd_reqid_t reqid,
- eversion_t pg_trim_to,
- eversion_t pg_roll_forward_to,
- hobject_t new_temp_oid,
- hobject_t discard_temp_oid,
- const vector<pg_log_entry_t> &log_entries,
- boost::optional<pg_hit_set_history_t> &hset_history,
- ObjectStore::Transaction &op_t,
- pg_shard_t peer,
- const pg_info_t &pinfo);
- void issue_op(
- const hobject_t &soid,
- const eversion_t &at_version,
- ceph_tid_t tid,
- osd_reqid_t reqid,
- eversion_t pg_trim_to,
- eversion_t pg_roll_forward_to,
- hobject_t new_temp_oid,
- hobject_t discard_temp_oid,
- const vector<pg_log_entry_t> &log_entries,
- boost::optional<pg_hit_set_history_t> &hset_history,
- InProgressOp *op,
- ObjectStore::Transaction &op_t);
- void op_applied(InProgressOp *op);
- void op_commit(InProgressOp *op);
- void do_repop_reply(OpRequestRef op);
- void do_repop(OpRequestRef op);
-
- struct RepModify {
- OpRequestRef op;
- bool applied, committed;
- int ackerosd;
- eversion_t last_complete;
- epoch_t epoch_started;
-
- ObjectStore::Transaction opt, localt;
-
- RepModify() : applied(false), committed(false), ackerosd(-1),
- epoch_started(0) {}
- };
- typedef ceph::shared_ptr<RepModify> RepModifyRef;
-
- struct C_OSD_RepModifyApply;
- struct C_OSD_RepModifyCommit;
-
- void repop_applied(RepModifyRef rm);
- void repop_commit(RepModifyRef rm);
- bool scrub_supported() override { return true; }
- bool auto_repair_supported() const override { return false; }
-
-
- void be_deep_scrub(
- const hobject_t &obj,
- uint32_t seed,
- ScrubMap::object &o,
- ThreadPool::TPHandle &handle) override;
- uint64_t be_get_ondisk_size(uint64_t logical_size) override { return logical_size; }
-};
-
-#endif
Code Review / stor4nfv.git / blobdiff