RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2025-01-02 17:12:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

osd: remove MOSDSubOp[Reply] jewel compat code

Browse Source 
Signed-off-by: Sage Weil <sage@redhat.com>
This commit is contained in:
Sage Weil 2017-08-24 12:34:18 -04:00
parent 0ccd074398
commit 1f680c3e96
14 changed files with 54 additions and 780 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

295
src/messages/MOSDSubOp.h
View File

@ -1,295 +0,0 @@
// -*- 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) 2004-2006 Sage Weil <sage@newdream.net>
*
* 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 CEPH_MOSDSUBOP_H
#define CEPH_MOSDSUBOP_H
#include "MOSDFastDispatchOp.h"
#include "include/ceph_features.h"
/*
* OSD sub op - for internal ops on pobjects between primary and replicas(/stripes/whatever)
*/
class MOSDSubOp : public MOSDFastDispatchOp {
static const int HEAD_VERSION = 12;
static const int COMPAT_VERSION = 7;
public:
epoch_t map_epoch = 0;
// metadata from original request
osd_reqid_t reqid;
// subop
pg_shard_t from;
spg_t pgid;
hobject_t poid;
object_locator_t oloc;
__u8 acks_wanted = 0;
// op to exec
vector<OSDOp> ops;
utime_t mtime;
bool old_exists = false;
uint64_t old_size = 0;
eversion_t old_version;
SnapSet snapset;
// transaction to exec
bufferlist logbl;
pg_stat_t pg_stats;
// subop metadata
eversion_t version;
// piggybacked osd/og state
eversion_t pg_trim_to; // primary->replica: trim to here
eversion_t pg_roll_forward_to; // primary->replica: trim rollback
// info to here
osd_peer_stat_t peer_stat;
map<string,bufferlist> attrset;
interval_set<uint64_t> data_subset;
map<hobject_t, interval_set<uint64_t>> clone_subsets;
bool first = false, complete = false;
bool bdata_encode;
interval_set<uint64_t> data_included;
ObjectRecoveryInfo recovery_info;
// reflects result of current push
ObjectRecoveryProgress recovery_progress;
// reflects progress before current push
ObjectRecoveryProgress current_progress;
map<string,bufferlist> omap_entries;
bufferlist omap_header;
hobject_t new_temp_oid; ///< new temp object that we must now start tracking
hobject_t discard_temp_oid; ///< previously used temp object that we can now stop tracking
/// non-empty if this transaction involves a hit_set history update
boost::optional<pg_hit_set_history_t> updated_hit_set_history;
epoch_t get_map_epoch() const override {
return map_epoch;
}
spg_t get_spg() const override {
return pgid;
}
int get_cost() const override {
if (ops.size() == 1 && ops[0].op.op == CEPH_OSD_OP_PULL)
return ops[0].op.extent.length;
return data.length();
}
void decode_payload() override {
//since we drop incorrect_pools flag, now we only support
//version >=7
assert (header.version >= 7);
bufferlist::iterator p = payload.begin();
::decode(map_epoch, p);
::decode(reqid, p);
::decode(pgid.pgid, p);
::decode(poid, p);
__u32 num_ops;
::decode(num_ops, p);
ops.resize(num_ops);
unsigned off = 0;
for (unsigned i = 0; i < num_ops; i++) {
::decode(ops[i].op, p);
ops[i].indata.substr_of(data, off, ops[i].op.payload_len);
off += ops[i].op.payload_len;
}
::decode(mtime, p);
//we don't need noop anymore
bool noop_dont_need;
::decode(noop_dont_need, p);
::decode(acks_wanted, p);
::decode(version, p);
::decode(old_exists, p);
::decode(old_size, p);
::decode(old_version, p);
::decode(snapset, p);
if (header.version <= 11) {
SnapContext snapc_dont_need;
::decode(snapc_dont_need, p);
}
::decode(logbl, p);
::decode(pg_stats, p);
::decode(pg_trim_to, p);
::decode(peer_stat, p);
::decode(attrset, p);
::decode(data_subset, p);
::decode(clone_subsets, p);
::decode(first, p);
::decode(complete, p);
::decode(oloc, p);
::decode(data_included, p);
recovery_info.decode(p, pgid.pool());
::decode(recovery_progress, p);
::decode(current_progress, p);
::decode(omap_entries, p);
::decode(omap_header, p);
if (header.version >= 8) {
::decode(new_temp_oid, p);
::decode(discard_temp_oid, p);
}
if (header.version >= 9) {
::decode(from, p);
::decode(pgid.shard, p);
} else {
from = pg_shard_t(
get_source().num(),
shard_id_t::NO_SHARD);
pgid.shard = shard_id_t::NO_SHARD;
}
if (header.version >= 10) {
::decode(updated_hit_set_history, p);
}
if (header.version >= 11) {
::decode(pg_roll_forward_to, p);
} else {
pg_roll_forward_to = pg_trim_to;
}
}
void finish_decode() { }
void encode_payload(uint64_t features) override {
header.version = HEAD_VERSION;
::encode(map_epoch, payload);
::encode(reqid, payload);
::encode(pgid.pgid, payload);
::encode(poid, payload);
__u32 num_ops = ops.size();
::encode(num_ops, payload);
for (unsigned i = 0; i < ops.size(); i++) {
ops[i].op.payload_len = ops[i].indata.length();
::encode(ops[i].op, payload);
if(false == bdata_encode) {
data.append(ops[i].indata);
bdata_encode = true;
}
}
::encode(mtime, payload);
//encode a false here for backward compatiable
::encode(false, payload);
::encode(acks_wanted, payload);
::encode(version, payload);
::encode(old_exists, payload);
::encode(old_size, payload);
::encode(old_version, payload);
::encode(snapset, payload);
if ((features & CEPH_FEATURE_OSDSUBOP_NO_SNAPCONTEXT) == 0) {
header.version = 11;
SnapContext dummy_snapc;
::encode(dummy_snapc, payload);
}
::encode(logbl, payload);
::encode(pg_stats, payload);
::encode(pg_trim_to, payload);
::encode(peer_stat, payload);
::encode(attrset, payload);
::encode(data_subset, payload);
::encode(clone_subsets, payload);
if (ops.size())
header.data_off = ops[0].op.extent.offset;
else
header.data_off = 0;
::encode(first, payload);
::encode(complete, payload);
::encode(oloc, payload);
::encode(data_included, payload);
::encode(recovery_info, payload, features);
::encode(recovery_progress, payload);
::encode(current_progress, payload);
::encode(omap_entries, payload);
::encode(omap_header, payload);
::encode(new_temp_oid, payload);
::encode(discard_temp_oid, payload);
::encode(from, payload);
::encode(pgid.shard, payload);
::encode(updated_hit_set_history, payload);
::encode(pg_roll_forward_to, payload);
}
MOSDSubOp()
: MOSDFastDispatchOp(MSG_OSD_SUBOP, HEAD_VERSION, COMPAT_VERSION),
bdata_encode(false) { }
MOSDSubOp(osd_reqid_t r, pg_shard_t from,
spg_t p, const hobject_t& po, int aw,
epoch_t mape, ceph_tid_t rtid, eversion_t v)
: MOSDFastDispatchOp(MSG_OSD_SUBOP, HEAD_VERSION, COMPAT_VERSION),
map_epoch(mape),
reqid(r),
from(from),
pgid(p),
poid(po),
acks_wanted(aw),
old_exists(false), old_size(0),
version(v),
first(false), complete(false),
bdata_encode(false) {
memset(&peer_stat, 0, sizeof(peer_stat));
set_tid(rtid);
}
private:
~MOSDSubOp() override {}
public:
const char *get_type_name() const override { return "osd_sub_op"; }
void print(ostream& out) const override {
out << "osd_sub_op(" << reqid
<< " " << pgid
<< " " << poid
<< " " << ops;
if (first)
out << " first";
if (complete)
out << " complete";
out << " v " << version
<< " snapset=" << snapset;
if (!data_subset.empty()) out << " subset " << data_subset;
if (updated_hit_set_history)
out << ", has_updated_hit_set_history";
out << ")";
}
};
#endif

