+++ /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.
- *
- */
-#include "common/errno.h"
-#include "ReplicatedBackend.h"
-#include "messages/MOSDOp.h"
-#include "messages/MOSDSubOp.h"
-#include "messages/MOSDRepOp.h"
-#include "messages/MOSDSubOpReply.h"
-#include "messages/MOSDRepOpReply.h"
-#include "messages/MOSDPGPush.h"
-#include "messages/MOSDPGPull.h"
-#include "messages/MOSDPGPushReply.h"
-#include "common/EventTrace.h"
-
-#define dout_context cct
-#define dout_subsys ceph_subsys_osd
-#define DOUT_PREFIX_ARGS this
-#undef dout_prefix
-#define dout_prefix _prefix(_dout, this)
-static ostream& _prefix(std::ostream *_dout, ReplicatedBackend *pgb) {
- return *_dout << pgb->get_parent()->gen_dbg_prefix();
-}
-
-namespace {
-class PG_SendMessageOnConn: public Context {
- PGBackend::Listener *pg;
- Message *reply;
- ConnectionRef conn;
- public:
- PG_SendMessageOnConn(
- PGBackend::Listener *pg,
- Message *reply,
- ConnectionRef conn) : pg(pg), reply(reply), conn(conn) {}
- void finish(int) override {
- pg->send_message_osd_cluster(reply, conn.get());
- }
-};
-
-class PG_RecoveryQueueAsync : public Context {
- PGBackend::Listener *pg;
- unique_ptr<GenContext<ThreadPool::TPHandle&>> c;
- public:
- PG_RecoveryQueueAsync(
- PGBackend::Listener *pg,
- GenContext<ThreadPool::TPHandle&> *c) : pg(pg), c(c) {}
- void finish(int) override {
- pg->schedule_recovery_work(c.release());
- }
-};
-}
-
-struct ReplicatedBackend::C_OSD_RepModifyApply : public Context {
- ReplicatedBackend *pg;
- RepModifyRef rm;
- C_OSD_RepModifyApply(ReplicatedBackend *pg, RepModifyRef r)
- : pg(pg), rm(r) {}
- void finish(int r) override {
- pg->repop_applied(rm);
- }
-};
-
-struct ReplicatedBackend::C_OSD_RepModifyCommit : public Context {
- ReplicatedBackend *pg;
- RepModifyRef rm;
- C_OSD_RepModifyCommit(ReplicatedBackend *pg, RepModifyRef r)
- : pg(pg), rm(r) {}
- void finish(int r) override {
- pg->repop_commit(rm);
- }
-};
-
-static void log_subop_stats(
- PerfCounters *logger,
- OpRequestRef op, int subop)
-{
- utime_t now = ceph_clock_now();
- utime_t latency = now;
- latency -= op->get_req()->get_recv_stamp();
-
-
- logger->inc(l_osd_sop);
- logger->tinc(l_osd_sop_lat, latency);
- logger->inc(subop);
-
- if (subop != l_osd_sop_pull) {
- uint64_t inb = op->get_req()->get_data().length();
- logger->inc(l_osd_sop_inb, inb);
- if (subop == l_osd_sop_w) {
- logger->inc(l_osd_sop_w_inb, inb);
- logger->tinc(l_osd_sop_w_lat, latency);
- } else if (subop == l_osd_sop_push) {
- logger->inc(l_osd_sop_push_inb, inb);
- logger->tinc(l_osd_sop_push_lat, latency);
- } else
- assert("no support subop" == 0);
- } else {
- logger->tinc(l_osd_sop_pull_lat, latency);
- }
-}
-
-ReplicatedBackend::ReplicatedBackend(
- PGBackend::Listener *pg,
- coll_t coll,
- ObjectStore::CollectionHandle &c,
- ObjectStore *store,
- CephContext *cct) :
- PGBackend(cct, pg, store, coll, c) {}
-
-void ReplicatedBackend::run_recovery_op(
- PGBackend::RecoveryHandle *_h,
- int priority)
-{
- RPGHandle *h = static_cast<RPGHandle *>(_h);
- send_pushes(priority, h->pushes);
- send_pulls(priority, h->pulls);
- send_recovery_deletes(priority, h->deletes);
- delete h;
-}
-
-int ReplicatedBackend::recover_object(
- const hobject_t &hoid,
- eversion_t v,
- ObjectContextRef head,
- ObjectContextRef obc,
- RecoveryHandle *_h
- )
-{
- dout(10) << __func__ << ": " << hoid << dendl;
- RPGHandle *h = static_cast<RPGHandle *>(_h);
- if (get_parent()->get_local_missing().is_missing(hoid)) {
- assert(!obc);
- // pull
- prepare_pull(
- v,
- hoid,
- head,
- h);
- } else {
- assert(obc);
- int started = start_pushes(
- hoid,
- obc,
- h);
- if (started < 0) {
- pushing[hoid].clear();
- return started;
- }
- }
- return 0;
-}
-
-void ReplicatedBackend::check_recovery_sources(const OSDMapRef& osdmap)
-{
- for(map<pg_shard_t, set<hobject_t> >::iterator i = pull_from_peer.begin();
- i != pull_from_peer.end();
- ) {
- if (osdmap->is_down(i->first.osd)) {
- dout(10) << "check_recovery_sources resetting pulls from osd." << i->first
- << ", osdmap has it marked down" << dendl;
- for (set<hobject_t>::iterator j = i->second.begin();
- j != i->second.end();
- ++j) {
- get_parent()->cancel_pull(*j);
- clear_pull(pulling.find(*j), false);
- }
- pull_from_peer.erase(i++);
- } else {
- ++i;
- }
- }
-}
-
-bool ReplicatedBackend::can_handle_while_inactive(OpRequestRef op)
-{
- dout(10) << __func__ << ": " << op << dendl;
- switch (op->get_req()->get_type()) {
- case MSG_OSD_PG_PULL:
- return true;
- default:
- return false;
- }
-}
-
-bool ReplicatedBackend::_handle_message(
- OpRequestRef op
- )
-{
- dout(10) << __func__ << ": " << op << dendl;
- switch (op->get_req()->get_type()) {
- case MSG_OSD_PG_PUSH:
- do_push(op);
- return true;
-
- case MSG_OSD_PG_PULL:
- do_pull(op);
- return true;
-
- case MSG_OSD_PG_PUSH_REPLY:
- do_push_reply(op);
- return true;
-
- case MSG_OSD_SUBOP: {
- const MOSDSubOp *m = static_cast<const MOSDSubOp*>(op->get_req());
- if (m->ops.size() == 0) {
- assert(0);
- }
- break;
- }
-
- case MSG_OSD_REPOP: {
- do_repop(op);
- return true;
- }
-
- case MSG_OSD_REPOPREPLY: {
- do_repop_reply(op);
- return true;
- }
-
- default:
- break;
- }
- return false;
-}
-
-void ReplicatedBackend::clear_recovery_state()
-{
- // clear pushing/pulling maps
- for (auto &&i: pushing) {
- for (auto &&j: i.second) {
- get_parent()->release_locks(j.second.lock_manager);
- }
- }
- pushing.clear();
-
- for (auto &&i: pulling) {
- get_parent()->release_locks(i.second.lock_manager);
- }
- pulling.clear();
- pull_from_peer.clear();
-}
-
-void ReplicatedBackend::on_change()
-{
- dout(10) << __func__ << dendl;
- for (map<ceph_tid_t, InProgressOp>::iterator i = in_progress_ops.begin();
- i != in_progress_ops.end();
- in_progress_ops.erase(i++)) {
- if (i->second.on_commit)
- delete i->second.on_commit;
- if (i->second.on_applied)
- delete i->second.on_applied;
- }
- clear_recovery_state();
-}
-
-void ReplicatedBackend::on_flushed()
-{
-}
-
-int ReplicatedBackend::objects_read_sync(
- const hobject_t &hoid,
- uint64_t off,
- uint64_t len,
- uint32_t op_flags,
- bufferlist *bl)
-{
- return store->read(ch, ghobject_t(hoid), off, len, *bl, op_flags);
-}
-
-struct AsyncReadCallback : public GenContext<ThreadPool::TPHandle&> {
- int r;
- Context *c;
- AsyncReadCallback(int r, Context *c) : r(r), c(c) {}
- void finish(ThreadPool::TPHandle&) override {
- c->complete(r);
- c = NULL;
- }
- ~AsyncReadCallback() override {
- delete c;
- }
-};
-void ReplicatedBackend::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)
-{
- // There is no fast read implementation for replication backend yet
- assert(!fast_read);
-
- int r = 0;
- for (list<pair<boost::tuple<uint64_t, uint64_t, uint32_t>,
- pair<bufferlist*, Context*> > >::const_iterator i =
- to_read.begin();
- i != to_read.end() && r >= 0;
- ++i) {
- int _r = store->read(ch, ghobject_t(hoid), i->first.get<0>(),
- i->first.get<1>(), *(i->second.first),
- i->first.get<2>());
- if (i->second.second) {
- get_parent()->schedule_recovery_work(
- get_parent()->bless_gencontext(
- new AsyncReadCallback(_r, i->second.second)));
- }
- if (_r < 0)
- r = _r;
- }
- get_parent()->schedule_recovery_work(
- get_parent()->bless_gencontext(
- new AsyncReadCallback(r, on_complete)));
-}
-
-class C_OSD_OnOpCommit : public Context {
- ReplicatedBackend *pg;
- ReplicatedBackend::InProgressOp *op;
-public:
- C_OSD_OnOpCommit(ReplicatedBackend *pg, ReplicatedBackend::InProgressOp *op)
- : pg(pg), op(op) {}
- void finish(int) override {
- pg->op_commit(op);
- }
-};
-
-class C_OSD_OnOpApplied : public Context {
- ReplicatedBackend *pg;
- ReplicatedBackend::InProgressOp *op;
-public:
- C_OSD_OnOpApplied(ReplicatedBackend *pg, ReplicatedBackend::InProgressOp *op)
- : pg(pg), op(op) {}
- void finish(int) override {
- pg->op_applied(op);
- }
-};
-
-void generate_transaction(
- PGTransactionUPtr &pgt,
- const coll_t &coll,
- bool legacy_log_entries,
- vector<pg_log_entry_t> &log_entries,
- ObjectStore::Transaction *t,
- set<hobject_t> *added,
- set<hobject_t> *removed)
-{
- assert(t);
- assert(added);
- assert(removed);
-
- for (auto &&le: log_entries) {
- le.mark_unrollbackable();
- auto oiter = pgt->op_map.find(le.soid);
- if (oiter != pgt->op_map.end() && oiter->second.updated_snaps) {
- bufferlist bl(oiter->second.updated_snaps->second.size() * 8 + 8);
- ::encode(oiter->second.updated_snaps->second, bl);
- le.snaps.swap(bl);
- le.snaps.reassign_to_mempool(mempool::mempool_osd_pglog);
- }
- }
-
- pgt->safe_create_traverse(
- [&](pair<const hobject_t, PGTransaction::ObjectOperation> &obj_op) {
- const hobject_t &oid = obj_op.first;
- const ghobject_t goid =
- ghobject_t(oid, ghobject_t::NO_GEN, shard_id_t::NO_SHARD);
- const PGTransaction::ObjectOperation &op = obj_op.second;
-
- if (oid.is_temp()) {
- if (op.is_fresh_object()) {
- added->insert(oid);
- } else if (op.is_delete()) {
- removed->insert(oid);
- }
- }
-
- if (op.delete_first) {
- t->remove(coll, goid);
- }
-
- match(
- op.init_type,
- [&](const PGTransaction::ObjectOperation::Init::None &) {
- },
- [&](const PGTransaction::ObjectOperation::Init::Create &op) {
- t->touch(coll, goid);
- },
- [&](const PGTransaction::ObjectOperation::Init::Clone &op) {
- t->clone(
- coll,
- ghobject_t(
- op.source, ghobject_t::NO_GEN, shard_id_t::NO_SHARD),
- goid);
- },
- [&](const PGTransaction::ObjectOperation::Init::Rename &op) {
- assert(op.source.is_temp());
- t->collection_move_rename(
- coll,
- ghobject_t(
- op.source, ghobject_t::NO_GEN, shard_id_t::NO_SHARD),
- coll,
- goid);
- });
-
- if (op.truncate) {
- t->truncate(coll, goid, op.truncate->first);
- if (op.truncate->first != op.truncate->second)
- t->truncate(coll, goid, op.truncate->second);
- }
-
- if (!op.attr_updates.empty()) {
- map<string, bufferlist> attrs;
- for (auto &&p: op.attr_updates) {
- if (p.second)
- attrs[p.first] = *(p.second);
- else
- t->rmattr(coll, goid, p.first);
- }
- t->setattrs(coll, goid, attrs);
- }
-
- if (op.clear_omap)
- t->omap_clear(coll, goid);
- if (op.omap_header)
- t->omap_setheader(coll, goid, *(op.omap_header));
-
- for (auto &&up: op.omap_updates) {
- using UpdateType = PGTransaction::ObjectOperation::OmapUpdateType;
- switch (up.first) {
- case UpdateType::Remove:
- t->omap_rmkeys(coll, goid, up.second);
- break;
- case UpdateType::Insert:
- t->omap_setkeys(coll, goid, up.second);
- break;
- }
- }
-
- // updated_snaps doesn't matter since we marked unrollbackable
-
- if (op.alloc_hint) {
- auto &hint = *(op.alloc_hint);
- t->set_alloc_hint(
- coll,
- goid,
- hint.expected_object_size,
- hint.expected_write_size,
- hint.flags);
- }
-
- for (auto &&extent: op.buffer_updates) {
- using BufferUpdate = PGTransaction::ObjectOperation::BufferUpdate;
- match(
- extent.get_val(),
- [&](const BufferUpdate::Write &op) {
- t->write(
- coll,
- goid,
- extent.get_off(),
- extent.get_len(),
- op.buffer);
- },
- [&](const BufferUpdate::Zero &op) {
- t->zero(
- coll,
- goid,
- extent.get_off(),
- extent.get_len());
- },
- [&](const BufferUpdate::CloneRange &op) {
- assert(op.len == extent.get_len());
- t->clone_range(
- coll,
- ghobject_t(op.from, ghobject_t::NO_GEN, shard_id_t::NO_SHARD),
- goid,
- op.offset,
- extent.get_len(),
- extent.get_off());
- });
- }
- });
-}
-
-void ReplicatedBackend::submit_transaction(
- const hobject_t &soid,
- 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_acked,
- Context *on_all_commit,
- ceph_tid_t tid,
- osd_reqid_t reqid,
- OpRequestRef orig_op)
-{
- parent->apply_stats(
- soid,
- delta_stats);
-
- vector<pg_log_entry_t> log_entries(_log_entries);
- ObjectStore::Transaction op_t;
- PGTransactionUPtr t(std::move(_t));
- set<hobject_t> added, removed;
- generate_transaction(
- t,
- coll,
- (get_osdmap()->require_osd_release < CEPH_RELEASE_KRAKEN),
- log_entries,
- &op_t,
- &added,
- &removed);
- assert(added.size() <= 1);
- assert(removed.size() <= 1);
-
- assert(!in_progress_ops.count(tid));
- InProgressOp &op = in_progress_ops.insert(
- make_pair(
- tid,
- InProgressOp(
- tid, on_all_commit, on_all_acked,
- orig_op, at_version)
- )
- ).first->second;
-
- op.waiting_for_applied.insert(
- parent->get_actingbackfill_shards().begin(),
- parent->get_actingbackfill_shards().end());
- op.waiting_for_commit.insert(
- parent->get_actingbackfill_shards().begin(),
- parent->get_actingbackfill_shards().end());
-
- issue_op(
- soid,
- at_version,
- tid,
- reqid,
- trim_to,
- at_version,
- added.size() ? *(added.begin()) : hobject_t(),
- removed.size() ? *(removed.begin()) : hobject_t(),
- log_entries,
- hset_history,
- &op,
- op_t);
-
- add_temp_objs(added);
- clear_temp_objs(removed);
-
- parent->log_operation(
- log_entries,
- hset_history,
- trim_to,
- at_version,
- true,
- op_t);
-
- op_t.register_on_applied_sync(on_local_applied_sync);
- op_t.register_on_applied(
- parent->bless_context(
- new C_OSD_OnOpApplied(this, &op)));
- op_t.register_on_commit(
- parent->bless_context(
- new C_OSD_OnOpCommit(this, &op)));
-
- vector<ObjectStore::Transaction> tls;
- tls.push_back(std::move(op_t));
-
- parent->queue_transactions(tls, op.op);
-}
-
-void ReplicatedBackend::op_applied(
- InProgressOp *op)
-{
- FUNCTRACE();
- OID_EVENT_TRACE_WITH_MSG((op && op->op) ? op->op->get_req() : NULL, "OP_APPLIED_BEGIN", true);
- dout(10) << __func__ << ": " << op->tid << dendl;
- if (op->op) {
- op->op->mark_event("op_applied");
- op->op->pg_trace.event("op applied");
- }
-
- op->waiting_for_applied.erase(get_parent()->whoami_shard());
- parent->op_applied(op->v);
-
- if (op->waiting_for_applied.empty()) {
- op->on_applied->complete(0);
- op->on_applied = 0;
- }
- if (op->done()) {
- assert(!op->on_commit && !op->on_applied);
- in_progress_ops.erase(op->tid);
- }
-}
-
-void ReplicatedBackend::op_commit(
- InProgressOp *op)
-{
- FUNCTRACE();
- OID_EVENT_TRACE_WITH_MSG((op && op->op) ? op->op->get_req() : NULL, "OP_COMMIT_BEGIN", true);
- dout(10) << __func__ << ": " << op->tid << dendl;
- if (op->op) {
- op->op->mark_event("op_commit");
- op->op->pg_trace.event("op commit");
- }
-
- op->waiting_for_commit.erase(get_parent()->whoami_shard());
-
- if (op->waiting_for_commit.empty()) {
- op->on_commit->complete(0);
- op->on_commit = 0;
- }
- if (op->done()) {
- assert(!op->on_commit && !op->on_applied);
- in_progress_ops.erase(op->tid);
- }
-}
-
-void ReplicatedBackend::do_repop_reply(OpRequestRef op)
-{
- static_cast<MOSDRepOpReply*>(op->get_nonconst_req())->finish_decode();
- const MOSDRepOpReply *r = static_cast<const MOSDRepOpReply *>(op->get_req());
- assert(r->get_header().type == MSG_OSD_REPOPREPLY);
-
- op->mark_started();
-
- // must be replication.
- ceph_tid_t rep_tid = r->get_tid();
- pg_shard_t from = r->from;
-
- if (in_progress_ops.count(rep_tid)) {
- map<ceph_tid_t, InProgressOp>::iterator iter =
- in_progress_ops.find(rep_tid);
- InProgressOp &ip_op = iter->second;
- const MOSDOp *m = NULL;
- if (ip_op.op)
- m = static_cast<const MOSDOp *>(ip_op.op->get_req());
-
- if (m)
- dout(7) << __func__ << ": tid " << ip_op.tid << " op " //<< *m
- << " ack_type " << (int)r->ack_type
- << " from " << from
- << dendl;
- else
- dout(7) << __func__ << ": tid " << ip_op.tid << " (no op) "
- << " ack_type " << (int)r->ack_type
- << " from " << from
- << dendl;
-
- // oh, good.
-
- if (r->ack_type & CEPH_OSD_FLAG_ONDISK) {
- assert(ip_op.waiting_for_commit.count(from));
- ip_op.waiting_for_commit.erase(from);
- if (ip_op.op) {
- ostringstream ss;
- ss << "sub_op_commit_rec from " << from;
- ip_op.op->mark_event_string(ss.str());
- ip_op.op->pg_trace.event("sub_op_commit_rec");
- }
- } else {
- assert(ip_op.waiting_for_applied.count(from));
- if (ip_op.op) {
- ostringstream ss;
- ss << "sub_op_applied_rec from " << from;
- ip_op.op->mark_event_string(ss.str());
- ip_op.op->pg_trace.event("sub_op_applied_rec");
- }
- }
- ip_op.waiting_for_applied.erase(from);
-
- parent->update_peer_last_complete_ondisk(
- from,
- r->get_last_complete_ondisk());
-
- if (ip_op.waiting_for_applied.empty() &&
- ip_op.on_applied) {
- ip_op.on_applied->complete(0);
- ip_op.on_applied = 0;
- }
- if (ip_op.waiting_for_commit.empty() &&
- ip_op.on_commit) {
- ip_op.on_commit->complete(0);
- ip_op.on_commit= 0;
- }
- if (ip_op.done()) {
- assert(!ip_op.on_commit && !ip_op.on_applied);
- in_progress_ops.erase(iter);
- }
- }
-}
-
-void ReplicatedBackend::be_deep_scrub(
- const hobject_t &poid,
- uint32_t seed,
- ScrubMap::object &o,
- ThreadPool::TPHandle &handle)
-{
- dout(10) << __func__ << " " << poid << " seed "
- << std::hex << seed << std::dec << dendl;
- bufferhash h(seed), oh(seed);
- bufferlist bl, hdrbl;
- int r;
- __u64 pos = 0;
-
- uint32_t fadvise_flags = CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL | CEPH_OSD_OP_FLAG_FADVISE_DONTNEED;
-
- while (true) {
- handle.reset_tp_timeout();
- r = store->read(
- ch,
- ghobject_t(
- poid, ghobject_t::NO_GEN, get_parent()->whoami_shard().shard),
- pos,
- cct->_conf->osd_deep_scrub_stride, bl,
- fadvise_flags);
- if (r <= 0)
- break;
-
- h << bl;
- pos += bl.length();
- bl.clear();
- }
- if (r == -EIO) {
- dout(25) << __func__ << " " << poid << " got "
- << r << " on read, read_error" << dendl;
- o.read_error = true;
- return;
- }
- o.digest = h.digest();
- o.digest_present = true;
-
- bl.clear();
- r = store->omap_get_header(
- coll,
- ghobject_t(
- poid, ghobject_t::NO_GEN, get_parent()->whoami_shard().shard),
- &hdrbl, true);
- // NOTE: bobtail to giant, we would crc the head as (len, head).
- // that changes at the same time we start using a non-zero seed.
- if (r == 0 && hdrbl.length()) {
- dout(25) << "CRC header " << string(hdrbl.c_str(), hdrbl.length())
- << dendl;
- if (seed == 0) {
- // legacy
- bufferlist bl;
- ::encode(hdrbl, bl);
- oh << bl;
- } else {
- oh << hdrbl;
- }
- } else if (r == -EIO) {
- dout(25) << __func__ << " " << poid << " got "
- << r << " on omap header read, read_error" << dendl;
- o.read_error = true;
- return;
- }
-
- ObjectMap::ObjectMapIterator iter = store->get_omap_iterator(
- coll,
- ghobject_t(
- poid, ghobject_t::NO_GEN, get_parent()->whoami_shard().shard));
- assert(iter);
- for (iter->seek_to_first(); iter->status() == 0 && iter->valid();
- iter->next(false)) {
- handle.reset_tp_timeout();
-
- dout(25) << "CRC key " << iter->key() << " value:\n";
- iter->value().hexdump(*_dout);
- *_dout << dendl;
-
- ::encode(iter->key(), bl);
- ::encode(iter->value(), bl);
- oh << bl;
- bl.clear();
- }
-
- if (iter->status() < 0) {
- dout(25) << __func__ << " " << poid
- << " on omap scan, db status error" << dendl;
- o.read_error = true;
- return;
- }
-
- //Store final calculated CRC32 of omap header & key/values
- o.omap_digest = oh.digest();
- o.omap_digest_present = true;
- dout(20) << __func__ << " " << poid << " omap_digest "
- << std::hex << o.omap_digest << std::dec << dendl;
-}
-
-void ReplicatedBackend::_do_push(OpRequestRef op)
-{
- const MOSDPGPush *m = static_cast<const MOSDPGPush *>(op->get_req());
- assert(m->get_type() == MSG_OSD_PG_PUSH);
- pg_shard_t from = m->from;
-
- op->mark_started();
-
- vector<PushReplyOp> replies;
- ObjectStore::Transaction t;
- ostringstream ss;
- if (get_parent()->check_failsafe_full(ss)) {
- dout(10) << __func__ << " Out of space (failsafe) processing push request: " << ss.str() << dendl;
- ceph_abort();
- }
- for (vector<PushOp>::const_iterator i = m->pushes.begin();
- i != m->pushes.end();
- ++i) {
- replies.push_back(PushReplyOp());
- handle_push(from, *i, &(replies.back()), &t);
- }
-
- MOSDPGPushReply *reply = new MOSDPGPushReply;
- reply->from = get_parent()->whoami_shard();
- reply->set_priority(m->get_priority());
- reply->pgid = get_info().pgid;
- reply->map_epoch = m->map_epoch;
- reply->min_epoch = m->min_epoch;
- reply->replies.swap(replies);
- reply->compute_cost(cct);
-
- t.register_on_complete(
- new PG_SendMessageOnConn(
- get_parent(), reply, m->get_connection()));
-
- get_parent()->queue_transaction(std::move(t));
-}
-
-struct C_ReplicatedBackend_OnPullComplete : GenContext<ThreadPool::TPHandle&> {
- ReplicatedBackend *bc;
- list<ReplicatedBackend::pull_complete_info> to_continue;
- int priority;
- C_ReplicatedBackend_OnPullComplete(ReplicatedBackend *bc, int priority)
- : bc(bc), priority(priority) {}
-
- void finish(ThreadPool::TPHandle &handle) override {
- ReplicatedBackend::RPGHandle *h = bc->_open_recovery_op();
- for (auto &&i: to_continue) {
- auto j = bc->pulling.find(i.hoid);
- assert(j != bc->pulling.end());
- ObjectContextRef obc = j->second.obc;
- bc->clear_pull(j, false /* already did it */);
- int started = bc->start_pushes(i.hoid, obc, h);
- if (started < 0) {
- bc->pushing[i.hoid].clear();
- bc->get_parent()->primary_failed(i.hoid);
- bc->get_parent()->primary_error(i.hoid, obc->obs.oi.version);
- } else if (!started) {
- bc->get_parent()->on_global_recover(
- i.hoid, i.stat, false);
- }
- handle.reset_tp_timeout();
- }
- bc->run_recovery_op(h, priority);
- }
-};
-
-void ReplicatedBackend::_do_pull_response(OpRequestRef op)
-{
- const MOSDPGPush *m = static_cast<const MOSDPGPush *>(op->get_req());
- assert(m->get_type() == MSG_OSD_PG_PUSH);
- pg_shard_t from = m->from;
-
- op->mark_started();
-
- vector<PullOp> replies(1);
-
- ostringstream ss;
- if (get_parent()->check_failsafe_full(ss)) {
- dout(10) << __func__ << " Out of space (failsafe) processing pull response (push): " << ss.str() << dendl;
- ceph_abort();
- }
-
- ObjectStore::Transaction t;
- list<pull_complete_info> to_continue;
- for (vector<PushOp>::const_iterator i = m->pushes.begin();
- i != m->pushes.end();
- ++i) {
- bool more = handle_pull_response(from, *i, &(replies.back()), &to_continue, &t);
- if (more)
- replies.push_back(PullOp());
- }
- if (!to_continue.empty()) {
- C_ReplicatedBackend_OnPullComplete *c =
- new C_ReplicatedBackend_OnPullComplete(
- this,
- m->get_priority());
- c->to_continue.swap(to_continue);
- t.register_on_complete(
- new PG_RecoveryQueueAsync(
- get_parent(),
- get_parent()->bless_gencontext(c)));
- }
- replies.erase(replies.end() - 1);
-
- if (replies.size()) {
- MOSDPGPull *reply = new MOSDPGPull;
- reply->from = parent->whoami_shard();
- reply->set_priority(m->get_priority());
- reply->pgid = get_info().pgid;
- reply->map_epoch = m->map_epoch;
- reply->min_epoch = m->min_epoch;
- reply->set_pulls(&replies);
- reply->compute_cost(cct);
-
- t.register_on_complete(
- new PG_SendMessageOnConn(
- get_parent(), reply, m->get_connection()));
- }
-
- get_parent()->queue_transaction(std::move(t));
-}
-
-void ReplicatedBackend::do_pull(OpRequestRef op)
-{
- MOSDPGPull *m = static_cast<MOSDPGPull *>(op->get_nonconst_req());
- assert(m->get_type() == MSG_OSD_PG_PULL);
- pg_shard_t from = m->from;
-
- map<pg_shard_t, vector<PushOp> > replies;
- vector<PullOp> pulls;
- m->take_pulls(&pulls);
- for (auto& i : pulls) {
- replies[from].push_back(PushOp());
- handle_pull(from, i, &(replies[from].back()));
- }
- send_pushes(m->get_priority(), replies);
-}
-
-void ReplicatedBackend::do_push_reply(OpRequestRef op)
-{
- const MOSDPGPushReply *m = static_cast<const MOSDPGPushReply *>(op->get_req());
- assert(m->get_type() == MSG_OSD_PG_PUSH_REPLY);
- pg_shard_t from = m->from;
-
- vector<PushOp> replies(1);
- for (vector<PushReplyOp>::const_iterator i = m->replies.begin();
- i != m->replies.end();
- ++i) {
- bool more = handle_push_reply(from, *i, &(replies.back()));
- if (more)
- replies.push_back(PushOp());
- }
- replies.erase(replies.end() - 1);
-
- map<pg_shard_t, vector<PushOp> > _replies;
- _replies[from].swap(replies);
- send_pushes(m->get_priority(), _replies);
-}
-
-Message * ReplicatedBackend::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_hist,
- ObjectStore::Transaction &op_t,
- pg_shard_t peer,
- const pg_info_t &pinfo)
-{
- int acks_wanted = CEPH_OSD_FLAG_ACK | CEPH_OSD_FLAG_ONDISK;
- // forward the write/update/whatever
- MOSDRepOp *wr = new MOSDRepOp(
- reqid, parent->whoami_shard(),
- spg_t(get_info().pgid.pgid, peer.shard),
- soid, acks_wanted,
- get_osdmap()->get_epoch(),
- parent->get_last_peering_reset_epoch(),
- tid, at_version);
-
- // ship resulting transaction, log entries, and pg_stats
- if (!parent->should_send_op(peer, soid)) {
- dout(10) << "issue_repop shipping empty opt to osd." << peer
- <<", object " << soid
- << " beyond MAX(last_backfill_started "
- << ", pinfo.last_backfill "
- << pinfo.last_backfill << ")" << dendl;
- ObjectStore::Transaction t;
- ::encode(t, wr->get_data());
- } else {
- ::encode(op_t, wr->get_data());
- wr->get_header().data_off = op_t.get_data_alignment();
- }
-
- ::encode(log_entries, wr->logbl);
-
- if (pinfo.is_incomplete())
- wr->pg_stats = pinfo.stats; // reflects backfill progress
- else
- wr->pg_stats = get_info().stats;
-
- wr->pg_trim_to = pg_trim_to;
- wr->pg_roll_forward_to = pg_roll_forward_to;
-
- wr->new_temp_oid = new_temp_oid;
- wr->discard_temp_oid = discard_temp_oid;
- wr->updated_hit_set_history = hset_hist;
- return wr;
-}
-
-void ReplicatedBackend::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_hist,
- InProgressOp *op,
- ObjectStore::Transaction &op_t)
-{
- if (op->op)
- op->op->pg_trace.event("issue replication ops");
-
- if (parent->get_actingbackfill_shards().size() > 1) {
- ostringstream ss;
- set<pg_shard_t> replicas = parent->get_actingbackfill_shards();
- replicas.erase(parent->whoami_shard());
- ss << "waiting for subops from " << replicas;
- if (op->op)
- op->op->mark_sub_op_sent(ss.str());
- }
- for (set<pg_shard_t>::const_iterator i =
- parent->get_actingbackfill_shards().begin();
- i != parent->get_actingbackfill_shards().end();
- ++i) {
- if (*i == parent->whoami_shard()) continue;
- pg_shard_t peer = *i;
- const pg_info_t &pinfo = parent->get_shard_info().find(peer)->second;
-
- Message *wr;
- wr = generate_subop(
- soid,
- at_version,
- tid,
- reqid,
- pg_trim_to,
- pg_roll_forward_to,
- new_temp_oid,
- discard_temp_oid,
- log_entries,
- hset_hist,
- op_t,
- peer,
- pinfo);
- if (op->op)
- wr->trace.init("replicated op", nullptr, &op->op->pg_trace);
- get_parent()->send_message_osd_cluster(
- peer.osd, wr, get_osdmap()->get_epoch());
- }
-}
-
-// sub op modify
-void ReplicatedBackend::do_repop(OpRequestRef op)
-{
- static_cast<MOSDRepOp*>(op->get_nonconst_req())->finish_decode();
- const MOSDRepOp *m = static_cast<const MOSDRepOp *>(op->get_req());
- int msg_type = m->get_type();
- assert(MSG_OSD_REPOP == msg_type);
-
- const hobject_t& soid = m->poid;
-
- dout(10) << __func__ << " " << soid
- << " v " << m->version
- << (m->logbl.length() ? " (transaction)" : " (parallel exec")
- << " " << m->logbl.length()
- << dendl;
-
- // sanity checks
- assert(m->map_epoch >= get_info().history.same_interval_since);
-
- // we better not be missing this.
- assert(!parent->get_log().get_missing().is_missing(soid));
-
- int ackerosd = m->get_source().num();
-
- op->mark_started();
-
- RepModifyRef rm(std::make_shared<RepModify>());
- rm->op = op;
- rm->ackerosd = ackerosd;
- rm->last_complete = get_info().last_complete;
- rm->epoch_started = get_osdmap()->get_epoch();
-
- assert(m->logbl.length());
- // shipped transaction and log entries
- vector<pg_log_entry_t> log;
-
- bufferlist::iterator p = const_cast<bufferlist&>(m->get_data()).begin();
- ::decode(rm->opt, p);
-
- if (m->new_temp_oid != hobject_t()) {
- dout(20) << __func__ << " start tracking temp " << m->new_temp_oid << dendl;
- add_temp_obj(m->new_temp_oid);
- }
- if (m->discard_temp_oid != hobject_t()) {
- dout(20) << __func__ << " stop tracking temp " << m->discard_temp_oid << dendl;
- if (rm->opt.empty()) {
- dout(10) << __func__ << ": removing object " << m->discard_temp_oid
- << " since we won't get the transaction" << dendl;
- rm->localt.remove(coll, ghobject_t(m->discard_temp_oid));
- }
- clear_temp_obj(m->discard_temp_oid);
- }
-
- p = const_cast<bufferlist&>(m->logbl).begin();
- ::decode(log, p);
- rm->opt.set_fadvise_flag(CEPH_OSD_OP_FLAG_FADVISE_DONTNEED);
-
- bool update_snaps = false;
- if (!rm->opt.empty()) {
- // If the opt is non-empty, we infer we are before
- // last_backfill (according to the primary, not our
- // not-quite-accurate value), and should update the
- // collections now. Otherwise, we do it later on push.
- update_snaps = true;
- }
- parent->update_stats(m->pg_stats);
- parent->log_operation(
- log,
- m->updated_hit_set_history,
- m->pg_trim_to,
- m->pg_roll_forward_to,
- update_snaps,
- rm->localt);
-
- rm->opt.register_on_commit(
- parent->bless_context(
- new C_OSD_RepModifyCommit(this, rm)));
- rm->localt.register_on_applied(
- parent->bless_context(
- new C_OSD_RepModifyApply(this, rm)));
- vector<ObjectStore::Transaction> tls;
- tls.reserve(2);
- tls.push_back(std::move(rm->localt));
- tls.push_back(std::move(rm->opt));
- parent->queue_transactions(tls, op);
- // op is cleaned up by oncommit/onapply when both are executed
-}
-
-void ReplicatedBackend::repop_applied(RepModifyRef rm)
-{
- rm->op->mark_event("sub_op_applied");
- rm->applied = true;
- rm->op->pg_trace.event("sup_op_applied");
-
- dout(10) << __func__ << " on " << rm << " op "
- << *rm->op->get_req() << dendl;
- const Message *m = rm->op->get_req();
- const MOSDRepOp *req = static_cast<const MOSDRepOp*>(m);
- eversion_t version = req->version;
-
- // send ack to acker only if we haven't sent a commit already
- if (!rm->committed) {
- Message *ack = new MOSDRepOpReply(
- req, parent->whoami_shard(),
- 0, get_osdmap()->get_epoch(), req->min_epoch, CEPH_OSD_FLAG_ACK);
- ack->set_priority(CEPH_MSG_PRIO_HIGH); // this better match commit priority!
- ack->trace = rm->op->pg_trace;
- get_parent()->send_message_osd_cluster(
- rm->ackerosd, ack, get_osdmap()->get_epoch());
- }
-
- parent->op_applied(version);
-}
-
-void ReplicatedBackend::repop_commit(RepModifyRef rm)
-{
- rm->op->mark_commit_sent();
- rm->op->pg_trace.event("sup_op_commit");
- rm->committed = true;
-
- // send commit.
- const MOSDRepOp *m = static_cast<const MOSDRepOp*>(rm->op->get_req());
- assert(m->get_type() == MSG_OSD_REPOP);
- dout(10) << __func__ << " on op " << *m
- << ", sending commit to osd." << rm->ackerosd
- << dendl;
- assert(get_osdmap()->is_up(rm->ackerosd));
-
- get_parent()->update_last_complete_ondisk(rm->last_complete);
-
- MOSDRepOpReply *reply = new MOSDRepOpReply(
- m,
- get_parent()->whoami_shard(),
- 0, get_osdmap()->get_epoch(), m->get_min_epoch(), CEPH_OSD_FLAG_ONDISK);
- reply->set_last_complete_ondisk(rm->last_complete);
- reply->set_priority(CEPH_MSG_PRIO_HIGH); // this better match ack priority!
- reply->trace = rm->op->pg_trace;
- get_parent()->send_message_osd_cluster(
- rm->ackerosd, reply, get_osdmap()->get_epoch());
-
- log_subop_stats(get_parent()->get_logger(), rm->op, l_osd_sop_w);
-}
-
-
-// ===========================================================
-
-void ReplicatedBackend::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 &manager)
-{
- dout(10) << "calc_head_subsets " << head
- << " clone_overlap " << snapset.clone_overlap << dendl;
-
- uint64_t size = obc->obs.oi.size;
- if (size)
- data_subset.insert(0, size);
-
- if (get_parent()->get_pool().allow_incomplete_clones()) {
- dout(10) << __func__ << ": caching (was) enabled, skipping clone subsets" << dendl;
- return;
- }
-
- if (!cct->_conf->osd_recover_clone_overlap) {
- dout(10) << "calc_head_subsets " << head << " -- osd_recover_clone_overlap disabled" << dendl;
- return;
- }
-
-
- interval_set<uint64_t> cloning;
- interval_set<uint64_t> prev;
- if (size)
- prev.insert(0, size);
-
- for (int j=snapset.clones.size()-1; j>=0; j--) {
- hobject_t c = head;
- c.snap = snapset.clones[j];
- prev.intersection_of(snapset.clone_overlap[snapset.clones[j]]);
- if (!missing.is_missing(c) &&
- c < last_backfill &&
- get_parent()->try_lock_for_read(c, manager)) {
- dout(10) << "calc_head_subsets " << head << " has prev " << c
- << " overlap " << prev << dendl;
- clone_subsets[c] = prev;
- cloning.union_of(prev);
- break;
- }
- dout(10) << "calc_head_subsets " << head << " does not have prev " << c
- << " overlap " << prev << dendl;
- }
-
-
- if (cloning.num_intervals() > cct->_conf->osd_recover_clone_overlap_limit) {
- dout(10) << "skipping clone, too many holes" << dendl;
- get_parent()->release_locks(manager);
- clone_subsets.clear();
- cloning.clear();
- }
-
- // what's left for us to push?
- data_subset.subtract(cloning);
-
- dout(10) << "calc_head_subsets " << head
- << " data_subset " << data_subset
- << " clone_subsets " << clone_subsets << dendl;
-}
-
-void ReplicatedBackend::calc_clone_subsets(
- SnapSet& snapset, const hobject_t& soid,
- 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 &manager)
-{
- dout(10) << "calc_clone_subsets " << soid
- << " clone_overlap " << snapset.clone_overlap << dendl;
-
- uint64_t size = snapset.clone_size[soid.snap];
- if (size)
- data_subset.insert(0, size);
-
- if (get_parent()->get_pool().allow_incomplete_clones()) {
- dout(10) << __func__ << ": caching (was) enabled, skipping clone subsets" << dendl;
- return;
- }
-
- if (!cct->_conf->osd_recover_clone_overlap) {
- dout(10) << "calc_clone_subsets " << soid << " -- osd_recover_clone_overlap disabled" << dendl;
- return;
- }
-
- unsigned i;
- for (i=0; i < snapset.clones.size(); i++)
- if (snapset.clones[i] == soid.snap)
- break;
-
- // any overlap with next older clone?
- interval_set<uint64_t> cloning;
- interval_set<uint64_t> prev;
- if (size)
- prev.insert(0, size);
- for (int j=i-1; j>=0; j--) {
- hobject_t c = soid;
- c.snap = snapset.clones[j];
- prev.intersection_of(snapset.clone_overlap[snapset.clones[j]]);
- if (!missing.is_missing(c) &&
- c < last_backfill &&
- get_parent()->try_lock_for_read(c, manager)) {
- dout(10) << "calc_clone_subsets " << soid << " has prev " << c
- << " overlap " << prev << dendl;
- clone_subsets[c] = prev;
- cloning.union_of(prev);
- break;
- }
- dout(10) << "calc_clone_subsets " << soid << " does not have prev " << c
- << " overlap " << prev << dendl;
- }
-
- // overlap with next newest?
- interval_set<uint64_t> next;
- if (size)
- next.insert(0, size);
- for (unsigned j=i+1; j<snapset.clones.size(); j++) {
- hobject_t c = soid;
- c.snap = snapset.clones[j];
- next.intersection_of(snapset.clone_overlap[snapset.clones[j-1]]);
- if (!missing.is_missing(c) &&
- c < last_backfill &&
- get_parent()->try_lock_for_read(c, manager)) {
- dout(10) << "calc_clone_subsets " << soid << " has next " << c
- << " overlap " << next << dendl;
- clone_subsets[c] = next;
- cloning.union_of(next);
- break;
- }
- dout(10) << "calc_clone_subsets " << soid << " does not have next " << c
- << " overlap " << next << dendl;
- }
-
- if (cloning.num_intervals() > cct->_conf->osd_recover_clone_overlap_limit) {
- dout(10) << "skipping clone, too many holes" << dendl;
- get_parent()->release_locks(manager);
- clone_subsets.clear();
- cloning.clear();
- }
-
-
- // what's left for us to push?
- data_subset.subtract(cloning);
-
- dout(10) << "calc_clone_subsets " << soid
- << " data_subset " << data_subset
- << " clone_subsets " << clone_subsets << dendl;
-}
-
-void ReplicatedBackend::prepare_pull(
- eversion_t v,
- const hobject_t& soid,
- ObjectContextRef headctx,
- RPGHandle *h)
-{
- assert(get_parent()->get_local_missing().get_items().count(soid));
- eversion_t _v = get_parent()->get_local_missing().get_items().find(
- soid)->second.need;
- assert(_v == v);
- const map<hobject_t, set<pg_shard_t>> &missing_loc(
- get_parent()->get_missing_loc_shards());
- const map<pg_shard_t, pg_missing_t > &peer_missing(
- get_parent()->get_shard_missing());
- map<hobject_t, set<pg_shard_t>>::const_iterator q = missing_loc.find(soid);
- assert(q != missing_loc.end());
- assert(!q->second.empty());
-
- // pick a pullee
- vector<pg_shard_t> shuffle(q->second.begin(), q->second.end());
- random_shuffle(shuffle.begin(), shuffle.end());
- vector<pg_shard_t>::iterator p = shuffle.begin();
- assert(get_osdmap()->is_up(p->osd));
- pg_shard_t fromshard = *p;
-
- dout(7) << "pull " << soid
- << " v " << v
- << " on osds " << q->second
- << " from osd." << fromshard
- << dendl;
-
- assert(peer_missing.count(fromshard));
- const pg_missing_t &pmissing = peer_missing.find(fromshard)->second;
- if (pmissing.is_missing(soid, v)) {
- assert(pmissing.get_items().find(soid)->second.have != v);
- dout(10) << "pulling soid " << soid << " from osd " << fromshard
- << " at version " << pmissing.get_items().find(soid)->second.have
- << " rather than at version " << v << dendl;
- v = pmissing.get_items().find(soid)->second.have;
- assert(get_parent()->get_log().get_log().objects.count(soid) &&
- (get_parent()->get_log().get_log().objects.find(soid)->second->op ==
- pg_log_entry_t::LOST_REVERT) &&
- (get_parent()->get_log().get_log().objects.find(
- soid)->second->reverting_to ==
- v));
- }
-
- ObjectRecoveryInfo recovery_info;
- ObcLockManager lock_manager;
-
- if (soid.is_snap()) {
- assert(!get_parent()->get_local_missing().is_missing(
- soid.get_head()) ||
- !get_parent()->get_local_missing().is_missing(
- soid.get_snapdir()));
- assert(headctx);
- // check snapset
- SnapSetContext *ssc = headctx->ssc;
- assert(ssc);
- dout(10) << " snapset " << ssc->snapset << dendl;
- recovery_info.ss = ssc->snapset;
- calc_clone_subsets(
- ssc->snapset, soid, get_parent()->get_local_missing(),
- get_info().last_backfill,
- recovery_info.copy_subset,
- recovery_info.clone_subset,
- lock_manager);
- // FIXME: this may overestimate if we are pulling multiple clones in parallel...
- dout(10) << " pulling " << recovery_info << dendl;
-
- assert(ssc->snapset.clone_size.count(soid.snap));
- recovery_info.size = ssc->snapset.clone_size[soid.snap];
- } else {
- // pulling head or unversioned object.
- // always pull the whole thing.
- recovery_info.copy_subset.insert(0, (uint64_t)-1);
- recovery_info.size = ((uint64_t)-1);
- }
-
- h->pulls[fromshard].push_back(PullOp());
- PullOp &op = h->pulls[fromshard].back();
- op.soid = soid;
-
- op.recovery_info = recovery_info;
- op.recovery_info.soid = soid;
- op.recovery_info.version = v;
- op.recovery_progress.data_complete = false;
- op.recovery_progress.omap_complete = false;
- op.recovery_progress.data_recovered_to = 0;
- op.recovery_progress.first = true;
-
- assert(!pulling.count(soid));
- pull_from_peer[fromshard].insert(soid);
- PullInfo &pi = pulling[soid];
- pi.from = fromshard;
- pi.soid = soid;
- pi.head_ctx = headctx;
- pi.recovery_info = op.recovery_info;
- pi.recovery_progress = op.recovery_progress;
- pi.cache_dont_need = h->cache_dont_need;
- pi.lock_manager = std::move(lock_manager);
-}
-
-/*
- * intelligently push an object to a replica. make use of existing
- * clones/heads and dup data ranges where possible.
- */
-int ReplicatedBackend::prep_push_to_replica(
- ObjectContextRef obc, const hobject_t& soid, pg_shard_t peer,
- PushOp *pop, bool cache_dont_need)
-{
- const object_info_t& oi = obc->obs.oi;
- uint64_t size = obc->obs.oi.size;
-
- dout(10) << __func__ << ": " << soid << " v" << oi.version
- << " size " << size << " to osd." << peer << dendl;
-
- map<hobject_t, interval_set<uint64_t>> clone_subsets;
- interval_set<uint64_t> data_subset;
-
- ObcLockManager lock_manager;
- // are we doing a clone on the replica?
- if (soid.snap && soid.snap < CEPH_NOSNAP) {
- hobject_t head = soid;
- head.snap = CEPH_NOSNAP;
-
- // try to base push off of clones that succeed/preceed poid
- // we need the head (and current SnapSet) locally to do that.
- if (get_parent()->get_local_missing().is_missing(head)) {
- dout(15) << "push_to_replica missing head " << head << ", pushing raw clone" << dendl;
- return prep_push(obc, soid, peer, pop, cache_dont_need);
- }
- hobject_t snapdir = head;
- snapdir.snap = CEPH_SNAPDIR;
- if (get_parent()->get_local_missing().is_missing(snapdir)) {
- dout(15) << "push_to_replica missing snapdir " << snapdir
- << ", pushing raw clone" << dendl;
- return prep_push(obc, soid, peer, pop, cache_dont_need);
- }
-
- SnapSetContext *ssc = obc->ssc;
- assert(ssc);
- dout(15) << "push_to_replica snapset is " << ssc->snapset << dendl;
- pop->recovery_info.ss = ssc->snapset;
- map<pg_shard_t, pg_missing_t>::const_iterator pm =
- get_parent()->get_shard_missing().find(peer);
- assert(pm != get_parent()->get_shard_missing().end());
- map<pg_shard_t, pg_info_t>::const_iterator pi =
- get_parent()->get_shard_info().find(peer);
- assert(pi != get_parent()->get_shard_info().end());
- calc_clone_subsets(
- ssc->snapset, soid,
- pm->second,
- pi->second.last_backfill,
- data_subset, clone_subsets,
- lock_manager);
- } else if (soid.snap == CEPH_NOSNAP) {
- // pushing head or unversioned object.
- // base this on partially on replica's clones?
- SnapSetContext *ssc = obc->ssc;
- assert(ssc);
- dout(15) << "push_to_replica snapset is " << ssc->snapset << dendl;
- calc_head_subsets(
- obc,
- ssc->snapset, soid, get_parent()->get_shard_missing().find(peer)->second,
- get_parent()->get_shard_info().find(peer)->second.last_backfill,
- data_subset, clone_subsets,
- lock_manager);
- }
-
- return prep_push(
- obc,
- soid,
- peer,
- oi.version,
- data_subset,
- clone_subsets,
- pop,
- cache_dont_need,
- std::move(lock_manager));
-}
-
-int ReplicatedBackend::prep_push(ObjectContextRef obc,
- const hobject_t& soid, pg_shard_t peer,
- PushOp *pop, bool cache_dont_need)
-{
- interval_set<uint64_t> data_subset;
- if (obc->obs.oi.size)
- data_subset.insert(0, obc->obs.oi.size);
- map<hobject_t, interval_set<uint64_t>> clone_subsets;
-
- return prep_push(obc, soid, peer,
- obc->obs.oi.version, data_subset, clone_subsets,
- pop, cache_dont_need, ObcLockManager());
-}
-
-int ReplicatedBackend::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 *pop,
- bool cache_dont_need,
- ObcLockManager &&lock_manager)
-{
- get_parent()->begin_peer_recover(peer, soid);
- // take note.
- PushInfo &pi = pushing[soid][peer];
- pi.obc = obc;
- pi.recovery_info.size = obc->obs.oi.size;
- pi.recovery_info.copy_subset = data_subset;
- pi.recovery_info.clone_subset = clone_subsets;
- pi.recovery_info.soid = soid;
- pi.recovery_info.oi = obc->obs.oi;
- pi.recovery_info.ss = pop->recovery_info.ss;
- pi.recovery_info.version = version;
- pi.lock_manager = std::move(lock_manager);
-
- ObjectRecoveryProgress new_progress;
- int r = build_push_op(pi.recovery_info,
- pi.recovery_progress,
- &new_progress,
- pop,
- &(pi.stat), cache_dont_need);
- if (r < 0)
- return r;
- pi.recovery_progress = new_progress;
- return 0;
-}
-
-void ReplicatedBackend::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)
-{
- hobject_t target_oid;
- if (first && complete) {
- target_oid = recovery_info.soid;
- } else {
- target_oid = get_parent()->get_temp_recovery_object(recovery_info.soid,
- recovery_info.version);
- if (first) {
- dout(10) << __func__ << ": Adding oid "
- << target_oid << " in the temp collection" << dendl;
- add_temp_obj(target_oid);
- }
- }
-
- if (first) {
- t->remove(coll, ghobject_t(target_oid));
- t->touch(coll, ghobject_t(target_oid));
- t->truncate(coll, ghobject_t(target_oid), recovery_info.size);
- if (omap_header.length())
- t->omap_setheader(coll, ghobject_t(target_oid), omap_header);
-
- bufferlist bv = attrs.at(OI_ATTR);
- object_info_t oi(bv);
- t->set_alloc_hint(coll, ghobject_t(target_oid),
- oi.expected_object_size,
- oi.expected_write_size,
- oi.alloc_hint_flags);
- }
- uint64_t off = 0;
- uint32_t fadvise_flags = CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL;
- if (cache_dont_need)
- fadvise_flags |= CEPH_OSD_OP_FLAG_FADVISE_DONTNEED;
- for (interval_set<uint64_t>::const_iterator p = intervals_included.begin();
- p != intervals_included.end();
- ++p) {
- bufferlist bit;
- bit.substr_of(data_included, off, p.get_len());
- t->write(coll, ghobject_t(target_oid),
- p.get_start(), p.get_len(), bit, fadvise_flags);
- off += p.get_len();
- }
-
- if (!omap_entries.empty())
- t->omap_setkeys(coll, ghobject_t(target_oid), omap_entries);
- if (!attrs.empty())
- t->setattrs(coll, ghobject_t(target_oid), attrs);
-
- if (complete) {
- if (!first) {
- dout(10) << __func__ << ": Removing oid "
- << target_oid << " from the temp collection" << dendl;
- clear_temp_obj(target_oid);
- t->remove(coll, ghobject_t(recovery_info.soid));
- t->collection_move_rename(coll, ghobject_t(target_oid),
- coll, ghobject_t(recovery_info.soid));
- }
-
- submit_push_complete(recovery_info, t);
- }
-}
-
-void ReplicatedBackend::submit_push_complete(
- const ObjectRecoveryInfo &recovery_info,
- ObjectStore::Transaction *t)
-{
- for (map<hobject_t, interval_set<uint64_t>>::const_iterator p =
- recovery_info.clone_subset.begin();
- p != recovery_info.clone_subset.end();
- ++p) {
- for (interval_set<uint64_t>::const_iterator q = p->second.begin();
- q != p->second.end();
- ++q) {
- dout(15) << " clone_range " << p->first << " "
- << q.get_start() << "~" << q.get_len() << dendl;
- t->clone_range(coll, ghobject_t(p->first), ghobject_t(recovery_info.soid),
- q.get_start(), q.get_len(), q.get_start());
- }
- }
-}
-
-ObjectRecoveryInfo ReplicatedBackend::recalc_subsets(
- const ObjectRecoveryInfo& recovery_info,
- SnapSetContext *ssc,
- ObcLockManager &manager)
-{
- if (!recovery_info.soid.snap || recovery_info.soid.snap >= CEPH_NOSNAP)
- return recovery_info;
- ObjectRecoveryInfo new_info = recovery_info;
- new_info.copy_subset.clear();
- new_info.clone_subset.clear();
- assert(ssc);
- get_parent()->release_locks(manager); // might already have locks
- calc_clone_subsets(
- ssc->snapset, new_info.soid, get_parent()->get_local_missing(),
- get_info().last_backfill,
- new_info.copy_subset, new_info.clone_subset,
- manager);
- return new_info;
-}
-
-bool ReplicatedBackend::handle_pull_response(
- pg_shard_t from, const PushOp &pop, PullOp *response,
- list<pull_complete_info> *to_continue,
- ObjectStore::Transaction *t)
-{
- interval_set<uint64_t> data_included = pop.data_included;
- bufferlist data;
- data = pop.data;
- dout(10) << "handle_pull_response "
- << pop.recovery_info
- << pop.after_progress
- << " data.size() is " << data.length()
- << " data_included: " << data_included
- << dendl;
- if (pop.version == eversion_t()) {
- // replica doesn't have it!
- _failed_pull(from, pop.soid);
- return false;
- }
-
- const hobject_t &hoid = pop.soid;
- assert((data_included.empty() && data.length() == 0) ||
- (!data_included.empty() && data.length() > 0));
-
- auto piter = pulling.find(hoid);
- if (piter == pulling.end()) {
- return false;
- }
-
- PullInfo &pi = piter->second;
- if (pi.recovery_info.size == (uint64_t(-1))) {
- pi.recovery_info.size = pop.recovery_info.size;
- pi.recovery_info.copy_subset.intersection_of(
- pop.recovery_info.copy_subset);
- }
- // If primary doesn't have object info and didn't know version
- if (pi.recovery_info.version == eversion_t()) {
- pi.recovery_info.version = pop.version;
- }
-
- bool first = pi.recovery_progress.first;
- if (first) {
- // attrs only reference the origin bufferlist (decode from
- // MOSDPGPush message) whose size is much greater than attrs in
- // recovery. If obc cache it (get_obc maybe cache the attr), this
- // causes the whole origin bufferlist would not be free until obc
- // is evicted from obc cache. So rebuild the bufferlists before
- // cache it.
- auto attrset = pop.attrset;
- for (auto& a : attrset) {
- a.second.rebuild();
- }
- pi.obc = get_parent()->get_obc(pi.recovery_info.soid, attrset);
- pi.recovery_info.oi = pi.obc->obs.oi;
- pi.recovery_info = recalc_subsets(
- pi.recovery_info,
- pi.obc->ssc,
- pi.lock_manager);
- }
-
-
- interval_set<uint64_t> usable_intervals;
- bufferlist usable_data;
- trim_pushed_data(pi.recovery_info.copy_subset,
- data_included,
- data,
- &usable_intervals,
- &usable_data);
- data_included = usable_intervals;
- data.claim(usable_data);
-
-
- pi.recovery_progress = pop.after_progress;
-
- dout(10) << "new recovery_info " << pi.recovery_info
- << ", new progress " << pi.recovery_progress
- << dendl;
-
- bool complete = pi.is_complete();
-
- submit_push_data(pi.recovery_info, first,
- complete, pi.cache_dont_need,
- data_included, data,
- pop.omap_header,
- pop.attrset,
- pop.omap_entries,
- t);
-
- pi.stat.num_keys_recovered += pop.omap_entries.size();
- pi.stat.num_bytes_recovered += data.length();
-
- if (complete) {
- pi.stat.num_objects_recovered++;
- clear_pull_from(piter);
- to_continue->push_back({hoid, pi.stat});
- get_parent()->on_local_recover(
- hoid, pi.recovery_info, pi.obc, false, t);
- return false;
- } else {
- response->soid = pop.soid;
- response->recovery_info = pi.recovery_info;
- response->recovery_progress = pi.recovery_progress;
- return true;
- }
-}
-
-void ReplicatedBackend::handle_push(
- pg_shard_t from, const PushOp &pop, PushReplyOp *response,
- ObjectStore::Transaction *t)
-{
- dout(10) << "handle_push "
- << pop.recovery_info
- << pop.after_progress
- << dendl;
- bufferlist data;
- data = pop.data;
- bool first = pop.before_progress.first;
- bool complete = pop.after_progress.data_complete &&
- pop.after_progress.omap_complete;
-
- response->soid = pop.recovery_info.soid;
- submit_push_data(pop.recovery_info,
- first,
- complete,
- true, // must be replicate
- pop.data_included,
- data,
- pop.omap_header,
- pop.attrset,
- pop.omap_entries,
- t);
-
- if (complete)
- get_parent()->on_local_recover(
- pop.recovery_info.soid,
- pop.recovery_info,
- ObjectContextRef(), // ok, is replica
- false,
- t);
-}
-
-void ReplicatedBackend::send_pushes(int prio, map<pg_shard_t, vector<PushOp> > &pushes)
-{
- for (map<pg_shard_t, vector<PushOp> >::iterator i = pushes.begin();
- i != pushes.end();
- ++i) {
- ConnectionRef con = get_parent()->get_con_osd_cluster(
- i->first.osd,
- get_osdmap()->get_epoch());
- if (!con)
- continue;
- vector<PushOp>::iterator j = i->second.begin();
- while (j != i->second.end()) {
- uint64_t cost = 0;
- uint64_t pushes = 0;
- MOSDPGPush *msg = new MOSDPGPush();
- msg->from = get_parent()->whoami_shard();
- msg->pgid = get_parent()->primary_spg_t();
- msg->map_epoch = get_osdmap()->get_epoch();
- msg->min_epoch = get_parent()->get_last_peering_reset_epoch();
- msg->set_priority(prio);
- for (;
- (j != i->second.end() &&
- cost < cct->_conf->osd_max_push_cost &&
- pushes < cct->_conf->osd_max_push_objects) ;
- ++j) {
- dout(20) << __func__ << ": sending push " << *j
- << " to osd." << i->first << dendl;
- cost += j->cost(cct);
- pushes += 1;
- msg->pushes.push_back(*j);
- }
- msg->set_cost(cost);
- get_parent()->send_message_osd_cluster(msg, con);
- }
- }
-}
-
-void ReplicatedBackend::send_pulls(int prio, map<pg_shard_t, vector<PullOp> > &pulls)
-{
- for (map<pg_shard_t, vector<PullOp> >::iterator i = pulls.begin();
- i != pulls.end();
- ++i) {
- ConnectionRef con = get_parent()->get_con_osd_cluster(
- i->first.osd,
- get_osdmap()->get_epoch());
- if (!con)
- continue;
- dout(20) << __func__ << ": sending pulls " << i->second
- << " to osd." << i->first << dendl;
- MOSDPGPull *msg = new MOSDPGPull();
- msg->from = parent->whoami_shard();
- msg->set_priority(prio);
- msg->pgid = get_parent()->primary_spg_t();
- msg->map_epoch = get_osdmap()->get_epoch();
- msg->min_epoch = get_parent()->get_last_peering_reset_epoch();
- msg->set_pulls(&i->second);
- msg->compute_cost(cct);
- get_parent()->send_message_osd_cluster(msg, con);
- }
-}
-
-int ReplicatedBackend::build_push_op(const ObjectRecoveryInfo &recovery_info,
- const ObjectRecoveryProgress &progress,
- ObjectRecoveryProgress *out_progress,
- PushOp *out_op,
- object_stat_sum_t *stat,
- bool cache_dont_need)
-{
- ObjectRecoveryProgress _new_progress;
- if (!out_progress)
- out_progress = &_new_progress;
- ObjectRecoveryProgress &new_progress = *out_progress;
- new_progress = progress;
-
- dout(7) << __func__ << " " << recovery_info.soid
- << " v " << recovery_info.version
- << " size " << recovery_info.size
- << " recovery_info: " << recovery_info
- << dendl;
-
- eversion_t v = recovery_info.version;
- if (progress.first) {
- int r = store->omap_get_header(coll, ghobject_t(recovery_info.soid), &out_op->omap_header);
- if(r < 0) {
- dout(1) << __func__ << " get omap header failed: " << cpp_strerror(-r) << dendl;
- return r;
- }
- r = store->getattrs(ch, ghobject_t(recovery_info.soid), out_op->attrset);
- if(r < 0) {
- dout(1) << __func__ << " getattrs failed: " << cpp_strerror(-r) << dendl;
- return r;
- }
-
- // Debug
- bufferlist bv = out_op->attrset[OI_ATTR];
- object_info_t oi;
- try {
- bufferlist::iterator bliter = bv.begin();
- ::decode(oi, bliter);
- } catch (...) {
- dout(0) << __func__ << ": bad object_info_t: " << recovery_info.soid << dendl;
- return -EINVAL;
- }
-
- // If requestor didn't know the version, use ours
- if (v == eversion_t()) {
- v = oi.version;
- } else if (oi.version != v) {
- get_parent()->clog_error() << get_info().pgid << " push "
- << recovery_info.soid << " v "
- << recovery_info.version
- << " failed because local copy is "
- << oi.version;
- return -EINVAL;
- }
-
- new_progress.first = false;
- }
- // Once we provide the version subsequent requests will have it, so
- // at this point it must be known.
- assert(v != eversion_t());
-
- uint64_t available = cct->_conf->osd_recovery_max_chunk;
- if (!progress.omap_complete) {
- ObjectMap::ObjectMapIterator iter =
- store->get_omap_iterator(coll,
- ghobject_t(recovery_info.soid));
- assert(iter);
- for (iter->lower_bound(progress.omap_recovered_to);
- iter->valid();
- iter->next(false)) {
- if (!out_op->omap_entries.empty() &&
- ((cct->_conf->osd_recovery_max_omap_entries_per_chunk > 0 &&
- out_op->omap_entries.size() >= cct->_conf->osd_recovery_max_omap_entries_per_chunk) ||
- available <= iter->key().size() + iter->value().length()))
- break;
- out_op->omap_entries.insert(make_pair(iter->key(), iter->value()));
-
- if ((iter->key().size() + iter->value().length()) <= available)
- available -= (iter->key().size() + iter->value().length());
- else
- available = 0;
- }
- if (!iter->valid())
- new_progress.omap_complete = true;
- else
- new_progress.omap_recovered_to = iter->key();
- }
-
- if (available > 0) {
- if (!recovery_info.copy_subset.empty()) {
- interval_set<uint64_t> copy_subset = recovery_info.copy_subset;
- map<uint64_t, uint64_t> m;
- int r = store->fiemap(ch, ghobject_t(recovery_info.soid), 0,
- copy_subset.range_end(), m);
- if (r >= 0) {
- interval_set<uint64_t> fiemap_included(m);
- copy_subset.intersection_of(fiemap_included);
- } else {
- // intersection of copy_subset and empty interval_set would be empty anyway
- copy_subset.clear();
- }
-
- out_op->data_included.span_of(copy_subset, progress.data_recovered_to,
- available);
- if (out_op->data_included.empty()) // zero filled section, skip to end!
- new_progress.data_recovered_to = recovery_info.copy_subset.range_end();
- else
- new_progress.data_recovered_to = out_op->data_included.range_end();
- }
- } else {
- out_op->data_included.clear();
- }
-
- for (interval_set<uint64_t>::iterator p = out_op->data_included.begin();
- p != out_op->data_included.end();
- ++p) {
- bufferlist bit;
- int r = store->read(ch, ghobject_t(recovery_info.soid),
- p.get_start(), p.get_len(), bit,
- cache_dont_need ? CEPH_OSD_OP_FLAG_FADVISE_DONTNEED: 0);
- if (cct->_conf->osd_debug_random_push_read_error &&
- (rand() % (int)(cct->_conf->osd_debug_random_push_read_error * 100.0)) == 0) {
- dout(0) << __func__ << ": inject EIO " << recovery_info.soid << dendl;
- r = -EIO;
- }
- if (r < 0) {
- return r;
- }
- if (p.get_len() != bit.length()) {
- dout(10) << " extent " << p.get_start() << "~" << p.get_len()
- << " is actually " << p.get_start() << "~" << bit.length()
- << dendl;
- interval_set<uint64_t>::iterator save = p++;
- if (bit.length() == 0)
- out_op->data_included.erase(save); //Remove this empty interval
- else
- save.set_len(bit.length());
- // Remove any other intervals present
- while (p != out_op->data_included.end()) {
- interval_set<uint64_t>::iterator save = p++;
- out_op->data_included.erase(save);
- }
- new_progress.data_complete = true;
- out_op->data.claim_append(bit);
- break;
- }
- out_op->data.claim_append(bit);
- }
-
- if (new_progress.is_complete(recovery_info)) {
- new_progress.data_complete = true;
- if (stat)
- stat->num_objects_recovered++;
- }
-
- if (stat) {
- stat->num_keys_recovered += out_op->omap_entries.size();
- stat->num_bytes_recovered += out_op->data.length();
- }
-
- get_parent()->get_logger()->inc(l_osd_push);
- get_parent()->get_logger()->inc(l_osd_push_outb, out_op->data.length());
-
- // send
- out_op->version = v;
- out_op->soid = recovery_info.soid;
- out_op->recovery_info = recovery_info;
- out_op->after_progress = new_progress;
- out_op->before_progress = progress;
- return 0;
-}
-
-void ReplicatedBackend::prep_push_op_blank(const hobject_t& soid, PushOp *op)
-{
- op->recovery_info.version = eversion_t();
- op->version = eversion_t();
- op->soid = soid;
-}
-
-bool ReplicatedBackend::handle_push_reply(
- pg_shard_t peer, const PushReplyOp &op, PushOp *reply)
-{
- const hobject_t &soid = op.soid;
- if (pushing.count(soid) == 0) {
- dout(10) << "huh, i wasn't pushing " << soid << " to osd." << peer
- << ", or anybody else"
- << dendl;
- return false;
- } else if (pushing[soid].count(peer) == 0) {
- dout(10) << "huh, i wasn't pushing " << soid << " to osd." << peer
- << dendl;
- return false;
- } else {
- PushInfo *pi = &pushing[soid][peer];
- bool error = pushing[soid].begin()->second.recovery_progress.error;
-
- if (!pi->recovery_progress.data_complete && !error) {
- dout(10) << " pushing more from, "
- << pi->recovery_progress.data_recovered_to
- << " of " << pi->recovery_info.copy_subset << dendl;
- ObjectRecoveryProgress new_progress;
- int r = build_push_op(
- pi->recovery_info,
- pi->recovery_progress, &new_progress, reply,
- &(pi->stat));
- // Handle the case of a read error right after we wrote, which is
- // hopefuilly extremely rare.
- if (r < 0) {
- dout(5) << __func__ << ": oid " << soid << " error " << r << dendl;
-
- error = true;
- goto done;
- }
- pi->recovery_progress = new_progress;
- return true;
- } else {
- // done!
-done:
- if (!error)
- get_parent()->on_peer_recover( peer, soid, pi->recovery_info);
-
- get_parent()->release_locks(pi->lock_manager);
- object_stat_sum_t stat = pi->stat;
- eversion_t v = pi->recovery_info.version;
- pushing[soid].erase(peer);
- pi = NULL;
-
- if (pushing[soid].empty()) {
- if (!error)
- get_parent()->on_global_recover(soid, stat, false);
- else
- get_parent()->on_primary_error(soid, v);
- pushing.erase(soid);
- } else {
- // This looks weird, but we erased the current peer and need to remember
- // the error on any other one, while getting more acks.
- if (error)
- pushing[soid].begin()->second.recovery_progress.error = true;
- dout(10) << "pushed " << soid << ", still waiting for push ack from "
- << pushing[soid].size() << " others" << dendl;
- }
- return false;
- }
- }
-}
-
-void ReplicatedBackend::handle_pull(pg_shard_t peer, PullOp &op, PushOp *reply)
-{
- const hobject_t &soid = op.soid;
- struct stat st;
- int r = store->stat(ch, ghobject_t(soid), &st);
- if (r != 0) {
- get_parent()->clog_error() << get_info().pgid << " "
- << peer << " tried to pull " << soid
- << " but got " << cpp_strerror(-r);
- prep_push_op_blank(soid, reply);
- } else {
- ObjectRecoveryInfo &recovery_info = op.recovery_info;
- ObjectRecoveryProgress &progress = op.recovery_progress;
- if (progress.first && recovery_info.size == ((uint64_t)-1)) {
- // Adjust size and copy_subset
- recovery_info.size = st.st_size;
- recovery_info.copy_subset.clear();
- if (st.st_size)
- recovery_info.copy_subset.insert(0, st.st_size);
- assert(recovery_info.clone_subset.empty());
- }
-
- r = build_push_op(recovery_info, progress, 0, reply);
- if (r < 0)
- prep_push_op_blank(soid, reply);
- }
-}
-
-/**
- * trim received data to remove what we don't want
- *
- * @param copy_subset intervals we want
- * @param data_included intervals we got
- * @param data_recieved data we got
- * @param intervals_usable intervals we want to keep
- * @param data_usable matching data we want to keep
- */
-void ReplicatedBackend::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)
-{
- if (intervals_received.subset_of(copy_subset)) {
- *intervals_usable = intervals_received;
- *data_usable = data_received;
- return;
- }
-
- intervals_usable->intersection_of(copy_subset,
- intervals_received);
-
- uint64_t off = 0;
- for (interval_set<uint64_t>::const_iterator p = intervals_received.begin();
- p != intervals_received.end();
- ++p) {
- interval_set<uint64_t> x;
- x.insert(p.get_start(), p.get_len());
- x.intersection_of(copy_subset);
- for (interval_set<uint64_t>::const_iterator q = x.begin();
- q != x.end();
- ++q) {
- bufferlist sub;
- uint64_t data_off = off + (q.get_start() - p.get_start());
- sub.substr_of(data_received, data_off, q.get_len());
- data_usable->claim_append(sub);
- }
- off += p.get_len();
- }
-}
-
-void ReplicatedBackend::_failed_pull(pg_shard_t from, const hobject_t &soid)
-{
- dout(20) << __func__ << ": " << soid << " from " << from << dendl;
- list<pg_shard_t> fl = { from };
- get_parent()->failed_push(fl, soid);
-
- clear_pull(pulling.find(soid));
-}
-
-void ReplicatedBackend::clear_pull_from(
- map<hobject_t, PullInfo>::iterator piter)
-{
- auto from = piter->second.from;
- pull_from_peer[from].erase(piter->second.soid);
- if (pull_from_peer[from].empty())
- pull_from_peer.erase(from);
-}
-
-void ReplicatedBackend::clear_pull(
- map<hobject_t, PullInfo>::iterator piter,
- bool clear_pull_from_peer)
-{
- if (clear_pull_from_peer) {
- clear_pull_from(piter);
- }
- get_parent()->release_locks(piter->second.lock_manager);
- pulling.erase(piter);
-}
-
-int ReplicatedBackend::start_pushes(
- const hobject_t &soid,
- ObjectContextRef obc,
- RPGHandle *h)
-{
- list< map<pg_shard_t, pg_missing_t>::const_iterator > shards;
-
- dout(20) << __func__ << " soid " << soid << dendl;
- // who needs it?
- assert(get_parent()->get_actingbackfill_shards().size() > 0);
- for (set<pg_shard_t>::iterator i =
- get_parent()->get_actingbackfill_shards().begin();
- i != get_parent()->get_actingbackfill_shards().end();
- ++i) {
- if (*i == get_parent()->whoami_shard()) continue;
- pg_shard_t peer = *i;
- map<pg_shard_t, pg_missing_t>::const_iterator j =
- get_parent()->get_shard_missing().find(peer);
- assert(j != get_parent()->get_shard_missing().end());
- if (j->second.is_missing(soid)) {
- shards.push_back(j);
- }
- }
-
- // If more than 1 read will occur ignore possible request to not cache
- bool cache = shards.size() == 1 ? h->cache_dont_need : false;
-
- for (auto j : shards) {
- pg_shard_t peer = j->first;
- h->pushes[peer].push_back(PushOp());
- int r = prep_push_to_replica(obc, soid, peer,
- &(h->pushes[peer].back()), cache);
- if (r < 0) {
- // Back out all failed reads
- for (auto k : shards) {
- pg_shard_t p = k->first;
- dout(10) << __func__ << " clean up peer " << p << dendl;
- h->pushes[p].pop_back();
- if (p == peer) break;
- }
- return r;
- }
- }
- return shards.size();
-}
Code Review / stor4nfv.git / blobdiff