--- /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 ECBACKEND_H
+#define ECBACKEND_H
+
+#include <boost/intrusive/set.hpp>
+#include <boost/intrusive/list.hpp>
+
+#include "OSD.h"
+#include "PGBackend.h"
+#include "erasure-code/ErasureCodeInterface.h"
+#include "ECUtil.h"
+#include "ECTransaction.h"
+#include "ExtentCache.h"
+
+//forward declaration
+struct ECSubWrite;
+struct ECSubWriteReply;
+struct ECSubRead;
+struct ECSubReadReply;
+
+struct RecoveryMessages;
+class ECBackend : public PGBackend {
+public:
+ RecoveryHandle *open_recovery_op() override;
+
+ void run_recovery_op(
+ RecoveryHandle *h,
+ int priority
+ ) override;
+
+ int recover_object(
+ const hobject_t &hoid,
+ eversion_t v,
+ ObjectContextRef head,
+ ObjectContextRef obc,
+ RecoveryHandle *h
+ ) override;
+
+ bool _handle_message(
+ OpRequestRef op
+ ) override;
+ bool can_handle_while_inactive(
+ OpRequestRef op
+ ) override;
+ friend struct SubWriteApplied;
+ friend struct SubWriteCommitted;
+ void sub_write_applied(
+ ceph_tid_t tid,
+ eversion_t version,
+ const ZTracer::Trace &trace);
+ void sub_write_committed(
+ ceph_tid_t tid,
+ eversion_t version,
+ eversion_t last_complete,
+ const ZTracer::Trace &trace);
+ void handle_sub_write(
+ pg_shard_t from,
+ OpRequestRef msg,
+ ECSubWrite &op,
+ const ZTracer::Trace &trace,
+ Context *on_local_applied_sync = 0
+ );
+ void handle_sub_read(
+ pg_shard_t from,
+ const ECSubRead &op,
+ ECSubReadReply *reply,
+ const ZTracer::Trace &trace
+ );
+ void handle_sub_write_reply(
+ pg_shard_t from,
+ const ECSubWriteReply &op,
+ const ZTracer::Trace &trace
+ );
+ void handle_sub_read_reply(
+ pg_shard_t from,
+ ECSubReadReply &op,
+ RecoveryMessages *m,
+ const ZTracer::Trace &trace
+ );
+
+ /// @see ReadOp below
+ void check_recovery_sources(const OSDMapRef& osdmap) override;
+
+ void on_change() override;
+ void clear_recovery_state() override;
+
+ void on_flushed() override;
+
+ void dump_recovery_info(Formatter *f) const override;
+
+ void call_write_ordered(std::function<void(void)> &&cb) override;
+
+ 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;
+
+ int objects_read_sync(
+ const hobject_t &hoid,
+ uint64_t off,
+ uint64_t len,
+ uint32_t op_flags,
+ bufferlist *bl) override;
+
+ /**
+ * Async read mechanism
+ *
+ * Async reads use the same async read mechanism as does recovery.
+ * CallClientContexts is responsible for reconstructing the response
+ * buffer as well as for calling the callbacks.
+ *
+ * One tricky bit is that two reads may possibly not read from the same
+ * set of replicas. This could result in two reads completing in the
+ * wrong (from the interface user's point of view) order. Thus, we
+ * maintain a queue of in progress reads (@see in_progress_client_reads)
+ * to ensure that we always call the completion callback in order.
+ *
+ * Another subtely is that while we may read a degraded object, we will
+ * still only perform a client read from shards in the acting set. This
+ * ensures that we won't ever have to restart a client initiated read in
+ * check_recovery_sources.
+ */
+ void objects_read_and_reconstruct(
+ const map<hobject_t, std::list<boost::tuple<uint64_t, uint64_t, uint32_t> >
+ > &reads,
+ bool fast_read,
+ GenContextURef<map<hobject_t,pair<int, extent_map> > &&> &&func);
+
+ friend struct CallClientContexts;
+ struct ClientAsyncReadStatus {
+ unsigned objects_to_read;
+ GenContextURef<map<hobject_t,pair<int, extent_map> > &&> func;
+ map<hobject_t,pair<int, extent_map> > results;
+ explicit ClientAsyncReadStatus(
+ unsigned objects_to_read,
+ GenContextURef<map<hobject_t,pair<int, extent_map> > &&> &&func)
+ : objects_to_read(objects_to_read), func(std::move(func)) {}
+ void complete_object(
+ const hobject_t &hoid,
+ int err,
+ extent_map &&buffers) {
+ assert(objects_to_read);
+ --objects_to_read;
+ assert(!results.count(hoid));
+ results.emplace(hoid, make_pair(err, std::move(buffers)));
+ }
+ bool is_complete() const {
+ return objects_to_read == 0;
+ }
+ void run() {
+ func.release()->complete(std::move(results));
+ }
+ };
+ list<ClientAsyncReadStatus> in_progress_client_reads;
+ 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;
+
+ template <typename Func>
+ void objects_read_async_no_cache(
+ const map<hobject_t,extent_set> &to_read,
+ Func &&on_complete) {
+ map<hobject_t,std::list<boost::tuple<uint64_t, uint64_t, uint32_t> > > _to_read;
+ for (auto &&hpair: to_read) {
+ auto &l = _to_read[hpair.first];
+ for (auto extent: hpair.second) {
+ l.emplace_back(extent.first, extent.second, 0);
+ }
+ }
+ objects_read_and_reconstruct(
+ _to_read,
+ false,
+ make_gen_lambda_context<
+ map<hobject_t,pair<int, extent_map> > &&, Func>(
+ std::forward<Func>(on_complete)));
+ }
+ void kick_reads() {
+ while (in_progress_client_reads.size() &&
+ in_progress_client_reads.front().is_complete()) {
+ in_progress_client_reads.front().run();
+ in_progress_client_reads.pop_front();
+ }
+ }
+
+private:
+ friend struct ECRecoveryHandle;
+ uint64_t get_recovery_chunk_size() const {
+ return ROUND_UP_TO(cct->_conf->osd_recovery_max_chunk,
+ sinfo.get_stripe_width());
+ }
+
+ void get_want_to_read_shards(set<int> *want_to_read) const {
+ const vector<int> &chunk_mapping = ec_impl->get_chunk_mapping();
+ for (int i = 0; i < (int)ec_impl->get_data_chunk_count(); ++i) {
+ int chunk = (int)chunk_mapping.size() > i ? chunk_mapping[i] : i;
+ want_to_read->insert(chunk);
+ }
+ }
+
+ /**
+ * Recovery
+ *
+ * Recovery uses the same underlying read mechanism as client reads
+ * with the slight difference that recovery reads may come from non
+ * acting shards. Thus, check_recovery_sources may wind up calling
+ * cancel_pull for a read originating with RecoveryOp.
+ *
+ * The recovery process is expressed as a state machine:
+ * - IDLE: Nothing is currently in progress, reads will be started and
+ * we will transition to READING
+ * - READING: We are awaiting a pending read op. Once complete, we will
+ * decode the buffers and proceed to WRITING
+ * - WRITING: We are awaiting a completed push. Once complete, we will
+ * either transition to COMPLETE or to IDLE to continue.
+ * - COMPLETE: complete
+ *
+ * We use the existing Push and PushReply messages and structures to
+ * handle actually shuffling the data over to the replicas. recovery_info
+ * and recovery_progress are expressed in terms of the logical offset
+ * space except for data_included which is in terms of the chunked object
+ * space (to match the passed buffer).
+ *
+ * xattrs are requested on the first read and used to initialize the
+ * object_context if missing on completion of the first read.
+ *
+ * In order to batch up reads and writes, we batch Push, PushReply,
+ * Transaction, and reads in a RecoveryMessages object which is passed
+ * among the recovery methods.
+ */
+ struct RecoveryOp {
+ hobject_t hoid;
+ eversion_t v;
+ set<pg_shard_t> missing_on;
+ set<shard_id_t> missing_on_shards;
+
+ ObjectRecoveryInfo recovery_info;
+ ObjectRecoveryProgress recovery_progress;
+
+ enum state_t { IDLE, READING, WRITING, COMPLETE } state;
+
+ static const char* tostr(state_t state) {
+ switch (state) {
+ case ECBackend::RecoveryOp::IDLE:
+ return "IDLE";
+ break;
+ case ECBackend::RecoveryOp::READING:
+ return "READING";
+ break;
+ case ECBackend::RecoveryOp::WRITING:
+ return "WRITING";
+ break;
+ case ECBackend::RecoveryOp::COMPLETE:
+ return "COMPLETE";
+ break;
+ default:
+ ceph_abort();
+ return "";
+ }
+ }
+
+ // must be filled if state == WRITING
+ map<int, bufferlist> returned_data;
+ map<string, bufferlist> xattrs;
+ ECUtil::HashInfoRef hinfo;
+ ObjectContextRef obc;
+ set<pg_shard_t> waiting_on_pushes;
+
+ // valid in state READING
+ pair<uint64_t, uint64_t> extent_requested;
+
+ void dump(Formatter *f) const;
+
+ RecoveryOp() : state(IDLE) {}
+ };
+ friend ostream &operator<<(ostream &lhs, const RecoveryOp &rhs);
+ map<hobject_t, RecoveryOp> recovery_ops;
+
+ void continue_recovery_op(
+ RecoveryOp &op,
+ RecoveryMessages *m);
+ void dispatch_recovery_messages(RecoveryMessages &m, int priority);
+ friend struct OnRecoveryReadComplete;
+ void handle_recovery_read_complete(
+ const hobject_t &hoid,
+ boost::tuple<uint64_t, uint64_t, map<pg_shard_t, bufferlist> > &to_read,
+ boost::optional<map<string, bufferlist> > attrs,
+ RecoveryMessages *m);
+ void handle_recovery_push(
+ const PushOp &op,
+ RecoveryMessages *m);
+ void handle_recovery_push_reply(
+ const PushReplyOp &op,
+ pg_shard_t from,
+ RecoveryMessages *m);
+
+public:
+ /**
+ * Low level async read mechanism
+ *
+ * To avoid duplicating the logic for requesting and waiting for
+ * multiple object shards, there is a common async read mechanism
+ * taking a map of hobject_t->read_request_t which defines callbacks
+ * taking read_result_ts as arguments.
+ *
+ * tid_to_read_map gives open read ops. check_recovery_sources uses
+ * shard_to_read_map and ReadOp::source_to_obj to restart reads
+ * involving down osds.
+ *
+ * The user is responsible for specifying replicas on which to read
+ * and for reassembling the buffer on the other side since client
+ * reads require the original object buffer while recovery only needs
+ * the missing pieces.
+ *
+ * Rather than handling reads on the primary directly, we simply send
+ * ourselves a message. This avoids a dedicated primary path for that
+ * part.
+ */
+ struct read_result_t {
+ int r;
+ map<pg_shard_t, int> errors;
+ boost::optional<map<string, bufferlist> > attrs;
+ list<
+ boost::tuple<
+ uint64_t, uint64_t, map<pg_shard_t, bufferlist> > > returned;
+ read_result_t() : r(0) {}
+ };
+ struct read_request_t {
+ const list<boost::tuple<uint64_t, uint64_t, uint32_t> > to_read;
+ const set<pg_shard_t> need;
+ const bool want_attrs;
+ GenContext<pair<RecoveryMessages *, read_result_t& > &> *cb;
+ read_request_t(
+ const list<boost::tuple<uint64_t, uint64_t, uint32_t> > &to_read,
+ const set<pg_shard_t> &need,
+ bool want_attrs,
+ GenContext<pair<RecoveryMessages *, read_result_t& > &> *cb)
+ : to_read(to_read), need(need), want_attrs(want_attrs),
+ cb(cb) {}
+ };
+ friend ostream &operator<<(ostream &lhs, const read_request_t &rhs);
+
+ struct ReadOp {
+ int priority;
+ ceph_tid_t tid;
+ OpRequestRef op; // may be null if not on behalf of a client
+ // True if redundant reads are issued, false otherwise,
+ // this is useful to tradeoff some resources (redundant ops) for
+ // low latency read, especially on relatively idle cluster
+ bool do_redundant_reads;
+ // True if reading for recovery which could possibly reading only a subset
+ // of the available shards.
+ bool for_recovery;
+
+ ZTracer::Trace trace;
+
+ map<hobject_t, read_request_t> to_read;
+ map<hobject_t, read_result_t> complete;
+
+ map<hobject_t, set<pg_shard_t>> obj_to_source;
+ map<pg_shard_t, set<hobject_t> > source_to_obj;
+
+ void dump(Formatter *f) const;
+
+ set<pg_shard_t> in_progress;
+
+ ReadOp(
+ int priority,
+ ceph_tid_t tid,
+ bool do_redundant_reads,
+ bool for_recovery,
+ OpRequestRef op,
+ map<hobject_t, read_request_t> &&_to_read)
+ : priority(priority), tid(tid), op(op), do_redundant_reads(do_redundant_reads),
+ for_recovery(for_recovery), to_read(std::move(_to_read)) {
+ for (auto &&hpair: to_read) {
+ auto &returned = complete[hpair.first].returned;
+ for (auto &&extent: hpair.second.to_read) {
+ returned.push_back(
+ boost::make_tuple(
+ extent.get<0>(),
+ extent.get<1>(),
+ map<pg_shard_t, bufferlist>()));
+ }
+ }
+ }
+ ReadOp() = delete;
+ ReadOp(const ReadOp &) = default;
+ ReadOp(ReadOp &&) = default;
+ };
+ friend struct FinishReadOp;
+ void filter_read_op(
+ const OSDMapRef& osdmap,
+ ReadOp &op);
+ void complete_read_op(ReadOp &rop, RecoveryMessages *m);
+ friend ostream &operator<<(ostream &lhs, const ReadOp &rhs);
+ map<ceph_tid_t, ReadOp> tid_to_read_map;
+ map<pg_shard_t, set<ceph_tid_t> > shard_to_read_map;
+ void start_read_op(
+ int priority,
+ map<hobject_t, read_request_t> &to_read,
+ OpRequestRef op,
+ bool do_redundant_reads, bool for_recovery);
+
+ void do_read_op(ReadOp &rop);
+ int send_all_remaining_reads(
+ const hobject_t &hoid,
+ ReadOp &rop);
+
+
+ /**
+ * Client writes
+ *
+ * ECTransaction is responsible for generating a transaction for
+ * each shard to which we need to send the write. As required
+ * by the PGBackend interface, the ECBackend write mechanism
+ * passes trim information with the write and last_complete back
+ * with the reply.
+ *
+ * As with client reads, there is a possibility of out-of-order
+ * completions. Thus, callbacks and completion are called in order
+ * on the writing list.
+ */
+ struct Op : boost::intrusive::list_base_hook<> {
+ /// From submit_transaction caller, decribes operation
+ hobject_t hoid;
+ object_stat_sum_t delta_stats;
+ eversion_t version;
+ eversion_t trim_to;
+ boost::optional<pg_hit_set_history_t> updated_hit_set_history;
+ vector<pg_log_entry_t> log_entries;
+ ceph_tid_t tid;
+ osd_reqid_t reqid;
+ ZTracer::Trace trace;
+
+ eversion_t roll_forward_to; /// Soon to be generated internally
+
+ /// Ancillary also provided from submit_transaction caller
+ map<hobject_t, ObjectContextRef> obc_map;
+
+ /// see call_write_ordered
+ std::list<std::function<void(void)> > on_write;
+
+ /// Generated internally
+ set<hobject_t> temp_added;
+ set<hobject_t> temp_cleared;
+
+ ECTransaction::WritePlan plan;
+ bool requires_rmw() const { return !plan.to_read.empty(); }
+ bool invalidates_cache() const { return plan.invalidates_cache; }
+
+ // must be true if requires_rmw(), must be false if invalidates_cache()
+ bool using_cache = false;
+
+ /// In progress read state;
+ map<hobject_t,extent_set> pending_read; // subset already being read
+ map<hobject_t,extent_set> remote_read; // subset we must read
+ map<hobject_t,extent_map> remote_read_result;
+ bool read_in_progress() const {
+ return !remote_read.empty() && remote_read_result.empty();
+ }
+
+ /// In progress write state
+ set<pg_shard_t> pending_commit;
+ set<pg_shard_t> pending_apply;
+ bool write_in_progress() const {
+ return !pending_commit.empty() || !pending_apply.empty();
+ }
+
+ /// optional, may be null, for tracking purposes
+ OpRequestRef client_op;
+
+ /// pin for cache
+ ExtentCache::write_pin pin;
+
+ /// Callbacks
+ Context *on_local_applied_sync = nullptr;
+ Context *on_all_applied = nullptr;
+ Context *on_all_commit = nullptr;
+ ~Op() {
+ delete on_local_applied_sync;
+ delete on_all_applied;
+ delete on_all_commit;
+ }
+ };
+ using op_list = boost::intrusive::list<Op>;
+ friend ostream &operator<<(ostream &lhs, const Op &rhs);
+
+ ExtentCache cache;
+ map<ceph_tid_t, Op> tid_to_op_map; /// Owns Op structure
+
+ /**
+ * We model the possible rmw states as a set of waitlists.
+ * All writes at this time complete in order, so a write blocked
+ * at waiting_state blocks all writes behind it as well (same for
+ * other states).
+ *
+ * Future work: We can break this up into a per-object pipeline
+ * (almost). First, provide an ordering token to submit_transaction
+ * and require that all operations within a single transaction take
+ * place on a subset of hobject_t space partitioned by that token
+ * (the hashid seem about right to me -- even works for temp objects
+ * if you recall that a temp object created for object head foo will
+ * only ever be referenced by other transactions on foo and aren't
+ * reused). Next, factor this part into a class and maintain one per
+ * ordering token. Next, fixup PrimaryLogPG's repop queue to be
+ * partitioned by ordering token. Finally, refactor the op pipeline
+ * so that the log entries passed into submit_tranaction aren't
+ * versioned. We can't assign versions to them until we actually
+ * submit the operation. That's probably going to be the hard part.
+ */
+ class pipeline_state_t {
+ enum {
+ CACHE_VALID = 0,
+ CACHE_INVALID = 1
+ } pipeline_state = CACHE_VALID;
+ public:
+ bool caching_enabled() const {
+ return pipeline_state == CACHE_VALID;
+ }
+ bool cache_invalid() const {
+ return !caching_enabled();
+ }
+ void invalidate() {
+ pipeline_state = CACHE_INVALID;
+ }
+ void clear() {
+ pipeline_state = CACHE_VALID;
+ }
+ friend ostream &operator<<(ostream &lhs, const pipeline_state_t &rhs);
+ } pipeline_state;
+
+
+ op_list waiting_state; /// writes waiting on pipe_state
+ op_list waiting_reads; /// writes waiting on partial stripe reads
+ op_list waiting_commit; /// writes waiting on initial commit
+ eversion_t completed_to;
+ eversion_t committed_to;
+ void start_rmw(Op *op, PGTransactionUPtr &&t);
+ bool try_state_to_reads();
+ bool try_reads_to_commit();
+ bool try_finish_rmw();
+ void check_ops();
+
+ ErasureCodeInterfaceRef ec_impl;
+
+
+ /**
+ * ECRecPred
+ *
+ * Determines the whether _have is suffient to recover an object
+ */
+ class ECRecPred : public IsPGRecoverablePredicate {
+ set<int> want;
+ ErasureCodeInterfaceRef ec_impl;
+ public:
+ explicit ECRecPred(ErasureCodeInterfaceRef ec_impl) : ec_impl(ec_impl) {
+ for (unsigned i = 0; i < ec_impl->get_chunk_count(); ++i) {
+ want.insert(i);
+ }
+ }
+ bool operator()(const set<pg_shard_t> &_have) const override {
+ set<int> have;
+ for (set<pg_shard_t>::const_iterator i = _have.begin();
+ i != _have.end();
+ ++i) {
+ have.insert(i->shard);
+ }
+ set<int> min;
+ return ec_impl->minimum_to_decode(want, have, &min) == 0;
+ }
+ };
+ IsPGRecoverablePredicate *get_is_recoverable_predicate() override {
+ return new ECRecPred(ec_impl);
+ }
+
+ /**
+ * ECReadPred
+ *
+ * Determines the whether _have is suffient to read an object
+ */
+ class ECReadPred : public IsPGReadablePredicate {
+ pg_shard_t whoami;
+ ECRecPred rec_pred;
+ public:
+ ECReadPred(
+ pg_shard_t whoami,
+ ErasureCodeInterfaceRef ec_impl) : whoami(whoami), rec_pred(ec_impl) {}
+ bool operator()(const set<pg_shard_t> &_have) const override {
+ return _have.count(whoami) && rec_pred(_have);
+ }
+ };
+ IsPGReadablePredicate *get_is_readable_predicate() override {
+ return new ECReadPred(get_parent()->whoami_shard(), ec_impl);
+ }
+
+
+ const ECUtil::stripe_info_t sinfo;
+ /// If modified, ensure that the ref is held until the update is applied
+ SharedPtrRegistry<hobject_t, ECUtil::HashInfo> unstable_hashinfo_registry;
+ ECUtil::HashInfoRef get_hash_info(const hobject_t &hoid, bool checks = true,
+ const map<string,bufferptr> *attr = NULL);
+
+public:
+ ECBackend(
+ PGBackend::Listener *pg,
+ coll_t coll,
+ ObjectStore::CollectionHandle &ch,
+ ObjectStore *store,
+ CephContext *cct,
+ ErasureCodeInterfaceRef ec_impl,
+ uint64_t stripe_width);
+
+ /// Returns to_read replicas sufficient to reconstruct want
+ int get_min_avail_to_read_shards(
+ const hobject_t &hoid, ///< [in] object
+ const set<int> &want, ///< [in] desired shards
+ bool for_recovery, ///< [in] true if we may use non-acting replicas
+ bool do_redundant_reads, ///< [in] true if we want to issue redundant reads to reduce latency
+ set<pg_shard_t> *to_read ///< [out] shards to read
+ ); ///< @return error code, 0 on success
+
+ int get_remaining_shards(
+ const hobject_t &hoid,
+ const set<int> &avail,
+ set<pg_shard_t> *to_read);
+
+ int objects_get_attrs(
+ const hobject_t &hoid,
+ map<string, bufferlist> *out) override;
+
+ void rollback_append(
+ const hobject_t &hoid,
+ uint64_t old_size,
+ ObjectStore::Transaction *t) override;
+
+ bool scrub_supported() override { return true; }
+ bool auto_repair_supported() const override { return true; }
+
+ 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 sinfo.logical_to_next_chunk_offset(logical_size);
+ }
+ void _failed_push(const hobject_t &hoid,
+ pair<RecoveryMessages *, ECBackend::read_result_t &> &in);
+};
+ostream &operator<<(ostream &lhs, const ECBackend::pipeline_state_t &rhs);
+
+#endif
Code Review / stor4nfv.git / blobdiff