177
src/messages/MOSDSubOpReply.h
View File

@ -1,177 +0,0 @@
// -*- 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) 2004-2006 Sage Weil <sage@newdream.net>
*
* 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 CEPH_MOSDSUBOPREPLY_H
#define CEPH_MOSDSUBOPREPLY_H
#include "MOSDFastDispatchOp.h"
#include "MOSDSubOp.h"
#include "os/ObjectStore.h"
/*
* OSD op reply
*
* oid - object id
* op - OSD_OP_DELETE, etc.
*
*/
class MOSDSubOpReply : public MOSDFastDispatchOp {
static const int HEAD_VERSION = 2;
static const int COMPAT_VERSION = 1;
public:
epoch_t map_epoch = 0;
// subop metadata
osd_reqid_t reqid;
pg_shard_t from;
spg_t pgid;
hobject_t poid;
vector<OSDOp> ops;
// result
__u8 ack_type = 0;
int32_t result = 0;
// piggybacked osd state
eversion_t last_complete_ondisk;
osd_peer_stat_t peer_stat;
map<string,bufferptr> attrset;
epoch_t get_map_epoch() const override {
return map_epoch;
}
spg_t get_spg() const override {
return pgid;
}
void decode_payload() override {
bufferlist::iterator p = payload.begin();
::decode(map_epoch, p);
::decode(reqid, p);
::decode(pgid.pgid, p);
::decode(poid, p);
unsigned num_ops;
::decode(num_ops, p);
ops.resize(num_ops);
for (unsigned i = 0; i < num_ops; i++) {
::decode(ops[i].op, p);
}
::decode(ack_type, p);
::decode(result, p);
::decode(last_complete_ondisk, p);
::decode(peer_stat, p);
::decode(attrset, p);
if (!poid.is_max() && poid.pool == -1)
poid.pool = pgid.pool();
if (header.version >= 2) {
::decode(from, p);
::decode(pgid.shard, p);
} else {
from = pg_shard_t(
get_source().num(),
shard_id_t::NO_SHARD);
pgid.shard = shard_id_t::NO_SHARD;
}
}
void finish_decode() { }
void encode_payload(uint64_t features) override {
::encode(map_epoch, payload);
::encode(reqid, payload);
::encode(pgid.pgid, payload);
::encode(poid, payload);
__u32 num_ops = ops.size();
::encode(num_ops, payload);
for (unsigned i = 0; i < ops.size(); i++) {
::encode(ops[i].op, payload);
}
::encode(ack_type, payload);
::encode(result, payload);
::encode(last_complete_ondisk, payload);
::encode(peer_stat, payload);
::encode(attrset, payload);
::encode(from, payload);
::encode(pgid.shard, payload);
}
epoch_t get_map_epoch() { return map_epoch; }
spg_t get_pg() const { return pgid; }
const hobject_t& get_poid() const { return poid; }
int get_ack_type() { return ack_type; }
bool is_ondisk() { return ack_type & CEPH_OSD_FLAG_ONDISK; }
bool is_onnvram() { return ack_type & CEPH_OSD_FLAG_ONNVRAM; }
int get_result() { return result; }
void set_last_complete_ondisk(eversion_t v) { last_complete_ondisk = v; }
eversion_t get_last_complete_ondisk() { return last_complete_ondisk; }
void set_peer_stat(const osd_peer_stat_t& stat) { peer_stat = stat; }
const osd_peer_stat_t& get_peer_stat() { return peer_stat; }
void set_attrset(map<string,bufferptr> &as) { attrset = as; }
map<string,bufferptr>& get_attrset() { return attrset; }
public:
MOSDSubOpReply(
const MOSDSubOp *req, pg_shard_t from, int result_, epoch_t e, int at)
: MOSDFastDispatchOp(MSG_OSD_SUBOPREPLY, HEAD_VERSION, COMPAT_VERSION),
map_epoch(e),
reqid(req->reqid),
from(from),
pgid(req->pgid.pgid, req->from.shard),
poid(req->poid),
ops(req->ops),
ack_type(at),
result(result_) {
memset(&peer_stat, 0, sizeof(peer_stat));
set_tid(req->get_tid());
}
MOSDSubOpReply()
: MOSDFastDispatchOp(MSG_OSD_SUBOPREPLY, HEAD_VERSION, COMPAT_VERSION) {}
private:
~MOSDSubOpReply() override {}
public:
const char *get_type_name() const override { return "osd_subop_reply"; }
void print(ostream& out) const override {
out << "osd_sub_op_reply(" << reqid
<< " " << pgid
<< " " << poid << " " << ops;
if (ack_type & CEPH_OSD_FLAG_ONDISK)
out << " ondisk";
if (ack_type & CEPH_OSD_FLAG_ONNVRAM)
out << " onnvram";
if (ack_type & CEPH_OSD_FLAG_ACK)
out << " ack";
out << ", result = " << result;
out << ")";
}
};
#endif

8
src/msg/Message.cc
View File

@ -65,8 +65,6 @@ using namespace std;
#include "messages/MOSDPing.h"
#include "messages/MOSDOp.h"
#include "messages/MOSDOpReply.h"
#include "messages/MOSDSubOp.h"
#include "messages/MOSDSubOpReply.h"
#include "messages/MOSDRepOp.h"
#include "messages/MOSDRepOpReply.h"
#include "messages/MOSDMap.h"
@ -463,12 +461,6 @@ Message *decode_message(CephContext *cct, int crcflags,
case CEPH_MSG_OSD_OPREPLY:
m = new MOSDOpReply();
break;
case MSG_OSD_SUBOP:
m = new MOSDSubOp();
break;
case MSG_OSD_SUBOPREPLY:
m = new MOSDSubOpReply();
break;
case MSG_OSD_REPOP:
m = new MOSDRepOp();
break;

5
src/msg/Message.h
View File

@ -68,8 +68,9 @@
#define MSG_OSD_MARK_ME_DOWN 74
#define MSG_OSD_FULL 75
#define MSG_OSD_SUBOP 76
#define MSG_OSD_SUBOPREPLY 77
// removed right after luminous
//#define MSG_OSD_SUBOP 76
//#define MSG_OSD_SUBOPREPLY 77
#define MSG_OSD_PGTEMP 78

2
src/osd/OSD.h
View File

@ -2292,9 +2292,7 @@ private:
switch (m->get_type()) {
case CEPH_MSG_OSD_OP:
case CEPH_MSG_OSD_BACKOFF:
case MSG_OSD_SUBOP:
case MSG_OSD_REPOP:
case MSG_OSD_SUBOPREPLY:
case MSG_OSD_REPOPREPLY:
case MSG_OSD_PG_PUSH:
case MSG_OSD_PG_PULL:

6
src/osd/OpRequest.cc
View File

@ -8,8 +8,6 @@
#include "common/config.h"
#include "msg/Message.h"
#include "messages/MOSDOp.h"
#include "messages/MOSDSubOp.h"
#include "messages/MOSDSubOpReply.h"
#include "messages/MOSDRepOp.h"
#include "messages/MOSDRepOpReply.h"
#include "include/assert.h"
@ -36,10 +34,6 @@ OpRequest::OpRequest(Message *req, OpTracker *tracker) :
}
if (req->get_type() == CEPH_MSG_OSD_OP) {
reqid = static_cast<MOSDOp*>(req)->get_reqid();
} else if (req->get_type() == MSG_OSD_SUBOP) {
reqid = static_cast<MOSDSubOp*>(req)->reqid;
} else if (req->get_type() == MSG_OSD_SUBOPREPLY) {
reqid = static_cast<MOSDSubOpReply*>(req)->reqid;
} else if (req->get_type() == MSG_OSD_REPOP) {
reqid = static_cast<MOSDRepOp*>(req)->reqid;
} else if (req->get_type() == MSG_OSD_REPOPREPLY) {

158
src/osd/PG.cc
View File

@ -50,9 +50,7 @@
#include "messages/MOSDPGUpdateLogMissingReply.h"
#include "messages/MOSDBackoff.h"
#include "messages/MOSDScrubReserve.h"
#include "messages/MOSDSubOp.h"
#include "messages/MOSDRepOp.h"
#include "messages/MOSDSubOpReply.h"
#include "messages/MOSDRepOpReply.h"
#include "messages/MOSDRepScrubMap.h"
#include "messages/MOSDPGRecoveryDelete.h"
@ -3725,46 +3723,6 @@ void PG::do_replica_scrub_map(OpRequestRef op)
}
}
void PG::sub_op_scrub_map(OpRequestRef op)
{
// for legacy jewel compatibility only
const MOSDSubOp *m = static_cast<const MOSDSubOp *>(op->get_req());
assert(m->get_type() == MSG_OSD_SUBOP);
dout(7) << "sub_op_scrub_map" << dendl;
if (m->map_epoch < info.history.same_interval_since) {
dout(10) << "sub_op_scrub discarding old sub_op from "
<< m->map_epoch << " < " << info.history.same_interval_since << dendl;
return;
}
if (!scrubber.is_chunky_scrub_active()) {
dout(10) << "sub_op_scrub_map scrub isn't active" << dendl;
return;
}
op->mark_started();
dout(10) << " got " << m->from << " scrub map" << dendl;
bufferlist::iterator p = const_cast<bufferlist&>(m->get_data()).begin();
scrubber.received_maps[m->from].decode(p, info.pgid.pool());
dout(10) << "map version is "
<< scrubber.received_maps[m->from].valid_through
<< dendl;
--scrubber.waiting_on;
scrubber.waiting_on_whom.erase(m->from);
if (scrubber.waiting_on == 0) {
if (ops_blocked_by_scrub()) {
requeue_scrub(true);
} else {
requeue_scrub(false);
}
}
}
// send scrub v3 messages (chunky scrub)
void PG::_request_scrub_map(
pg_shard_t replica, eversion_t version,
@ -3801,24 +3759,14 @@ void PG::handle_scrub_reserve_request(OpRequestRef op)
dout(20) << __func__ << ": failed to reserve remotely" << dendl;
scrubber.reserved = false;
}
if (op->get_req()->get_type() == MSG_OSD_SCRUB_RESERVE) {
const MOSDScrubReserve *m =
static_cast<const MOSDScrubReserve*>(op->get_req());
Message *reply = new MOSDScrubReserve(
spg_t(info.pgid.pgid, primary.shard),
m->map_epoch,
scrubber.reserved ? MOSDScrubReserve::GRANT : MOSDScrubReserve::REJECT,
pg_whoami);
osd->send_message_osd_cluster(reply, op->get_req()->get_connection());
} else {
// for jewel compat only
const MOSDSubOp *req = static_cast<const MOSDSubOp*>(op->get_req());
assert(req->get_type() == MSG_OSD_SUBOP);
MOSDSubOpReply *reply = new MOSDSubOpReply(
req, pg_whoami, 0, get_osdmap()->get_epoch(), CEPH_OSD_FLAG_ACK);
::encode(scrubber.reserved, reply->get_data());
osd->send_message_osd_cluster(reply, op->get_req()->get_connection());
}
const MOSDScrubReserve *m =
static_cast<const MOSDScrubReserve*>(op->get_req());
Message *reply = new MOSDScrubReserve(
spg_t(info.pgid.pgid, primary.shard),
m->map_epoch,
scrubber.reserved ? MOSDScrubReserve::GRANT : MOSDScrubReserve::REJECT,
pg_whoami);
osd->send_message_osd_cluster(reply, op->get_req()->get_connection());
}
void PG::handle_scrub_reserve_grant(OpRequestRef op, pg_shard_t from)
@ -3913,27 +3861,12 @@ void PG::scrub_reserve_replicas()
++i) {
if (*i == pg_whoami) continue;
dout(10) << "scrub requesting reserve from osd." << *i << dendl;
if (HAVE_FEATURE(get_min_acting_features(), SERVER_LUMINOUS)) {
osd->send_message_osd_cluster(
i->osd,
new MOSDScrubReserve(spg_t(info.pgid.pgid, i->shard),
get_osdmap()->get_epoch(),
MOSDScrubReserve::REQUEST, pg_whoami),
get_osdmap()->get_epoch());
} else {
// for jewel compat only
vector<OSDOp> scrub(1);
scrub[0].op.op = CEPH_OSD_OP_SCRUB_RESERVE;
hobject_t poid;
eversion_t v;
osd_reqid_t reqid;
MOSDSubOp *subop = new MOSDSubOp(
reqid, pg_whoami, spg_t(info.pgid.pgid, i->shard), poid, 0,
get_osdmap()->get_epoch(), osd->get_tid(), v);
subop->ops = scrub;
osd->send_message_osd_cluster(
i->osd, subop, get_osdmap()->get_epoch());
}
osd->send_message_osd_cluster(
i->osd,
new MOSDScrubReserve(spg_t(info.pgid.pgid, i->shard),
get_osdmap()->get_epoch(),
MOSDScrubReserve::REQUEST, pg_whoami),
get_osdmap()->get_epoch());
}
}
@ -3945,26 +3878,12 @@ void PG::scrub_unreserve_replicas()
++i) {
if (*i == pg_whoami) continue;
dout(10) << "scrub requesting unreserve from osd." << *i << dendl;
if (HAVE_FEATURE(get_min_acting_features(), SERVER_LUMINOUS)) {
osd->send_message_osd_cluster(
i->osd,
new MOSDScrubReserve(spg_t(info.pgid.pgid, i->shard),
get_osdmap()->get_epoch(),
MOSDScrubReserve::RELEASE, pg_whoami),
get_osdmap()->get_epoch());
} else {
// for jewel compat only
vector<OSDOp> scrub(1);
scrub[0].op.op = CEPH_OSD_OP_SCRUB_UNRESERVE;
hobject_t poid;
eversion_t v;
osd_reqid_t reqid;
MOSDSubOp *subop = new MOSDSubOp(
reqid, pg_whoami, spg_t(info.pgid.pgid, i->shard), poid, 0,
get_osdmap()->get_epoch(), osd->get_tid(), v);
subop->ops = scrub;
osd->send_message_osd_cluster(i->osd, subop, get_osdmap()->get_epoch());
}
osd->send_message_osd_cluster(
i->osd,
new MOSDScrubReserve(spg_t(info.pgid.pgid, i->shard),
get_osdmap()->get_epoch(),
MOSDScrubReserve::RELEASE, pg_whoami),
get_osdmap()->get_epoch());
}
}
@ -4289,33 +4208,12 @@ void PG::replica_scrub(
map, start, end, msg->deep, msg->seed,
handle);
if (HAVE_FEATURE(acting_features, SERVER_LUMINOUS)) {
MOSDRepScrubMap *reply = new MOSDRepScrubMap(
spg_t(info.pgid.pgid, get_primary().shard),
msg->map_epoch,
pg_whoami);
::encode(map, reply->get_data());
osd->send_message_osd_cluster(reply, msg->get_connection());
} else {
// for jewel compatibility
vector<OSDOp> scrub(1);
scrub[0].op.op = CEPH_OSD_OP_SCRUB_MAP;
hobject_t poid;
eversion_t v;
osd_reqid_t reqid;
MOSDSubOp *subop = new MOSDSubOp(
reqid,
pg_whoami,
spg_t(info.pgid.pgid, get_primary().shard),
poid,
0,
msg->map_epoch,
osd->get_tid(),
v);
::encode(map, subop->get_data());
subop->ops = scrub;
osd->send_message_osd_cluster(subop, msg->get_connection());
}
MOSDRepScrubMap *reply = new MOSDRepScrubMap(
spg_t(info.pgid.pgid, get_primary().shard),
msg->map_epoch,
pg_whoami);
::encode(map, reply->get_data());
osd->send_message_osd_cluster(reply, msg->get_connection());
}
/* Scrub:
@ -5749,8 +5647,6 @@ bool PG::can_discard_request(OpRequestRef& op)
return can_discard_op(op);
case CEPH_MSG_OSD_BACKOFF:
return false; // never discard
case MSG_OSD_SUBOP:
return can_discard_replica_op<MOSDSubOp, MSG_OSD_SUBOP>(op);
case MSG_OSD_REPOP:
return can_discard_replica_op<MOSDRepOp, MSG_OSD_REPOP>(op);
case MSG_OSD_PG_PUSH:
@ -5759,8 +5655,6 @@ bool PG::can_discard_request(OpRequestRef& op)
return can_discard_replica_op<MOSDPGPull, MSG_OSD_PG_PULL>(op);
case MSG_OSD_PG_PUSH_REPLY:
return can_discard_replica_op<MOSDPGPushReply, MSG_OSD_PG_PUSH_REPLY>(op);
case MSG_OSD_SUBOPREPLY:
return can_discard_replica_op<MOSDSubOpReply, MSG_OSD_SUBOPREPLY>(op);
case MSG_OSD_REPOPREPLY:
return can_discard_replica_op<MOSDRepOpReply, MSG_OSD_REPOPREPLY>(op);
case MSG_OSD_PG_RECOVERY_DELETE:

3
src/osd/PG.h
View File

@ -1401,7 +1401,6 @@ public:
OpRequestRef op,
ThreadPool::TPHandle &handle);
void do_replica_scrub_map(OpRequestRef op);
void sub_op_scrub_map(OpRequestRef op);
void handle_scrub_reserve_request(OpRequestRef op);
void handle_scrub_reserve_grant(OpRequestRef op, pg_shard_t from);
@ -2505,8 +2504,6 @@ public:
) = 0;
virtual void do_op(OpRequestRef& op) = 0;
virtual void do_sub_op(OpRequestRef op) = 0;
virtual void do_sub_op_reply(OpRequestRef op) = 0;
virtual void do_scan(
OpRequestRef op,
ThreadPool::TPHandle &handle

159
src/osd/PrimaryLogPG.cc
View File

@ -31,8 +31,6 @@
#include "messages/MOSDOp.h"
#include "messages/MOSDBackoff.h"
#include "messages/MOSDSubOp.h"
#include "messages/MOSDSubOpReply.h"
#include "messages/MOSDPGTrim.h"
#include "messages/MOSDPGScan.h"
#include "messages/MOSDRepScrub.h"
@ -1752,14 +1750,6 @@ void PrimaryLogPG::do_request(
}
break;
case MSG_OSD_SUBOP:
do_sub_op(op);
break;
case MSG_OSD_SUBOPREPLY:
do_sub_op_reply(op);
break;
case MSG_OSD_PG_SCAN:
do_scan(op, handle);
break;
@ -3455,66 +3445,6 @@ void PrimaryLogPG::log_op_stats(OpContext *ctx)
<< " lat " << latency << dendl;
}
void PrimaryLogPG::do_sub_op(OpRequestRef op)
{
const MOSDSubOp *m = static_cast<const MOSDSubOp*>(op->get_req());
assert(have_same_or_newer_map(m->map_epoch));
assert(m->get_type() == MSG_OSD_SUBOP);
dout(15) << "do_sub_op " << *op->get_req() << dendl;
if (!is_peered()) {
waiting_for_peered.push_back(op);
op->mark_delayed("waiting for active");
return;
}
const OSDOp *first = NULL;
if (m->ops.size() >= 1) {
first = &m->ops[0];
}
if (first) {
switch (first->op.op) {
case CEPH_OSD_OP_DELETE:
sub_op_remove(op);
return;
case CEPH_OSD_OP_SCRUB_RESERVE:
handle_scrub_reserve_request(op);
return;
case CEPH_OSD_OP_SCRUB_UNRESERVE:
handle_scrub_reserve_release(op);
return;
case CEPH_OSD_OP_SCRUB_MAP:
sub_op_scrub_map(op);
return;
}
}
}
void PrimaryLogPG::do_sub_op_reply(OpRequestRef op)
{
const MOSDSubOpReply *r = static_cast<const MOSDSubOpReply *>(op->get_req());
assert(r->get_type() == MSG_OSD_SUBOPREPLY);
if (r->ops.size() >= 1) {
const OSDOp& first = r->ops[0];
switch (first.op.op) {
case CEPH_OSD_OP_SCRUB_RESERVE:
{
pg_shard_t from = r->from;
bufferlist::iterator p = const_cast<bufferlist&>(r->get_data()).begin();
bool reserved;
::decode(reserved, p);
if (reserved) {
handle_scrub_reserve_grant(op, from);
} else {
handle_scrub_reserve_reject(op, from);
}
}
return;
}
}
}
void PrimaryLogPG::do_scan(
OpRequestRef op,
ThreadPool::TPHandle &handle)
@ -10380,25 +10310,6 @@ int PrimaryLogPG::recover_missing(
return PULL_YES;
}
void PrimaryLogPG::send_remove_op(
const hobject_t& oid, eversion_t v, pg_shard_t peer)
{
ceph_tid_t tid = osd->get_tid();
osd_reqid_t rid(osd->get_cluster_msgr_name(), 0, tid);
dout(10) << "send_remove_op " << oid << " from osd." << peer
<< " tid " << tid << dendl;
MOSDSubOp *subop = new MOSDSubOp(
rid, pg_whoami, spg_t(info.pgid.pgid, peer.shard),
oid, CEPH_OSD_FLAG_ACK,
get_osdmap()->get_epoch(), tid, v);
subop->ops = vector<OSDOp>(1);
subop->ops[0].op.op = CEPH_OSD_OP_DELETE;
osd->send_message_osd_cluster(peer.osd, subop, get_osdmap()->get_epoch());
}
void PrimaryLogPG::remove_missing_object(const hobject_t &soid,
eversion_t v, Context *on_complete)
{
@ -10566,20 +10477,6 @@ void PrimaryLogPG::failed_push(const list<pg_shard_t> &from, const hobject_t &so
finish_recovery_op(soid); // close out this attempt,
}
void PrimaryLogPG::sub_op_remove(OpRequestRef op)
{
const MOSDSubOp *m = static_cast<const MOSDSubOp*>(op->get_req());
assert(m->get_type() == MSG_OSD_SUBOP);
dout(7) << "sub_op_remove " << m->poid << dendl;
op->mark_started();
ObjectStore::Transaction t;
remove_snap_mapped_object(t, m->poid);
int r = osd->store->queue_transaction(osr.get(), std::move(t), NULL);
assert(r == 0);
}
eversion_t PrimaryLogPG::pick_newest_available(const hobject_t& oid)
{
eversion_t v;
@ -12136,43 +12033,29 @@ uint64_t PrimaryLogPG::recover_backfill(
add_object_context_to_pg_stat(obc, &stat);
pending_backfill_updates[*i] = stat;
}
if (HAVE_FEATURE(get_min_upacting_features(), SERVER_LUMINOUS)) {
map<pg_shard_t,MOSDPGBackfillRemove*> reqs;
for (unsigned i = 0; i < to_remove.size(); ++i) {
handle.reset_tp_timeout();
const hobject_t& oid = to_remove[i].get<0>();
eversion_t v = to_remove[i].get<1>();
pg_shard_t peer = to_remove[i].get<2>();
MOSDPGBackfillRemove *m;
auto it = reqs.find(peer);
if (it != reqs.end()) {
m = it->second;
} else {
m = reqs[peer] = new MOSDPGBackfillRemove(
spg_t(info.pgid.pgid, peer.shard),
get_osdmap()->get_epoch());
}
m->ls.push_back(make_pair(oid, v));
if (oid <= last_backfill_started)
pending_backfill_updates[oid]; // add empty stat!
map<pg_shard_t,MOSDPGBackfillRemove*> reqs;
for (unsigned i = 0; i < to_remove.size(); ++i) {
handle.reset_tp_timeout();
const hobject_t& oid = to_remove[i].get<0>();
eversion_t v = to_remove[i].get<1>();
pg_shard_t peer = to_remove[i].get<2>();
MOSDPGBackfillRemove *m;
auto it = reqs.find(peer);
if (it != reqs.end()) {
m = it->second;
} else {
m = reqs[peer] = new MOSDPGBackfillRemove(
spg_t(info.pgid.pgid, peer.shard),
get_osdmap()->get_epoch());
}
for (auto p : reqs) {
osd->send_message_osd_cluster(p.first.osd, p.second,
get_osdmap()->get_epoch());
}
} else {
// for jewel targets
for (unsigned i = 0; i < to_remove.size(); ++i) {
handle.reset_tp_timeout();
m->ls.push_back(make_pair(oid, v));
// ordered before any subsequent updates
send_remove_op(to_remove[i].get<0>(), to_remove[i].get<1>(),
to_remove[i].get<2>());
if (to_remove[i].get<0>() <= last_backfill_started)
pending_backfill_updates[to_remove[i].get<0>()]; // add empty stat!
}
if (oid <= last_backfill_started)
pending_backfill_updates[oid]; // add empty stat!
}
for (auto p : reqs) {
osd->send_message_osd_cluster(p.first.osd, p.second,
get_osdmap()->get_epoch());
}
pgbackend->run_recovery_op(h, get_recovery_op_priority());

3
src/osd/PrimaryLogPG.h
View File

@ -1238,7 +1238,6 @@ protected:
class C_OSD_AppliedRecoveredObject;
class C_OSD_CommittedPushedObject;
class C_OSD_AppliedRecoveredObjectReplica;
void sub_op_remove(OpRequestRef op);
void _applied_recovered_object(ObjectContextRef obc);
void _applied_recovered_object_replica();
@ -1386,8 +1385,6 @@ public:
void record_write_error(OpRequestRef op, const hobject_t &soid,
MOSDOpReply *orig_reply, int r);
void do_pg_op(OpRequestRef op);
void do_sub_op(OpRequestRef op) override;
void do_sub_op_reply(OpRequestRef op) override;
void do_scan(
OpRequestRef op,
ThreadPool::TPHandle &handle) override;

10
src/osd/ReplicatedBackend.cc
View File

@ -14,9 +14,7 @@
#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"
@ -210,14 +208,6 @@ bool ReplicatedBackend::_handle_message(
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;

2
src/osd/mClockClientQueue.cc
View File

@ -109,10 +109,12 @@ namespace ceph {
if (osd_op_type_t::client_op != type) {
return type;
/* fixme: this should match REPOP and probably others
} else if (MSG_OSD_SUBOP ==
boost::get<OpRequestRef>(
request.second.get_variant())->get_req()->get_header().type) {
return osd_op_type_t::osd_subop;
*/
} else {
return osd_op_type_t::client_op;
}

2
src/osd/mClockOpClassQueue.cc
View File

@ -106,10 +106,12 @@ namespace ceph {
if (osd_op_type_t::client_op != type) {
return type;
/* fixme: this should match REPOP and probably others
} else if (MSG_OSD_SUBOP ==
boost::get<OpRequestRef>(
request.second.get_variant())->get_req()->get_header().type) {
return osd_op_type_t::osd_subop;
*/
} else {
return osd_op_type_t::client_op;
}

4
src/test/encoding/types.h
View File

@ -612,10 +612,6 @@ MESSAGE(MOSDRepScrub)
MESSAGE(MOSDScrub)
#include "messages/MOSDForceRecovery.h"
MESSAGE(MOSDForceRecovery)
#include "messages/MOSDSubOp.h"
MESSAGE(MOSDSubOp)
#include "messages/MOSDSubOpReply.h"
MESSAGE(MOSDSubOpReply)
#include "messages/MPGStats.h"
MESSAGE(MPGStats)
#include "messages/MPGStatsAck.h"