mirror of
https://github.com/ceph/ceph
synced 2026-08-02 07:03:18 +00:00
Merge pull request #65411 from aainscow/wip-72561-tentacle
tentacle: Optimized Erasure Coding - Fixpack 2
This commit is contained in:
Laura Flores
GitHub
commit
f12c135e0b
@ -4142,9 +4142,11 @@ bool OSDMonitor::preprocess_pgtemp(MonOpRequestRef op)
|
||||
// change?
|
||||
// NOTE: we assume that this will clear pg_primary, so consider
|
||||
// an existing pg_primary field to imply a change
|
||||
std::vector<int> acting_set;
|
||||
osdmap.pg_to_acting_osds(p->first, acting_set);
|
||||
if (p->second.size() &&
|
||||
(osdmap.pg_temp->count(p->first) == 0 ||
|
||||
osdmap.pg_temp->get(p->first) != p->second ||
|
||||
acting_set != p->second ||
|
||||
osdmap.primary_temp->count(p->first)))
|
||||
return false;
|
||||
}
|
||||
|
||||
@ -308,18 +308,66 @@ void ECBackend::RecoveryBackend::handle_recovery_push_reply(
|
||||
continue_recovery_op(rop, m);
|
||||
}
|
||||
|
||||
void ECBackend::RecoveryBackend::update_object_size_after_read(
|
||||
uint64_t size,
|
||||
read_result_t &res,
|
||||
read_request_t &req) {
|
||||
// We didn't know the size before, meaning the zero for decode calculations
|
||||
// will be off. Recalculate them!
|
||||
ECUtil::shard_extent_set_t zero_mask(sinfo.get_k_plus_m());
|
||||
sinfo.ro_size_to_zero_mask(size, zero_mask);
|
||||
ECUtil::shard_extent_set_t read_mask(sinfo.get_k_plus_m());
|
||||
sinfo.ro_size_to_read_mask(size, read_mask);
|
||||
extent_set superset = res.buffers_read.get_extent_superset();
|
||||
|
||||
for (auto &&[shard, eset] : zero_mask) {
|
||||
eset.intersection_of(superset);
|
||||
if (!eset.empty() &&
|
||||
(res.zero_length_reads.contains(shard) ||
|
||||
res.buffers_read.contains(shard))) {
|
||||
req.zeros_for_decode[shard].insert(eset);
|
||||
}
|
||||
}
|
||||
|
||||
/* Correct the shard_want_to_read, to make sure everything is within scope
|
||||
* of the newly found object size.
|
||||
*/
|
||||
for (auto iter = req.shard_want_to_read.begin(); iter != req.shard_want_to_read.end();) {
|
||||
auto &&[shard, eset] = *iter;
|
||||
bool erase = false;
|
||||
|
||||
if (read_mask.contains(shard)) {
|
||||
eset.intersection_of(read_mask.get(shard));
|
||||
erase = eset.empty();
|
||||
} else {
|
||||
erase = true;
|
||||
}
|
||||
|
||||
/* Some shards may be empty */
|
||||
if (erase) {
|
||||
iter = req.shard_want_to_read.erase(iter);
|
||||
} else {
|
||||
++iter;
|
||||
}
|
||||
}
|
||||
|
||||
dout(20) << "Update want and zeros from read:size=" << size
|
||||
<< " res=" << res
|
||||
<< " req=" << req
|
||||
<< dendl;
|
||||
}
|
||||
|
||||
void ECBackend::RecoveryBackend::handle_recovery_read_complete(
|
||||
const hobject_t &hoid,
|
||||
ECUtil::shard_extent_map_t &&buffers_read,
|
||||
std::optional<map<string, bufferlist, less<>>> attrs,
|
||||
const ECUtil::shard_extent_set_t &want_to_read,
|
||||
read_result_t &&res,
|
||||
read_request_t &req,
|
||||
RecoveryMessages *m) {
|
||||
dout(10) << __func__ << ": returned " << hoid << " " << buffers_read << dendl;
|
||||
dout(10) << __func__ << ": returned " << hoid << " " << res << dendl;
|
||||
ceph_assert(recovery_ops.contains(hoid));
|
||||
RecoveryBackend::RecoveryOp &op = recovery_ops[hoid];
|
||||
|
||||
if (attrs) {
|
||||
op.xattrs.swap(*attrs);
|
||||
if (res.attrs) {
|
||||
op.xattrs.swap(*(res.attrs));
|
||||
|
||||
if (!op.obc) {
|
||||
// attrs only reference the origin bufferlist (decode from
|
||||
@ -341,53 +389,31 @@ void ECBackend::RecoveryBackend::handle_recovery_read_complete(
|
||||
ceph_assert(op.obc);
|
||||
op.recovery_info.size = op.obc->obs.oi.size;
|
||||
op.recovery_info.oi = op.obc->obs.oi;
|
||||
|
||||
update_object_size_after_read(op.recovery_info.size, res, req);
|
||||
}
|
||||
}
|
||||
ceph_assert(op.xattrs.size());
|
||||
ceph_assert(op.obc);
|
||||
|
||||
op.returned_data.emplace(std::move(buffers_read));
|
||||
|
||||
ECUtil::shard_extent_set_t read_mask(sinfo.get_k_plus_m());
|
||||
sinfo.ro_size_to_read_mask(op.recovery_info.size, read_mask);
|
||||
ECUtil::shard_extent_set_t shard_want_to_read(sinfo.get_k_plus_m());
|
||||
|
||||
for (auto &[shard, eset] : want_to_read) {
|
||||
/* Read buffers do not need recovering! */
|
||||
if (buffers_read.contains(shard)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
/* Read-buffers will be truncated to the end-of-object. Do not attempt
|
||||
* to recover off-the-end.
|
||||
*/
|
||||
shard_want_to_read[shard].intersection_of(read_mask.get(shard),eset);
|
||||
|
||||
/* Some shards may be empty */
|
||||
if (shard_want_to_read[shard].empty()) {
|
||||
shard_want_to_read.erase(shard);
|
||||
}
|
||||
}
|
||||
|
||||
op.returned_data.emplace(std::move(res.buffers_read));
|
||||
uint64_t aligned_size = ECUtil::align_next(op.obc->obs.oi.size);
|
||||
|
||||
int r = op.returned_data->decode(ec_impl, shard_want_to_read, aligned_size, get_parent()->get_dpp(), true);
|
||||
dout(20) << __func__ << " before decode: oid=" << op.hoid << " EC_DEBUG_BUFFERS: "
|
||||
<< op.returned_data->debug_string(2048, 0)
|
||||
<< dendl;
|
||||
|
||||
op.returned_data->add_zero_padding_for_decode(req.zeros_for_decode);
|
||||
int r = op.returned_data->decode(ec_impl, req.shard_want_to_read, aligned_size, get_parent()->get_dpp(), true);
|
||||
ceph_assert(r == 0);
|
||||
|
||||
// Finally, we don't want to write any padding, so truncate the buffer
|
||||
// to remove it.
|
||||
op.returned_data->erase_after_ro_offset(aligned_size);
|
||||
|
||||
for (auto &&shard: op.missing_on_shards) {
|
||||
if (read_mask.contains(shard) && op.returned_data->contains_shard(shard)) {
|
||||
ceph_assert(read_mask.at(shard).range_end() >=
|
||||
op.returned_data->get_extent_map(shard).get_end_off());
|
||||
}
|
||||
}
|
||||
|
||||
dout(20) << __func__ << ": oid=" << op.hoid << dendl;
|
||||
dout(30) << __func__ << "EC_DEBUG_BUFFERS: "
|
||||
<< op.returned_data->debug_string(2048, 8)
|
||||
dout(20) << __func__ << " after decode: oid=" << op.hoid << " EC_DEBUG_BUFFERS: "
|
||||
<< op.returned_data->debug_string(2048, 0)
|
||||
<< dendl;
|
||||
|
||||
continue_recovery_op(op, m);
|
||||
@ -440,9 +466,8 @@ struct RecoveryReadCompleter : ECCommon::ReadCompleter {
|
||||
ceph_assert(req.to_read.size() == 0);
|
||||
backend.handle_recovery_read_complete(
|
||||
hoid,
|
||||
std::move(res.buffers_read),
|
||||
res.attrs,
|
||||
req.shard_want_to_read,
|
||||
std::move(res),
|
||||
req,
|
||||
&rm);
|
||||
}
|
||||
|
||||
@ -503,6 +528,14 @@ void ECBackend::RecoveryBackend::dispatch_recovery_messages(
|
||||
|
||||
#if 1
|
||||
if (!replies.empty()) {
|
||||
dout(20) << __func__ << " recovery_transactions=";
|
||||
Formatter *f = Formatter::create("json");
|
||||
f->open_object_section("t");
|
||||
m.t.dump(f);
|
||||
f->close_section();
|
||||
f->flush(*_dout);
|
||||
delete f;
|
||||
*_dout << dendl;
|
||||
commit_txn_send_replies(std::move(m.t), std::move(replies));
|
||||
}
|
||||
#endif
|
||||
@ -1005,6 +1038,18 @@ void ECBackend::handle_sub_write(
|
||||
tls.reserve(2);
|
||||
tls.push_back(std::move(op.t));
|
||||
tls.push_back(std::move(localt));
|
||||
dout(20) << __func__ << " queue_transactions=";
|
||||
Formatter *f = Formatter::create("json");
|
||||
f->open_array_section("tls");
|
||||
for (ObjectStore::Transaction t: tls) {
|
||||
f->open_object_section("t");
|
||||
t.dump(f);
|
||||
f->close_section();
|
||||
}
|
||||
f->close_section();
|
||||
f->flush(*_dout);
|
||||
delete f;
|
||||
*_dout << dendl;
|
||||
get_parent()->queue_transactions(tls, msg);
|
||||
dout(30) << __func__ << " missing after" << get_parent()->get_log().
|
||||
get_missing().
|
||||
@ -1192,6 +1237,11 @@ void ECBackend::handle_sub_read_reply(
|
||||
buffers_read.insert_in_shard(from.shard, offset, buffer_list);
|
||||
}
|
||||
rop.debug_log.emplace_back(ECUtil::READ_DONE, op.from, buffers_read);
|
||||
|
||||
// zero length reads may need to be zero padded during recovery
|
||||
if (!buffers_read.contains_shard(from.shard)) {
|
||||
rop.complete.at(hoid).zero_length_reads.insert(from.shard);
|
||||
}
|
||||
}
|
||||
for (auto &&[hoid, req]: rop.to_read) {
|
||||
if (!rop.complete.contains(hoid)) {
|
||||
@ -1230,6 +1280,13 @@ void ECBackend::handle_sub_read_reply(
|
||||
rop.debug_log.emplace_back(ECUtil::ERROR, op.from, complete.buffers_read);
|
||||
complete.buffers_read.erase_shard(from.shard);
|
||||
complete.processed_read_requests.erase(from.shard);
|
||||
// If we are doing redundant reads, then we must take care that any failed
|
||||
// reads are not replaced with a zero buffer. When fast_reads are disabled,
|
||||
// the send_all_remaining_reads() call will replace the zeros_for_decode
|
||||
// based on the recovery read.
|
||||
if (rop.do_redundant_reads) {
|
||||
rop.to_read.at(hoid).zeros_for_decode.erase(from.shard);
|
||||
}
|
||||
dout(20) << __func__ << " shard=" << from << " error=" << err << dendl;
|
||||
}
|
||||
|
||||
@ -1243,9 +1300,10 @@ void ECBackend::handle_sub_read_reply(
|
||||
rop.in_progress.erase(from);
|
||||
unsigned is_complete = 0;
|
||||
bool need_resend = false;
|
||||
bool all_sub_reads_done = rop.in_progress.empty();
|
||||
// For redundant reads check for completion as each shard comes in,
|
||||
// or in a non-recovery read check for completion once all the shards read.
|
||||
if (rop.do_redundant_reads || rop.in_progress.empty()) {
|
||||
if (rop.do_redundant_reads || all_sub_reads_done) {
|
||||
for (auto &&[oid, read_result]: rop.complete) {
|
||||
shard_id_set have;
|
||||
read_result.processed_read_requests.populate_shard_id_set(have);
|
||||
@ -1255,29 +1313,34 @@ void ECBackend::handle_sub_read_reply(
|
||||
populate_shard_id_set(want_to_read);
|
||||
|
||||
dout(20) << __func__ << " read_result: " << read_result << dendl;
|
||||
// If all reads are done, we can safely assume that zero buffers can
|
||||
// be applied.
|
||||
if (all_sub_reads_done) {
|
||||
rop.to_read.at(oid).zeros_for_decode.populate_shard_id_set(have);
|
||||
}
|
||||
|
||||
int err = -EIO; // If attributes needed but not read.
|
||||
if (!rop.to_read.at(oid).want_attrs || rop.complete.at(oid).attrs) {
|
||||
err = ec_impl->minimum_to_decode(want_to_read, have, dummy_minimum,
|
||||
nullptr);
|
||||
}
|
||||
|
||||
int err = ec_impl->minimum_to_decode(want_to_read, have, dummy_minimum,
|
||||
nullptr);
|
||||
if (err) {
|
||||
dout(20) << __func__ << " minimum_to_decode failed" << dendl;
|
||||
if (rop.in_progress.empty()) {
|
||||
if (all_sub_reads_done) {
|
||||
// If we don't have enough copies, try other pg_shard_ts if available.
|
||||
// During recovery there may be multiple osds with copies of the same shard,
|
||||
// so getting EIO from one may result in multiple passes through this code path.
|
||||
if (!rop.do_redundant_reads) {
|
||||
rop.debug_log.emplace_back(ECUtil::REQUEST_MISSING, op.from);
|
||||
int r = read_pipeline.send_all_remaining_reads(oid, rop);
|
||||
if (r == 0) {
|
||||
// We found that new reads are required to do a decode.
|
||||
need_resend = true;
|
||||
continue;
|
||||
} else if (r > 0) {
|
||||
// No new reads were requested. This means that some parity
|
||||
// shards can be assumed to be zeros.
|
||||
err = 0;
|
||||
}
|
||||
// else insufficient shards are available, keep the errors.
|
||||
|
||||
rop.debug_log.emplace_back(ECUtil::REQUEST_MISSING, op.from);
|
||||
int r = read_pipeline.send_all_remaining_reads(oid, rop);
|
||||
if (r == 0 && !rop.do_redundant_reads) {
|
||||
// We found that new reads are required to do a decode.
|
||||
need_resend = true;
|
||||
continue;
|
||||
}
|
||||
// else insufficient shards are available, keep the errors.
|
||||
|
||||
// Couldn't read any additional shards so handle as completed with errors
|
||||
// We don't want to confuse clients / RBD with objectstore error
|
||||
// values in particular ENOENT. We may have different error returns
|
||||
@ -1321,6 +1384,17 @@ void ECBackend::handle_sub_read_reply(
|
||||
dout(20) << __func__ << " Complete: " << rop << dendl;
|
||||
rop.trace.event("ec read complete");
|
||||
rop.debug_log.emplace_back(ECUtil::COMPLETE, op.from);
|
||||
|
||||
/* If do_redundant_reads is set then there might be some in progress
|
||||
* reads remaining. We need to make sure that these non-read shards
|
||||
* do not get padded. If there was no in progress read, then the zero
|
||||
* padding is allowed to stay.
|
||||
*/
|
||||
for (auto pg_shard : rop.in_progress) {
|
||||
for (auto &&[oid, read] : rop.to_read) {
|
||||
read.zeros_for_decode.erase(pg_shard.shard);
|
||||
}
|
||||
}
|
||||
read_pipeline.complete_read_op(std::move(rop));
|
||||
} else {
|
||||
dout(10) << __func__ << " readop not complete: " << rop << dendl;
|
||||
|
||||
@ -324,12 +324,14 @@ class ECBackend : public ECCommon {
|
||||
void continue_recovery_op(
|
||||
RecoveryBackend::RecoveryOp &op,
|
||||
RecoveryMessages *m);
|
||||
void update_object_size_after_read(
|
||||
uint64_t size,
|
||||
read_result_t &res,
|
||||
read_request_t &req);
|
||||
void handle_recovery_read_complete(
|
||||
const hobject_t &hoid,
|
||||
ECUtil::shard_extent_map_t &&buffers_read,
|
||||
std::optional<std::map<std::string, ceph::buffer::list, std::less<>>>
|
||||
attrs,
|
||||
const ECUtil::shard_extent_set_t &want_to_read,
|
||||
read_result_t &&res,
|
||||
read_request_t &req,
|
||||
RecoveryMessages *m);
|
||||
void handle_recovery_push(
|
||||
const PushOp &op,
|
||||
@ -445,10 +447,6 @@ class ECBackend : public ECCommon {
|
||||
return sinfo.is_nonprimary_shard(shard);
|
||||
}
|
||||
|
||||
bool get_is_hinfo_required() const {
|
||||
return sinfo.get_is_hinfo_required();
|
||||
}
|
||||
|
||||
/**
|
||||
* ECReadPred
|
||||
*
|
||||
|
||||
@ -267,12 +267,20 @@ int ECCommon::ReadPipeline::get_min_avail_to_read_shards(
|
||||
* redundant reads is set, then we want to have the same reads on
|
||||
* every extent. Otherwise, we need to read every shard only if the
|
||||
* necessary shard is missing.
|
||||
* In some (recovery) scenarios, we can "want" a shard, but not "need" to
|
||||
* read it. This typically happens when we do not have a data shard and the
|
||||
* recovery for this will read enough shards to also generate all the parity.
|
||||
*
|
||||
* Since parity shards are often larger than data shards, we must make sure
|
||||
* to read the extra bit!
|
||||
*/
|
||||
if (!have.contains(shard) || do_redundant_reads) {
|
||||
if (!have.contains(shard) || do_redundant_reads ||
|
||||
(want.contains(shard) && !need_set.contains(shard))) {
|
||||
extra_extents.union_of(extent_set);
|
||||
}
|
||||
}
|
||||
|
||||
read_request.zeros_for_decode.clear();
|
||||
for (auto &shard: need_set) {
|
||||
if (!have.contains(shard)) {
|
||||
continue;
|
||||
@ -294,6 +302,15 @@ int ECCommon::ReadPipeline::get_min_avail_to_read_shards(
|
||||
shard_read.extents.intersection_of(extents, read_mask.at(shard));
|
||||
}
|
||||
|
||||
if (zero_mask.contains(shard)) {
|
||||
extents.intersection_of(zero_mask.at(shard));
|
||||
|
||||
/* Any remaining extents can be assumed ot be zeros... so record these. */
|
||||
if (!extents.empty()) {
|
||||
read_request.zeros_for_decode.emplace(shard, std::move(extents));
|
||||
}
|
||||
}
|
||||
|
||||
if (!shard_read.extents.empty()) {
|
||||
read_request.shard_reads[shard_id] = std::move(shard_read);
|
||||
}
|
||||
@ -322,17 +339,18 @@ int ECCommon::ReadPipeline::get_remaining_shards(
|
||||
read_result_t &read_result,
|
||||
read_request_t &read_request,
|
||||
const bool for_recovery,
|
||||
const bool fast_read) {
|
||||
shard_id_map<pg_shard_t> shards(sinfo.get_k_plus_m());
|
||||
bool want_attrs) {
|
||||
set<pg_shard_t> error_shards;
|
||||
for (auto &shard: std::views::keys(read_result.errors)) {
|
||||
error_shards.insert(shard);
|
||||
}
|
||||
|
||||
/* fast-reads should already have scheduled reads to everything, so
|
||||
* this function is irrelevant. */
|
||||
const int r = get_min_avail_to_read_shards(
|
||||
hoid,
|
||||
for_recovery,
|
||||
fast_read,
|
||||
false,
|
||||
read_request,
|
||||
error_shards);
|
||||
|
||||
@ -342,6 +360,9 @@ int ECCommon::ReadPipeline::get_remaining_shards(
|
||||
return -EIO;
|
||||
}
|
||||
|
||||
bool need_attr_request = want_attrs;
|
||||
read_request.want_attrs = want_attrs;
|
||||
|
||||
// Rather than repeating whole read, we can remove everything we already have.
|
||||
for (auto iter = read_request.shard_reads.begin();
|
||||
iter != read_request.shard_reads.end();) {
|
||||
@ -358,11 +379,37 @@ int ECCommon::ReadPipeline::get_remaining_shards(
|
||||
if (do_erase) {
|
||||
iter = read_request.shard_reads.erase(iter);
|
||||
} else {
|
||||
if (need_attr_request && !sinfo.is_nonprimary_shard(shard_id)) {
|
||||
// We have a suitable candidate for attr requests.
|
||||
need_attr_request = false;
|
||||
}
|
||||
++iter;
|
||||
}
|
||||
}
|
||||
|
||||
return read_request.shard_reads.empty()?1:0;
|
||||
if (need_attr_request) {
|
||||
// This happens if we got an error from the shard where we were requesting
|
||||
// the attributes from and the recovery does not require any non primary
|
||||
// shards. The example seen in test was a 2+1 EC being recovered. shards 0
|
||||
// and 2 were being requested and read as part of recovery. Shard was reading
|
||||
// the attributes and failed. The recovery required shard 1, but that does
|
||||
// not have valid attributes on it, so the attribute read failed.
|
||||
// This is a pretty obscure case, so no need to optimise that much. Do an
|
||||
// empty read!
|
||||
shard_id_set have;
|
||||
shard_id_map<pg_shard_t> pg_shards(sinfo.get_k_plus_m());
|
||||
get_all_avail_shards(hoid, have, pg_shards, for_recovery, error_shards);
|
||||
for (auto shard : have) {
|
||||
if (!sinfo.is_nonprimary_shard(shard)) {
|
||||
shard_read_t shard_read;
|
||||
shard_read.pg_shard = pg_shards[shard];
|
||||
read_request.shard_reads.insert(shard, shard_read_t());
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
void ECCommon::ReadPipeline::start_read_op(
|
||||
@ -395,6 +442,7 @@ void ECCommon::ReadPipeline::start_read_op(
|
||||
void ECCommon::ReadPipeline::do_read_op(ReadOp &rop) {
|
||||
const int priority = rop.priority;
|
||||
const ceph_tid_t tid = rop.tid;
|
||||
bool reads_sent = false;
|
||||
|
||||
dout(10) << __func__ << ": starting read " << rop << dendl;
|
||||
ceph_assert(!rop.to_read.empty());
|
||||
@ -402,7 +450,6 @@ void ECCommon::ReadPipeline::do_read_op(ReadOp &rop) {
|
||||
map<pg_shard_t, ECSubRead> messages;
|
||||
for (auto &&[hoid, read_request]: rop.to_read) {
|
||||
bool need_attrs = read_request.want_attrs;
|
||||
ceph_assert(!read_request.shard_reads.empty());
|
||||
|
||||
for (auto &&[shard, shard_read]: read_request.shard_reads) {
|
||||
if (need_attrs && !sinfo.is_nonprimary_shard(shard)) {
|
||||
@ -425,9 +472,11 @@ void ECCommon::ReadPipeline::do_read_op(ReadOp &rop) {
|
||||
for (auto &[start, len]: shard_read.extents) {
|
||||
messages[shard_read.pg_shard].to_read[hoid].emplace_back(
|
||||
boost::make_tuple(start, len, read_request.flags));
|
||||
reads_sent = true;
|
||||
}
|
||||
}
|
||||
ceph_assert(!need_attrs);
|
||||
ceph_assert(reads_sent);
|
||||
}
|
||||
|
||||
std::vector<std::pair<int, Message*>> m;
|
||||
@ -496,17 +545,18 @@ struct ClientReadCompleter final : ECCommon::ReadCompleter {
|
||||
extent_map result;
|
||||
if (res.r == 0) {
|
||||
ceph_assert(res.errors.empty());
|
||||
dout(30) << __func__ << ": before decode: "
|
||||
<< res.buffers_read.debug_string(2048, 8)
|
||||
dout(20) << __func__ << ": before decode: "
|
||||
<< res.buffers_read.debug_string(2048, 0)
|
||||
<< dendl;
|
||||
/* Decode any missing buffers */
|
||||
res.buffers_read.add_zero_padding_for_decode(req.zeros_for_decode);
|
||||
int r = res.buffers_read.decode(read_pipeline.ec_impl,
|
||||
req.shard_want_to_read,
|
||||
req.object_size,
|
||||
read_pipeline.get_parent()->get_dpp());
|
||||
ceph_assert( r == 0 );
|
||||
dout(30) << __func__ << ": after decode: "
|
||||
<< res.buffers_read.debug_string(2048, 8)
|
||||
dout(20) << __func__ << ": after decode: "
|
||||
<< res.buffers_read.debug_string(2048, 0)
|
||||
<< dendl;
|
||||
|
||||
for (auto &&read: req.to_read) {
|
||||
@ -741,12 +791,15 @@ void ECCommon::RMWPipeline::cache_ready(Op &op) {
|
||||
* As such we must never skip a transaction completely. Note that if
|
||||
* should_send is false, then an empty transaction is sent.
|
||||
*/
|
||||
if (should_send && op.skip_transaction(pending_roll_forward, shard, transaction)) {
|
||||
if (!next_write_all_shards && should_send && op.skip_transaction(pending_roll_forward, shard, transaction)) {
|
||||
// Must be an empty transaction
|
||||
ceph_assert(transaction.empty());
|
||||
dout(20) << __func__ << " Skipping transaction for shard " << shard << dendl;
|
||||
continue;
|
||||
}
|
||||
if (!should_send || transaction.empty()) {
|
||||
dout(20) << __func__ << " Sending empty transaction for shard " << shard << dendl;
|
||||
}
|
||||
op.pending_commits++;
|
||||
const pg_stat_t &stats =
|
||||
(should_send || !backfill_shards.contains(pg_shard))
|
||||
@ -795,6 +848,8 @@ void ECCommon::RMWPipeline::cache_ready(Op &op) {
|
||||
}
|
||||
}
|
||||
|
||||
next_write_all_shards = false;
|
||||
|
||||
if (!messages.empty()) {
|
||||
get_parent()->send_message_osd_cluster(messages, get_osdmap_epoch());
|
||||
}
|
||||
@ -858,7 +913,7 @@ void ECCommon::RMWPipeline::finish_rmw(OpRef const &op) {
|
||||
dout(20) << __func__ << " op=" << *op << dendl;
|
||||
|
||||
if (op->on_all_commit) {
|
||||
dout(10) << __func__ << " Calling on_all_commit on " << op << dendl;
|
||||
dout(10) << __func__ << " Calling on_all_commit on " << *op << dendl;
|
||||
op->on_all_commit->complete(0);
|
||||
op->on_all_commit = nullptr;
|
||||
op->trace.event("ec write all committed");
|
||||
@ -906,6 +961,7 @@ void ECCommon::RMWPipeline::on_change() {
|
||||
tid_to_op_map.clear();
|
||||
oid_to_version.clear();
|
||||
waiting_commit.clear();
|
||||
next_write_all_shards = false;
|
||||
}
|
||||
|
||||
void ECCommon::RMWPipeline::on_change2() {
|
||||
@ -913,5 +969,6 @@ void ECCommon::RMWPipeline::on_change2() {
|
||||
}
|
||||
|
||||
void ECCommon::RMWPipeline::call_write_ordered(std::function<void(void)> &&cb) {
|
||||
next_write_all_shards = true;
|
||||
extent_cache.add_on_write(std::move(cb));
|
||||
}
|
||||
@ -95,7 +95,8 @@ struct ECCommon {
|
||||
struct read_request_t {
|
||||
const std::list<ec_align_t> to_read;
|
||||
const uint32_t flags = 0;
|
||||
const ECUtil::shard_extent_set_t shard_want_to_read;
|
||||
ECUtil::shard_extent_set_t shard_want_to_read;
|
||||
ECUtil::shard_extent_set_t zeros_for_decode;
|
||||
shard_id_map<shard_read_t> shard_reads;
|
||||
bool want_attrs = false;
|
||||
uint64_t object_size;
|
||||
@ -107,6 +108,7 @@ struct ECCommon {
|
||||
to_read(to_read),
|
||||
flags(to_read.front().flags),
|
||||
shard_want_to_read(shard_want_to_read),
|
||||
zeros_for_decode(shard_want_to_read.get_max_shards()),
|
||||
shard_reads(shard_want_to_read.get_max_shards()),
|
||||
want_attrs(want_attrs),
|
||||
object_size(object_size) {}
|
||||
@ -114,6 +116,7 @@ struct ECCommon {
|
||||
read_request_t(const ECUtil::shard_extent_set_t &shard_want_to_read,
|
||||
bool want_attrs, uint64_t object_size) :
|
||||
shard_want_to_read(shard_want_to_read),
|
||||
zeros_for_decode(shard_want_to_read.get_max_shards()),
|
||||
shard_reads(shard_want_to_read.get_max_shards()),
|
||||
want_attrs(want_attrs),
|
||||
object_size(object_size) {}
|
||||
@ -124,6 +127,7 @@ struct ECCommon {
|
||||
os << "read_request_t(to_read=[" << to_read << "]"
|
||||
<< ", flags=" << flags
|
||||
<< ", shard_want_to_read=" << shard_want_to_read
|
||||
<< ", zeros_for_decode=" << zeros_for_decode
|
||||
<< ", shard_reads=" << shard_reads
|
||||
<< ", want_attrs=" << want_attrs
|
||||
<< ")";
|
||||
@ -162,6 +166,7 @@ struct ECCommon {
|
||||
std::optional<std::map<std::string, ceph::buffer::list, std::less<>>> attrs;
|
||||
ECUtil::shard_extent_map_t buffers_read;
|
||||
ECUtil::shard_extent_set_t processed_read_requests;
|
||||
shard_id_set zero_length_reads;
|
||||
|
||||
read_result_t(const ECUtil::stripe_info_t *sinfo) :
|
||||
r(0), buffers_read(sinfo),
|
||||
@ -175,7 +180,8 @@ struct ECCommon {
|
||||
os << ", noattrs";
|
||||
}
|
||||
os << ", buffers_read=" << buffers_read;
|
||||
os << ", processed_read_requests=" << processed_read_requests << ")";
|
||||
os << ", processed_read_requests=" << processed_read_requests;
|
||||
os << ", zero_length_reads=" << zero_length_reads << ")";
|
||||
}
|
||||
};
|
||||
|
||||
@ -378,7 +384,7 @@ struct ECCommon {
|
||||
read_result_t &read_result,
|
||||
read_request_t &read_request,
|
||||
bool for_recovery,
|
||||
bool fast_read);
|
||||
bool want_attrs);
|
||||
|
||||
void get_all_avail_shards(
|
||||
const hobject_t &hoid,
|
||||
@ -606,6 +612,7 @@ struct ECCommon {
|
||||
ECCommon &ec_backend;
|
||||
ECExtentCache extent_cache;
|
||||
uint64_t ec_pdw_write_mode;
|
||||
bool next_write_all_shards = false;
|
||||
|
||||
RMWPipeline(CephContext *cct,
|
||||
ceph::ErasureCodeInterfaceRef ec_impl,
|
||||
|
||||
@ -366,10 +366,7 @@ public:
|
||||
return false;
|
||||
}
|
||||
bool get_is_hinfo_required() const final {
|
||||
if (is_optimized()) {
|
||||
return optimized.get_is_hinfo_required();
|
||||
}
|
||||
return true;
|
||||
return !is_optimized();
|
||||
}
|
||||
bool get_is_ec_optimized() const final {
|
||||
return is_optimized();
|
||||
|
||||
@ -37,12 +37,14 @@ using ceph::ErasureCodeInterfaceRef;
|
||||
void debug(const hobject_t &oid, const std::string &str,
|
||||
const ECUtil::shard_extent_map_t &map, DoutPrefixProvider *dpp) {
|
||||
ldpp_dout(dpp, 20)
|
||||
<< " generate_transactions: " << "oid: " << oid << str << map << dendl;
|
||||
ldpp_dout(dpp, 30)
|
||||
<< "EC_DEBUG_BUFFERS: " << map.debug_string(2048, 8) << dendl;
|
||||
<< " generate_transactions: " << "oid: " << oid << " " << str << " " << map << dendl;
|
||||
ldpp_dout(dpp, 20)
|
||||
<< "EC_DEBUG_BUFFERS: " << map.debug_string(2048, 0) << dendl;
|
||||
}
|
||||
|
||||
void ECTransaction::Generate::encode_and_write() {
|
||||
ldpp_dout(dpp, 20)<< __func__ << dendl;
|
||||
|
||||
// For PDW, we already have necessary parity buffers.
|
||||
if (!plan.do_parity_delta_write) {
|
||||
to_write.insert_parity_buffers();
|
||||
@ -253,17 +255,22 @@ ECTransaction::WritePlanObj::WritePlanObj(
|
||||
*/
|
||||
if (op.truncate && op.truncate->first < orig_size) {
|
||||
ECUtil::shard_extent_set_t truncate_read(sinfo.get_k_plus_m());
|
||||
extent_set truncate_write;
|
||||
uint64_t prev_stripe = sinfo.ro_offset_to_prev_stripe_ro_offset(op.truncate->first);
|
||||
uint64_t next_align = ECUtil::align_next(op.truncate->first);
|
||||
sinfo.ro_range_to_shard_extent_set_with_superset(
|
||||
sinfo.ro_range_to_shard_extent_set(
|
||||
prev_stripe, next_align - prev_stripe,
|
||||
truncate_read, truncate_write);
|
||||
truncate_read);
|
||||
|
||||
/* We must always update the entire parity chunk, even if we only read
|
||||
* a small amount of one shard.
|
||||
/* Unless we are doing a full stripe write, we must always read the data
|
||||
* for the partial stripe and update the parity. For the purposes of
|
||||
* parity, the truncated shards are all zero.
|
||||
*/
|
||||
truncate_write.align(sinfo.get_chunk_size());
|
||||
extent_set truncate_write;
|
||||
|
||||
if (next_align != 0) {
|
||||
truncate_write = truncate_read.at(shard_id_t(0));
|
||||
truncate_write.align(EC_ALIGN_SIZE);
|
||||
}
|
||||
|
||||
if (!truncate_read.empty()) {
|
||||
if (to_read) {
|
||||
@ -557,8 +564,23 @@ ECTransaction::Generate::Generate(PGTransaction &t,
|
||||
entry->mod_desc.append(ECUtil::align_next(plan.orig_size));
|
||||
}
|
||||
|
||||
// On a size change, we want to update OI on all shards
|
||||
if (plan.orig_size != plan.projected_size) {
|
||||
// On a size change or when clearing whiteout,
|
||||
// we want to update OI on all shards
|
||||
bool size_change = plan.orig_size != plan.projected_size;
|
||||
bool clear_whiteout = false;
|
||||
|
||||
// If we are updating the OI and we have a cache of the previous OI values
|
||||
if (op.attr_updates.contains(OI_ATTR) && obc && obc->attr_cache.contains(OI_ATTR))
|
||||
{
|
||||
object_info_t oi_cache((obc->attr_cache[OI_ATTR]));
|
||||
if (oi_cache.test_flag(object_info_t::FLAG_WHITEOUT))
|
||||
{
|
||||
object_info_t oi_updates(*(op.attr_updates[OI_ATTR]));
|
||||
clear_whiteout = !oi_updates.test_flag(object_info_t::FLAG_WHITEOUT);
|
||||
}
|
||||
}
|
||||
|
||||
if (size_change || clear_whiteout) {
|
||||
all_shards_written();
|
||||
} else {
|
||||
// All primary shards must always be written, regardless of the write plan.
|
||||
@ -586,7 +608,7 @@ ECTransaction::Generate::Generate(PGTransaction &t,
|
||||
* not simply construct written shards here.
|
||||
*/
|
||||
for (auto &&[shard, t] : transactions) {
|
||||
if (t.get_num_ops() > old_transaction_counts[int(shard)] &&
|
||||
if (std::cmp_greater(t.get_num_ops(), old_transaction_counts[int(shard)]) &&
|
||||
!entry->is_written_shard(shard)) {
|
||||
ldpp_dout(dpp, 20) << __func__ << " Transaction for shard " << shard << ": ";
|
||||
Formatter *f = Formatter::create("json");
|
||||
@ -625,17 +647,29 @@ void ECTransaction::Generate::truncate() {
|
||||
uint64_t clone_start = std::numeric_limits<uint64_t>::max();
|
||||
uint64_t clone_end = 0;
|
||||
|
||||
shard_id_set clone_shards; // intentionally left blank!
|
||||
shard_id_set clone_shards;
|
||||
|
||||
for (auto &&[shard, eset]: truncate_eset) {
|
||||
clone_shards.insert(shard);
|
||||
uint64_t start = eset.range_start();
|
||||
uint64_t start_align_prev = ECUtil::align_prev(start);
|
||||
uint64_t end = eset.range_end();
|
||||
|
||||
if (clone_start > start_align_prev) {
|
||||
clone_start = start_align_prev;
|
||||
}
|
||||
if (clone_end < end) {
|
||||
clone_end = end;
|
||||
}
|
||||
}
|
||||
|
||||
for (auto &&[shard, eset]: truncate_eset) {
|
||||
if (!transactions.contains(shard)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
auto &t = transactions.at(shard);
|
||||
uint64_t start = eset.range_start();
|
||||
uint64_t start_align_prev = ECUtil::align_prev(start);
|
||||
uint64_t start_align_next = ECUtil::align_next(start);
|
||||
uint64_t end = eset.range_end();
|
||||
t.touch(
|
||||
@ -645,9 +679,9 @@ void ECTransaction::Generate::truncate() {
|
||||
coll_t(spg_t(pgid, shard)),
|
||||
ghobject_t(oid, ghobject_t::NO_GEN, shard),
|
||||
ghobject_t(oid, entry->version.version, shard),
|
||||
start_align_prev,
|
||||
end - start_align_prev,
|
||||
start_align_prev);
|
||||
clone_start,
|
||||
end - clone_start,
|
||||
clone_start);
|
||||
|
||||
// First truncate to exactly the right size.
|
||||
t.truncate(
|
||||
@ -664,21 +698,15 @@ void ECTransaction::Generate::truncate() {
|
||||
ghobject_t(oid, ghobject_t::NO_GEN, shard),
|
||||
start_align_next);
|
||||
}
|
||||
|
||||
if (clone_start > start_align_prev) {
|
||||
clone_start = start_align_prev;
|
||||
}
|
||||
if (clone_end < end) {
|
||||
clone_end = end;
|
||||
}
|
||||
}
|
||||
shards_written(clone_shards);
|
||||
rollback_extents.emplace_back(make_pair(clone_start, clone_end));
|
||||
rollback_extents.emplace_back(make_pair(clone_start, clone_end - clone_start));
|
||||
rollback_shards.emplace_back(clone_shards);
|
||||
}
|
||||
}
|
||||
|
||||
void ECTransaction::Generate::overlay_writes() {
|
||||
ldpp_dout(dpp, 20) << __func__ << " start " << dendl;
|
||||
|
||||
for (auto &&extent: op.buffer_updates) {
|
||||
using BufferUpdate = PGTransaction::ObjectOperation::BufferUpdate;
|
||||
bufferlist bl;
|
||||
@ -708,6 +736,7 @@ void ECTransaction::Generate::appends_and_clone_ranges() {
|
||||
|
||||
extent_set clone_ranges = plan.will_write.get_extent_superset();
|
||||
uint64_t clone_max = ECUtil::align_next(plan.orig_size);
|
||||
ldpp_dout(dpp, 20) << __func__ << dendl;
|
||||
|
||||
if (op.delete_first) {
|
||||
clone_max = 0;
|
||||
@ -853,6 +882,7 @@ void ECTransaction::Generate::written_and_present_shards() {
|
||||
// written
|
||||
if (op.attr_updates.contains(OI_ATTR)) {
|
||||
object_info_t oi(*(op.attr_updates[OI_ATTR]));
|
||||
|
||||
// The majority of the updates to OI are made before a transaction is
|
||||
// submitted to ECBackend, these are cached by OBC and are encoded into
|
||||
// the OI attr update for the transaction. By the time the transaction
|
||||
|
||||
@ -571,11 +571,7 @@ void shard_extent_map_t::pad_on_shards(const shard_extent_set_t &pad_to,
|
||||
if (!pad_to.contains(shard)) {
|
||||
continue;
|
||||
}
|
||||
for (auto &[off, length] : pad_to.at(shard)) {
|
||||
bufferlist bl;
|
||||
bl.push_back(buffer::create_aligned(length, EC_ALIGN_SIZE));
|
||||
insert_in_shard(shard, off, bl);
|
||||
}
|
||||
pad_on_shard(pad_to.at(shard), shard);
|
||||
}
|
||||
}
|
||||
|
||||
@ -585,10 +581,14 @@ void shard_extent_map_t::pad_on_shard(const extent_set &pad_to,
|
||||
if (pad_to.size() == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
for (auto &[off, length] : pad_to) {
|
||||
bufferlist bl;
|
||||
bl.push_back(buffer::create_aligned(length, EC_ALIGN_SIZE));
|
||||
insert_in_shard(shard, off, bl);
|
||||
uint64_t start = align_prev(off);
|
||||
uint64_t end = align_next(off + length);
|
||||
|
||||
bl.push_back(buffer::create_aligned(end - start, EC_ALIGN_SIZE));
|
||||
insert_in_shard(shard, start, bl);
|
||||
}
|
||||
}
|
||||
|
||||
@ -599,11 +599,7 @@ void shard_extent_map_t::pad_on_shards(const extent_set &pad_to,
|
||||
return;
|
||||
}
|
||||
for (auto &shard : shards) {
|
||||
for (auto &[off, length] : pad_to) {
|
||||
bufferlist bl;
|
||||
bl.push_back(buffer::create_aligned(length, EC_ALIGN_SIZE));
|
||||
insert_in_shard(shard, off, bl);
|
||||
}
|
||||
pad_on_shard(pad_to, shard);
|
||||
}
|
||||
}
|
||||
|
||||
@ -657,32 +653,44 @@ int shard_extent_map_t::decode(const ErasureCodeInterfaceRef &ec_impl,
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (add_zero_padding_for_decode(object_size, need_set)) {
|
||||
// We added some zero buffers, which means our have and need set may change
|
||||
extent_maps.populate_bitset_set(have_set);
|
||||
need_set = shard_id_set::difference(want_set, have_set);
|
||||
}
|
||||
|
||||
shard_id_set decode_set = shard_id_set::intersection(need_set, sinfo->get_data_shards());
|
||||
shard_id_set encode_set = shard_id_set::intersection(need_set, sinfo->get_parity_shards());
|
||||
shard_extent_set_t read_mask(sinfo->get_k_plus_m());
|
||||
sinfo->ro_size_to_read_mask(object_size, read_mask);
|
||||
|
||||
if (!encode_set.empty()) {
|
||||
shard_extent_set_t read_mask(sinfo->get_k_plus_m());
|
||||
sinfo->ro_size_to_read_mask(object_size, read_mask);
|
||||
|
||||
/* The function has been asked to "decode" parity. To achieve this, we
|
||||
* need all the data shards to be present... So first see if there are
|
||||
* any missing...
|
||||
*/
|
||||
shard_id_set decode_for_parity_shards = shard_id_set::difference(sinfo->get_data_shards(), have_set);
|
||||
decode_for_parity_shards = shard_id_set::intersection(decode_for_parity_shards, read_mask.get_shard_id_set());
|
||||
|
||||
if (!decode_for_parity_shards.empty()) {
|
||||
/* So there are missing data shards which need decoding before we encode,
|
||||
* We need to add these to the decode set and insert buffers for the
|
||||
* decode to happen.
|
||||
*/
|
||||
decode_set.insert(decode_for_parity_shards);
|
||||
need_set.insert(decode_for_parity_shards);
|
||||
extent_set decode_for_parity = get_extent_superset();
|
||||
|
||||
for (auto shard : decode_for_parity_shards) {
|
||||
extent_set parity_pad;
|
||||
parity_pad.intersection_of(decode_for_parity, read_mask.at(shard));
|
||||
pad_on_shard(decode_for_parity, shard);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
int r = 0;
|
||||
if (!decode_set.empty()) {
|
||||
pad_on_shards(want, decode_set);
|
||||
/* If we are going to be encoding, we need to make sure all the necessary
|
||||
* shards are decoded. The get_min_available functions should have already
|
||||
* worked out what needs to be read for this.
|
||||
*/
|
||||
for (auto shard : encode_set) {
|
||||
extent_set decode_for_parity;
|
||||
decode_for_parity.intersection_of(want.at(shard), read_mask.at(shard));
|
||||
pad_on_shard(decode_for_parity, shard);
|
||||
}
|
||||
r = _decode(ec_impl, want_set, decode_set, dpp);
|
||||
}
|
||||
if (!r && !encode_set.empty()) {
|
||||
pad_on_shards(want, encode_set);
|
||||
pad_on_shards(get_extent_superset(), sinfo->get_parity_shards());
|
||||
r = encode(ec_impl, dpp, dedup_zeros?&need_set:nullptr);
|
||||
}
|
||||
|
||||
@ -705,14 +713,20 @@ int shard_extent_map_t::_decode(const ErasureCodeInterfaceRef &ec_impl,
|
||||
const shard_id_set &need_set,
|
||||
DoutPrefixProvider *dpp) {
|
||||
bool rebuild_req = false;
|
||||
|
||||
for (auto iter = begin_slice_iterator(need_set, dpp); !iter.is_end(); ++iter) {
|
||||
if (!iter.is_page_aligned()) {
|
||||
rebuild_req = true;
|
||||
break;
|
||||
}
|
||||
|
||||
shard_id_map<bufferptr> &in = iter.get_in_bufferptrs();
|
||||
shard_id_map<bufferptr> &out = iter.get_out_bufferptrs();
|
||||
|
||||
if (out.empty()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (int ret = ec_impl->decode_chunks(want_set, in, out)) {
|
||||
return ret;
|
||||
}
|
||||
|
||||
@ -211,22 +211,26 @@ class slice_iterator {
|
||||
|
||||
// If we have reached the end of the extent, we need to move that on too.
|
||||
if (bl_iter == emap_iter.get_val().end()) {
|
||||
// NOTE: Despite appearances, the following is happening BEFORE
|
||||
// the caller gets to use the buffer pointers (since the in/out is
|
||||
// set a few lines above). This means that the caller must not
|
||||
// check the CRC.
|
||||
if (out_set.contains(shard)) {
|
||||
invalidate_crcs(shard);
|
||||
}
|
||||
++emap_iter;
|
||||
if (emap_iter == input[shard].end()) {
|
||||
erase = true;
|
||||
} else {
|
||||
if (out_set.contains(shard)) {
|
||||
bufferlist bl = emap_iter.get_val();
|
||||
bl.invalidate_crc();
|
||||
}
|
||||
iters.at(shard).second = emap_iter.get_val().begin();
|
||||
if (zeros) {
|
||||
zeros->emplace(shard, emap_iter.get_off(), emap_iter.get_len());
|
||||
}
|
||||
}
|
||||
}
|
||||
} else
|
||||
} else {
|
||||
ceph_assert(iter_offset > start);
|
||||
}
|
||||
|
||||
if (erase) {
|
||||
iter = iters.erase(iter);
|
||||
@ -249,6 +253,11 @@ class slice_iterator {
|
||||
}
|
||||
}
|
||||
|
||||
void invalidate_crcs(shard_id_t shard) {
|
||||
bufferlist bl = iters.at(shard).first.get_val();
|
||||
bl.invalidate_crc();
|
||||
}
|
||||
|
||||
public:
|
||||
slice_iterator(
|
||||
mini_flat_map<shard_id_t, extent_map> &_input,
|
||||
@ -650,10 +659,6 @@ public:
|
||||
ErasureCodeInterface::FLAG_EC_PLUGIN_REQUIRE_SUB_CHUNKS) != 0;
|
||||
}
|
||||
|
||||
bool get_is_hinfo_required() const {
|
||||
return !supports_ec_overwrites();
|
||||
}
|
||||
|
||||
bool supports_partial_reads() const {
|
||||
return (plugin_flags &
|
||||
ErasureCodeInterface::FLAG_EC_PLUGIN_PARTIAL_READ_OPTIMIZATION) != 0;
|
||||
@ -1054,16 +1059,9 @@ public:
|
||||
}
|
||||
}
|
||||
|
||||
bool add_zero_padding_for_decode(uint64_t object_size, shard_id_set &exclude_set) {
|
||||
shard_extent_set_t zeros(sinfo->get_k_plus_m());
|
||||
sinfo->ro_size_to_zero_mask(object_size, zeros);
|
||||
extent_set superset = get_extent_superset();
|
||||
void add_zero_padding_for_decode(ECUtil::shard_extent_set_t &zeros) {
|
||||
bool changed = false;
|
||||
for (auto &&[shard, z] : zeros) {
|
||||
if (exclude_set.contains(shard)) {
|
||||
continue;
|
||||
}
|
||||
z.intersection_of(superset);
|
||||
for (auto [off, len] : z) {
|
||||
changed = true;
|
||||
bufferlist bl;
|
||||
@ -1075,8 +1073,6 @@ public:
|
||||
if (changed) {
|
||||
compute_ro_range();
|
||||
}
|
||||
|
||||
return changed;
|
||||
}
|
||||
|
||||
template <typename IntervalSetT> requires is_interval_set_v<IntervalSetT>
|
||||
|
||||
@ -2034,9 +2034,17 @@ void OSDMap::clean_temps(CephContext *cct,
|
||||
// force a change of primary shard - do not remove pg_temp
|
||||
// if it is being used for this purpose
|
||||
if (pool->allows_ecoptimizations()) {
|
||||
if (nextmap._pick_primary(pg.second) != primary) {
|
||||
// pg_temp still required
|
||||
keep = true;
|
||||
// primary might not be in raw_up - so keep pg_temp unless
|
||||
// proven that the primary is not a non-primary shard
|
||||
keep = true;
|
||||
for (unsigned int i = 0; i < raw_up.size(); ++i) {
|
||||
if (raw_up[i] == primary) {
|
||||
if (!pool->is_nonprimary_shard(shard_id_t(i))) {
|
||||
// pg_temp not required
|
||||
keep = false;
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!keep) {
|
||||
|
||||
@ -440,8 +440,23 @@ void PGBackend::partial_write(
|
||||
}
|
||||
auto &&[old_v, new_v] = pwlc_iter->second;
|
||||
if (old_v == new_v) {
|
||||
old_v = previous_version;
|
||||
new_v = entry.version;
|
||||
if (old_v.version == eversion_t::max().version) {
|
||||
// shard is backfilling or in async recovery, pwlc is
|
||||
// invalid
|
||||
ldpp_dout(dpp, 20) << __func__ << " pwlc invalid " << shard
|
||||
<< dendl;
|
||||
} else if (old_v.version >= entry.version.version) {
|
||||
// Abnormal case - consider_adjusting_pwlc may advance pwlc
|
||||
// during peering because all shards have updates but these
|
||||
// have not been marked complete. At the end of peering
|
||||
// partial_write catches up with these entries - these need
|
||||
// to be ignored to preserve old_v.epoch
|
||||
ldpp_dout(dpp, 20) << __func__ << " pwlc is ahead of entry " << shard
|
||||
<< dendl;
|
||||
} else {
|
||||
old_v = previous_version;
|
||||
new_v = entry.version;
|
||||
}
|
||||
} else if (new_v == previous_version) {
|
||||
// Subsequent partial write, contiguous versions
|
||||
new_v = entry.version;
|
||||
@ -453,9 +468,17 @@ void PGBackend::partial_write(
|
||||
} else if (pwlc_iter != info->partial_writes_last_complete.end()) {
|
||||
auto &&[old_v, new_v] = pwlc_iter->second;
|
||||
// Log updated or shard absent, partial write entry is a no-op
|
||||
// FIXME: In a later commit (or later PR) we should look at other ways of
|
||||
// actually clearing the PWLC once all shards have seen the update.
|
||||
old_v = new_v = entry.version;
|
||||
if (old_v.version == eversion_t::max().version) {
|
||||
// shard is backfilling or in async recovery, pwlc is invalid
|
||||
ldpp_dout(dpp, 20) << __func__ << " pwlc invalid " << shard
|
||||
<< dendl;
|
||||
} else if (old_v.version >= entry.version.version) {
|
||||
// Abnormal case - see above
|
||||
ldpp_dout(dpp, 20) << __func__ << " pwlc is ahead of entry " << shard
|
||||
<< dendl;
|
||||
} else {
|
||||
old_v = new_v = entry.version;
|
||||
}
|
||||
}
|
||||
}
|
||||
ldpp_dout(dpp, 20) << __func__ << " after pwlc="
|
||||
|
||||
@ -349,10 +349,18 @@ void PGLog::rewind_divergent_log(eversion_t newhead,
|
||||
if (info.last_complete > newhead)
|
||||
info.last_complete = newhead;
|
||||
|
||||
auto divergent = log.rewind_from_head(newhead);
|
||||
bool need_dirty_log;
|
||||
auto divergent = log.rewind_from_head(newhead, &need_dirty_log);
|
||||
if (!divergent.empty()) {
|
||||
mark_dirty_from(divergent.front().version);
|
||||
} else if (need_dirty_log) {
|
||||
// can_rollback_to and/or rollback_info_trimmed_to have been modified
|
||||
// and need checkpointing
|
||||
dout(10) << "rewind_divergent_log crt = "
|
||||
<< log.get_can_rollback_to() << dendl;
|
||||
dirty_log = true;
|
||||
}
|
||||
|
||||
for (auto &&entry: divergent) {
|
||||
dout(10) << "rewind_divergent_log future divergent " << entry << dendl;
|
||||
}
|
||||
|
||||
@ -296,8 +296,8 @@ public:
|
||||
eversion_t previous_version) {});
|
||||
}
|
||||
|
||||
mempool::osd_pglog::list<pg_log_entry_t> rewind_from_head(eversion_t newhead) {
|
||||
auto divergent = pg_log_t::rewind_from_head(newhead);
|
||||
mempool::osd_pglog::list<pg_log_entry_t> rewind_from_head(eversion_t newhead, bool *dirty_log = nullptr) {
|
||||
auto divergent = pg_log_t::rewind_from_head(newhead, dirty_log);
|
||||
index();
|
||||
reset_rollback_info_trimmed_to_riter();
|
||||
return divergent;
|
||||
@ -1404,11 +1404,7 @@ public:
|
||||
invalidate_stats = invalidate_stats || !p->is_error();
|
||||
if (log) {
|
||||
ldpp_dout(dpp, 20) << "update missing, append " << *p << dendl;
|
||||
// Skip the log entry if it is a partial write that did not involve
|
||||
// this shard
|
||||
if (!pool.is_nonprimary_shard(shard) || p->is_written_shard(shard)) {
|
||||
log->add(*p);
|
||||
}
|
||||
log->add(*p);
|
||||
}
|
||||
if (p->soid <= last_backfill &&
|
||||
!p->is_error()) {
|
||||
@ -1716,10 +1712,18 @@ public:
|
||||
if (debug_verify_stored_missing) {
|
||||
auto miter = missing.get_items().find(i->soid);
|
||||
ceph_assert(miter != missing.get_items().end());
|
||||
ceph_assert(miter->second.need == i->version);
|
||||
// the 'have' version is reset if an object is deleted,
|
||||
// then created again
|
||||
ceph_assert(miter->second.have == oi.version || miter->second.have == eversion_t());
|
||||
if (ec_optimizations_enabled) {
|
||||
// non-primary shards in an optimized pool may not have updates
|
||||
// because of partial writes, which may result in oi.version being
|
||||
// less than have
|
||||
ceph_assert(miter->second.need >= i->version);
|
||||
ceph_assert(miter->second.have >= oi.version || miter->second.have == eversion_t());
|
||||
} else {
|
||||
ceph_assert(miter->second.need == i->version);
|
||||
ceph_assert(miter->second.have == oi.version || miter->second.have == eversion_t());
|
||||
}
|
||||
checked.insert(i->soid);
|
||||
} else {
|
||||
missing.add(i->soid, i->version, oi.version, i->is_delete());
|
||||
|
||||
@ -321,9 +321,54 @@ void PeeringState::query_unfound(Formatter *f, string state)
|
||||
return;
|
||||
}
|
||||
|
||||
void PeeringState::apply_pwlc(const std::pair<eversion_t, eversion_t> pwlc,
|
||||
const pg_shard_t &shard,
|
||||
pg_info_t &info,
|
||||
pg_log_t *log1,
|
||||
PGLog *log2)
|
||||
{
|
||||
// Check if last_complete and last_update can be advanced based on
|
||||
// knowledge of partial_writes
|
||||
const auto & [fromversion, toversion] = pwlc;
|
||||
if (toversion > info.last_update) {
|
||||
if (fromversion.version <= info.last_update.version) {
|
||||
if (info.last_complete == info.last_update) {
|
||||
psdout(10) << "osd." << shard << " has last_complete"
|
||||
<< "=last_update " << info.last_update
|
||||
<< " pwlc can advance both to " << toversion
|
||||
<< dendl;
|
||||
info.last_complete = toversion;
|
||||
} else {
|
||||
psdout(10) << "osd." << shard << " has last_complete "
|
||||
<< info.last_complete << " and last_update "
|
||||
<< info.last_update
|
||||
<< " pwlc can advance last_update to " << toversion
|
||||
<< dendl;
|
||||
}
|
||||
info.last_update = toversion;
|
||||
if (log1 && toversion > log1->head) {
|
||||
log1->head = toversion;
|
||||
}
|
||||
if (log2 && toversion > log2->get_head()) {
|
||||
log2->set_head(toversion);
|
||||
}
|
||||
} else {
|
||||
psdout(10) << "osd." << shard << " has last_complete "
|
||||
<< info.last_complete << " and last_update "
|
||||
<< info.last_update
|
||||
<< " cannot apply pwlc from " << fromversion
|
||||
<< " to " << toversion
|
||||
<< dendl;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void PeeringState::update_peer_info(const pg_shard_t &from,
|
||||
const pg_info_t &oinfo)
|
||||
{
|
||||
// Merge pwlc information from another shard into
|
||||
// info.partial_writes_last_complete keeping the newest
|
||||
// updates
|
||||
if (!oinfo.partial_writes_last_complete.empty()) {
|
||||
bool updated = false;
|
||||
// oinfo includes partial_writes_last_complete data.
|
||||
@ -336,8 +381,10 @@ void PeeringState::update_peer_info(const pg_shard_t &from,
|
||||
info.partial_writes_last_complete[shard];
|
||||
// Prefer pwlc with a newer toversion, if toversion matches prefer an
|
||||
// older fromversion.
|
||||
if ((otoversion > toversion) ||
|
||||
((otoversion == toversion) && (ofromversion < fromversion))) {
|
||||
if ((ofromversion.epoch > fromversion.epoch) ||
|
||||
((ofromversion.epoch == fromversion.epoch) && (otoversion > toversion)) ||
|
||||
((ofromversion.epoch == fromversion.epoch) && (otoversion == toversion) &&
|
||||
(ofromversion.version < fromversion.version))) {
|
||||
if (!updated) {
|
||||
updated = true;
|
||||
psdout(10) << "osd." << from
|
||||
@ -363,61 +410,25 @@ void PeeringState::update_peer_info(const pg_shard_t &from,
|
||||
}
|
||||
}
|
||||
// 3 cases:
|
||||
// We are the primary - from is the shard that sent the oinfo
|
||||
// We are a replica - from is the primary, it will not have pwlc infomation for itself
|
||||
// Merge - from is pg_whoami, oinfo is a source pg that is being merged
|
||||
// 1. This is the primary, from is the shard that sent the oinfo which may
|
||||
// have more up to date pwlc. There may be multiple peer_info's for the
|
||||
// shard id that can be updated by applying pwlc
|
||||
// 2. This is a replica/stray - from is the primary (which never has pwlc
|
||||
// information), there is nothing to update
|
||||
// 3. This is a merge - from is pg_whoami, there is nothing to update
|
||||
if ((from != pg_whoami) &&
|
||||
info.partial_writes_last_complete.contains(from.shard)) {
|
||||
// Check if last_complete and last_update can be advanced based on
|
||||
// knowledge of partial_writes
|
||||
const auto & [fromversion, toversion] =
|
||||
info.partial_writes_last_complete[from.shard];
|
||||
if (toversion > peer_info[from].last_complete) {
|
||||
if (fromversion <= peer_info[from].last_complete) {
|
||||
psdout(10) << "osd." << from << " has last_complete "
|
||||
<< peer_info[from].last_complete
|
||||
<< " but pwlc says its at " << toversion
|
||||
<< dendl;
|
||||
peer_info[from].last_complete = toversion;
|
||||
if (toversion > peer_info[from].last_update) {
|
||||
peer_info[from].last_update = toversion;
|
||||
}
|
||||
} else {
|
||||
psdout(10) << "osd." << from << " has last_complete "
|
||||
<< peer_info[from].last_complete
|
||||
<< " cannot apply pwlc from " << fromversion
|
||||
<< " to " << toversion
|
||||
<< dendl;
|
||||
for (auto & [shard, peer] : peer_info) {
|
||||
if (shard.shard != from.shard) {
|
||||
continue;
|
||||
}
|
||||
apply_pwlc(info.partial_writes_last_complete[from.shard], shard, peer);
|
||||
}
|
||||
}
|
||||
// Non-primary shards might need to apply pwlc to update info
|
||||
if (info.partial_writes_last_complete.contains(pg_whoami.shard)) {
|
||||
// Check if last_complete and last_update can be advanced based on
|
||||
// knowledge of partial_writes
|
||||
const auto & [fromversion, toversion] =
|
||||
info.partial_writes_last_complete[pg_whoami.shard];
|
||||
if (toversion > info.last_complete) {
|
||||
if (fromversion <= info.last_complete) {
|
||||
psdout(10) << "osd." << pg_whoami << " has last_complete "
|
||||
<< info.last_complete
|
||||
<< " but pwlc says its at " << toversion
|
||||
<< dendl;
|
||||
info.last_complete = toversion;
|
||||
if (toversion > info.last_update) {
|
||||
info.last_update = toversion;
|
||||
}
|
||||
if (toversion > pg_log.get_head()) {
|
||||
pg_log.set_head(toversion);
|
||||
}
|
||||
} else {
|
||||
psdout(10) << "osd." << pg_whoami << " has last_complete "
|
||||
<< info.last_complete
|
||||
<< " cannot apply pwlc from " << fromversion
|
||||
<< " to " << toversion
|
||||
<< dendl;
|
||||
}
|
||||
}
|
||||
apply_pwlc(info.partial_writes_last_complete[pg_whoami.shard], pg_whoami,
|
||||
info, &pg_log);
|
||||
}
|
||||
}
|
||||
|
||||
@ -2722,6 +2733,9 @@ bool PeeringState::search_for_missing(
|
||||
tinfo.pgid.shard = pg_whoami.shard;
|
||||
// add partial write from our info
|
||||
tinfo.partial_writes_last_complete = info.partial_writes_last_complete;
|
||||
if (info.partial_writes_last_complete.contains(from.shard)) {
|
||||
apply_pwlc(info.partial_writes_last_complete[from.shard], from, tinfo);
|
||||
}
|
||||
if (!tinfo.partial_writes_last_complete.empty()) {
|
||||
psdout(20) << "sending info to " << from
|
||||
<< " pwlc=" << tinfo.partial_writes_last_complete
|
||||
@ -2983,7 +2997,7 @@ void PeeringState::activate(
|
||||
<< " is up to date, queueing in pending_activators" << dendl;
|
||||
if (!info.partial_writes_last_complete.empty()) {
|
||||
psdout(20) << "sending info to " << peer
|
||||
<< " pwcl=" << info.partial_writes_last_complete
|
||||
<< " pwlc=" << info.partial_writes_last_complete
|
||||
<< " info=" << info
|
||||
<< dendl;
|
||||
}
|
||||
@ -3019,6 +3033,7 @@ void PeeringState::activate(
|
||||
<< "] " << pi.last_backfill
|
||||
<< " to " << info.last_update;
|
||||
|
||||
pi.partial_writes_last_complete = info.partial_writes_last_complete;
|
||||
pi.last_update = info.last_update;
|
||||
pi.last_complete = info.last_update;
|
||||
pi.set_last_backfill(hobject_t());
|
||||
@ -3253,22 +3268,44 @@ void PeeringState::proc_primary_info(
|
||||
}
|
||||
}
|
||||
|
||||
void PeeringState::consider_adjusting_pwlc(eversion_t last_complete)
|
||||
{
|
||||
for (const auto & [shard, versionrange] :
|
||||
info.partial_writes_last_complete) {
|
||||
auto [fromversion, toversion] = versionrange;
|
||||
if (last_complete > toversion) {
|
||||
// Full writes are being rolled forward, eventually
|
||||
// partial_write will be called to advance pwlc, but we need
|
||||
// to preempt that here before proc_master_log considers
|
||||
// rolling forward partial writes
|
||||
info.partial_writes_last_complete[shard] = std::pair(last_complete,
|
||||
last_complete);
|
||||
psdout(10) << "shard " << shard << " pwlc rolled forward to "
|
||||
<< info.partial_writes_last_complete[shard] << dendl;
|
||||
} else if (last_complete < toversion) {
|
||||
// A divergent update has advanced pwlc adhead of last_complete,
|
||||
// roll backwards to the last completed full write and then
|
||||
// let proc_master_log roll forward partial writes
|
||||
info.partial_writes_last_complete[shard] = std::pair(last_complete,
|
||||
last_complete);
|
||||
psdout(10) << "shard " << shard << " pwlc rolled backward to "
|
||||
<< info.partial_writes_last_complete[shard] << dendl;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void PeeringState::consider_rollback_pwlc(eversion_t last_complete)
|
||||
{
|
||||
for (const auto & [shard, versionrange] :
|
||||
info.partial_writes_last_complete) {
|
||||
auto [fromversion, toversion] = versionrange;
|
||||
if (last_complete < fromversion) {
|
||||
if (last_complete.version < fromversion.version) {
|
||||
// It is possible that we need to rollback pwlc, this can happen if
|
||||
// peering is attempted with an OSD missing but does not manage to
|
||||
// activate (typically because of a wait upthru) before the missing
|
||||
// OSD returns
|
||||
info.partial_writes_last_complete[shard] = std::pair(last_complete,
|
||||
last_complete);
|
||||
// Assign the current epoch to the version number so that this is
|
||||
// recognised as the newest pwlc update
|
||||
info.partial_writes_last_complete[shard].second.epoch =
|
||||
get_osdmap_epoch();
|
||||
psdout(10) << "shard " << shard << " pwlc rolled back to "
|
||||
<< info.partial_writes_last_complete[shard] << dendl;
|
||||
} else if (last_complete < toversion) {
|
||||
@ -3276,6 +3313,11 @@ void PeeringState::consider_rollback_pwlc(eversion_t last_complete)
|
||||
psdout(10) << "shard " << shard << " pwlc rolled back to "
|
||||
<< info.partial_writes_last_complete[shard] << dendl;
|
||||
}
|
||||
// Always assign the current epoch to the version number so that
|
||||
// pwlc adjustments made by the whole proc_master_log process
|
||||
// are recognized as the newest updates
|
||||
info.partial_writes_last_complete[shard].first.epoch =
|
||||
get_osdmap_epoch();
|
||||
}
|
||||
}
|
||||
|
||||
@ -3288,28 +3330,8 @@ void PeeringState::proc_master_log(
|
||||
ceph_assert(!is_peered() && is_primary());
|
||||
|
||||
if (info.partial_writes_last_complete.contains(from.shard)) {
|
||||
// Check if last_complete and last_update can be advanced based on
|
||||
// knowledge of partial_writes
|
||||
const auto & [fromversion, toversion] =
|
||||
info.partial_writes_last_complete[from.shard];
|
||||
if (toversion > oinfo.last_complete) {
|
||||
if (fromversion <= oinfo.last_complete) {
|
||||
psdout(10) << "osd." << from << " has last_complete "
|
||||
<< oinfo.last_complete
|
||||
<< " but pwlc says its at " << toversion << dendl;
|
||||
oinfo.last_complete = toversion;
|
||||
if (toversion > oinfo.last_update) {
|
||||
oinfo.last_update = toversion;
|
||||
}
|
||||
if (toversion > olog.head) {
|
||||
olog.head = toversion;
|
||||
}
|
||||
} else {
|
||||
psdout(10) << "osd." << from << " has last_complete "
|
||||
<< oinfo.last_complete << " cannot apply pwlc from "
|
||||
<< fromversion << " to " << toversion << dendl;
|
||||
}
|
||||
}
|
||||
apply_pwlc(info.partial_writes_last_complete[from.shard], from, oinfo,
|
||||
&olog);
|
||||
}
|
||||
// For partial writes we may be able to keep some of the divergent entries
|
||||
if (olog.head < pg_log.get_head()) {
|
||||
@ -3329,7 +3351,7 @@ void PeeringState::proc_master_log(
|
||||
all_info[pg_whoami] = info;
|
||||
// Normal case is that both logs have entry olog.head
|
||||
bool can_check_next_entry = (p->version == olog.head);
|
||||
if (p->version < olog.head) {
|
||||
if ((p->version < olog.head) || (p == pg_log.get_log().log.begin())) {
|
||||
// After a PG split there may be gaps in the log where entries were
|
||||
// split to the other PG. This can result in olog.head being ahead
|
||||
// of p->version. So long as there are no entries in olog between
|
||||
@ -3342,6 +3364,10 @@ void PeeringState::proc_master_log(
|
||||
can_check_next_entry = true;
|
||||
}
|
||||
}
|
||||
if (can_check_next_entry) {
|
||||
consider_adjusting_pwlc(p->version);
|
||||
}
|
||||
PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(t)};
|
||||
while (can_check_next_entry) {
|
||||
++p;
|
||||
if (p == pg_log.get_log().log.end()) {
|
||||
@ -3384,7 +3410,14 @@ void PeeringState::proc_master_log(
|
||||
// This entry can be kept, only shards that didn't participate in
|
||||
// the partial write missed the update
|
||||
psdout(20) << "keeping entry " << p->version << dendl;
|
||||
olog.head = p->version;
|
||||
eversion_t previous_version;
|
||||
if (p == pg_log.get_log().log.begin()) {
|
||||
previous_version = pg_log.get_tail();
|
||||
} else {
|
||||
previous_version = std::prev(p)->version;
|
||||
}
|
||||
rollbacker.get()->partial_write(&info, previous_version, *p);
|
||||
olog.head = p->version;
|
||||
|
||||
// We need to continue processing the log, so don't break.
|
||||
}
|
||||
@ -3395,7 +3428,6 @@ void PeeringState::proc_master_log(
|
||||
// non-divergent).
|
||||
merge_log(t, oinfo, std::move(olog), from);
|
||||
peer_info[from] = oinfo;
|
||||
update_peer_info(from, oinfo);
|
||||
psdout(10) << " peer osd." << from << " now " << oinfo
|
||||
<< " " << omissing << dendl;
|
||||
might_have_unfound.insert(from);
|
||||
@ -3422,13 +3454,17 @@ void PeeringState::proc_master_log(
|
||||
|
||||
void PeeringState::proc_replica_log(
|
||||
pg_info_t &oinfo,
|
||||
const pg_log_t &olog,
|
||||
pg_log_t &olog,
|
||||
pg_missing_t&& omissing,
|
||||
pg_shard_t from)
|
||||
{
|
||||
psdout(10) << "proc_replica_log for osd." << from << ": "
|
||||
<< oinfo << " " << olog << " " << omissing << dendl;
|
||||
|
||||
if (info.partial_writes_last_complete.contains(from.shard)) {
|
||||
apply_pwlc(info.partial_writes_last_complete[from.shard], from, oinfo,
|
||||
&olog);
|
||||
}
|
||||
pg_log.proc_replica_log(oinfo, olog, omissing, from, pool.info.allows_ecoptimizations());
|
||||
|
||||
peer_info[from] = oinfo;
|
||||
@ -3767,13 +3803,15 @@ void PeeringState::merge_from(
|
||||
past_intervals = source->past_intervals;
|
||||
}
|
||||
|
||||
// merge pwlc
|
||||
// merge pwlc - reset
|
||||
if (!info.partial_writes_last_complete.empty()) {
|
||||
psdout(10) << "before pwlc=" << info.partial_writes_last_complete << dendl;
|
||||
}
|
||||
update_peer_info(pg_whoami, source->info);
|
||||
if (!info.partial_writes_last_complete.empty()) {
|
||||
psdout(10) << "after pwlc=" << info.partial_writes_last_complete << dendl;
|
||||
for (auto &&[shard, versionrange] :
|
||||
info.partial_writes_last_complete) {
|
||||
auto &&[old_v, new_v] = versionrange;
|
||||
old_v = new_v = info.last_update;
|
||||
old_v.epoch = get_osdmap_epoch();
|
||||
}
|
||||
psdout(10) << "merged pwlc=" << info.partial_writes_last_complete << dendl;
|
||||
}
|
||||
}
|
||||
|
||||
@ -4538,7 +4576,6 @@ void PeeringState::add_log_entry(const pg_log_entry_t& e, ObjectStore::Transacti
|
||||
enum PGLog::NonPrimary nonprimary{pool.info.is_nonprimary_shard(info.pgid.shard)};
|
||||
PGLog::LogEntryHandlerRef handler{pl->get_log_handler(t)};
|
||||
pg_log.add(e, nonprimary, applied, &info, handler.get());
|
||||
psdout(10) << "add_log_entry " << e << dendl;
|
||||
}
|
||||
|
||||
|
||||
@ -4590,6 +4627,16 @@ void PeeringState::append_log(
|
||||
* object is deleted before we can _merge_object_divergent_entries().
|
||||
*/
|
||||
pg_log.skip_rollforward(&info, handler.get());
|
||||
/* Invalidate pwlc for this shard until the next interval when
|
||||
* it will be updated with the pwlc from another shard
|
||||
*/
|
||||
for (auto & [shard, versionrange] :
|
||||
info.partial_writes_last_complete) {
|
||||
auto & [fromversion, toversion] = versionrange;
|
||||
fromversion.epoch = 0;
|
||||
fromversion.version = eversion_t::max().version;
|
||||
toversion = fromversion;
|
||||
}
|
||||
}
|
||||
|
||||
for (auto p = logv.begin(); p != logv.end(); ++p) {
|
||||
@ -6834,7 +6881,7 @@ boost::statechart::result PeeringState::ReplicaActive::react(
|
||||
i.history.last_epoch_started = evt.activation_epoch;
|
||||
i.history.last_interval_started = i.history.same_interval_since;
|
||||
if (!i.partial_writes_last_complete.empty()) {
|
||||
psdout(20) << "sending info to " << ps->get_primary() << " pwcl="
|
||||
psdout(20) << "sending info to " << ps->get_primary() << " pwlc="
|
||||
<< i.partial_writes_last_complete << " info=" << i << dendl;
|
||||
}
|
||||
rctx.send_info(
|
||||
@ -6887,31 +6934,8 @@ boost::statechart::result PeeringState::ReplicaActive::react(const MLogRec& loge
|
||||
MOSDPGLog *msg = logevt.msg.get();
|
||||
ObjectStore::Transaction &t = context<PeeringMachine>().get_cur_transaction();
|
||||
if (msg->info.partial_writes_last_complete.contains(ps->pg_whoami.shard)) {
|
||||
// Check if last_complete and last_update can be advanced based on
|
||||
// knowledge of partial_writes
|
||||
const auto & [fromversion, toversion] =
|
||||
msg->info.partial_writes_last_complete[ps->pg_whoami.shard];
|
||||
if (toversion > ps->info.last_complete) {
|
||||
if (fromversion <= ps->info.last_complete) {
|
||||
psdout(10) << "last_complete " << ps->info.last_complete
|
||||
<< " but pwlc from " << logevt.from
|
||||
<< " is at " << toversion << dendl;
|
||||
ps->info.last_complete = toversion;
|
||||
if (toversion > ps->info.last_update) {
|
||||
ps->info.last_update = toversion;
|
||||
}
|
||||
// Advance head to avoid an assert in merge log
|
||||
if (msg->log.tail > ps->pg_log.get_head()) {
|
||||
psdout(10) << "pwlc advancing log head from "
|
||||
<< ps->pg_log.get_head() << " to " << toversion << dendl;
|
||||
ps->pg_log.set_head(toversion);
|
||||
}
|
||||
} else {
|
||||
psdout(10) << "last_complete " << ps->info.last_complete
|
||||
<< " cannot apply pwlc from "
|
||||
<< fromversion << " to " << toversion << dendl;
|
||||
}
|
||||
}
|
||||
ps->apply_pwlc(msg->info.partial_writes_last_complete[ps->pg_whoami.shard],
|
||||
ps->pg_whoami, ps->info, &ps->pg_log);
|
||||
}
|
||||
ps->merge_log(t, logevt.msg->info, std::move(logevt.msg->log), logevt.from);
|
||||
ps->update_peer_info(logevt.from, logevt.msg->info);
|
||||
@ -7028,32 +7052,8 @@ boost::statechart::result PeeringState::Stray::react(const MLogRec& logevt)
|
||||
ps->pg_log.reset_backfill();
|
||||
} else {
|
||||
if (msg->info.partial_writes_last_complete.contains(ps->pg_whoami.shard)) {
|
||||
// Check if last_complete and last_update can be advanced based on
|
||||
// knowledge of partial_writes
|
||||
const auto & [fromversion, toversion] =
|
||||
msg->info.partial_writes_last_complete[ps->pg_whoami.shard];
|
||||
if (toversion > ps->info.last_complete) {
|
||||
if (fromversion <= ps->info.last_complete) {
|
||||
psdout(10) << "last_complete " << ps->info.last_complete
|
||||
<< " but pwlc from " << logevt.from
|
||||
<< " is at " << toversion << dendl;
|
||||
ps->info.last_complete = toversion;
|
||||
if (toversion > ps->info.last_update) {
|
||||
ps->info.last_update = toversion;
|
||||
}
|
||||
// Need to do this to avoid an assert in merge log
|
||||
if (msg->log.tail > ps->pg_log.get_head()) {
|
||||
psdout(10) << "pwlc advancing log head from "
|
||||
<< ps->pg_log.get_head() << " to " << toversion
|
||||
<< dendl;
|
||||
ps->pg_log.set_head(toversion);
|
||||
}
|
||||
} else {
|
||||
psdout(10) << "last_complete " << ps->info.last_complete
|
||||
<< " cannot apply pwlc from "
|
||||
<< fromversion << " to " << toversion << dendl;
|
||||
}
|
||||
}
|
||||
ps->apply_pwlc(msg->info.partial_writes_last_complete[ps->pg_whoami.shard],
|
||||
ps->pg_whoami, ps->info, &ps->pg_log);
|
||||
}
|
||||
ps->merge_log(t, msg->info, std::move(msg->log), logevt.from);
|
||||
ps->update_peer_info(logevt.from, msg->info);
|
||||
@ -7105,7 +7105,7 @@ boost::statechart::result PeeringState::Stray::react(const MInfoRec& infoevt)
|
||||
<< " our last_update=" << ps->info.last_update << dendl;
|
||||
// Our last update must be in the range described by partial write
|
||||
// last_complete
|
||||
ceph_assert(ps->info.last_update >= pwlc.first);
|
||||
ceph_assert(ps->info.last_update.version >= pwlc.first.version);
|
||||
// Last complete must match the partial write last_update
|
||||
ceph_assert(pwlc.second == infoevt.info.last_update);
|
||||
} else {
|
||||
@ -7536,7 +7536,9 @@ boost::statechart::result PeeringState::GetLog::react(const GotLog&)
|
||||
// Our log was behind that of the auth_log_shard which was a non-primary
|
||||
// with a sparse log. We have just got a log from a primary shard to
|
||||
// catch up and now need to recheck if we need to rollback the log to
|
||||
// the auth_log_shard
|
||||
// the auth_log_shard. Discard the received missing log as this does
|
||||
// may not be consistent with the authorative log
|
||||
ps->peer_missing.erase(auth_log_shard);
|
||||
psdout(10) << "repeating auth_log_shard selection" << dendl;
|
||||
post_event(RepeatGetLog());
|
||||
return discard_event();
|
||||
@ -7820,67 +7822,6 @@ PeeringState::GetMissing::GetMissing(my_context ctx)
|
||||
continue;
|
||||
}
|
||||
|
||||
// If the peer log is only divergent because of partial writes then
|
||||
// roll forward the peer to cover writes it was not involved in.
|
||||
if (pi.last_update < ps->info.last_update) {
|
||||
// Search backwards through log looking for a match with peer's head
|
||||
// entry
|
||||
mempool::osd_pglog::list<pg_log_entry_t>::const_iterator p =
|
||||
ps->pg_log.get_log().log.end();
|
||||
while (p != ps->pg_log.get_log().log.begin()) {
|
||||
--p;
|
||||
if (p->version.version <= pi.last_update.version) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (pi.last_update == pi.last_complete &&
|
||||
p->version == pi.last_update) {
|
||||
// Matched peer's head entry - see if we can advance last_update
|
||||
// because of partial written shards
|
||||
eversion_t old_last_update = pi.last_update;
|
||||
if (ps->info.partial_writes_last_complete.contains(i->shard) &&
|
||||
ps->info.partial_writes_last_complete[i->shard].first <
|
||||
old_last_update) {
|
||||
old_last_update =
|
||||
ps->info.partial_writes_last_complete[i->shard].first;
|
||||
}
|
||||
++p;
|
||||
bool advanced = false;
|
||||
while (p != ps->pg_log.get_log().log.end()) {
|
||||
if (p->is_written_shard(i->shard)) {
|
||||
psdout(20) << "log entry " << p->version
|
||||
<< " written_shards=" << p->written_shards
|
||||
<< " is divergent" << dendl;
|
||||
break;
|
||||
}
|
||||
pi.last_update = p->version;
|
||||
pi.last_complete = p->version;
|
||||
// Update partial_writes_last_complete
|
||||
if (ps->info.partial_writes_last_complete.contains(i->shard)) {
|
||||
// Existing pwlc entry - only update if p->version is newer
|
||||
if (ps->info.partial_writes_last_complete[i->shard].second <
|
||||
p->version) {
|
||||
ps->info.partial_writes_last_complete[i->shard] =
|
||||
std::pair(old_last_update, p->version);
|
||||
}
|
||||
} else {
|
||||
// No existing pwlc entry - create one
|
||||
ps->info.partial_writes_last_complete[i->shard] =
|
||||
std::pair(old_last_update, p->version);
|
||||
}
|
||||
advanced = true;
|
||||
++p;
|
||||
}
|
||||
if (advanced) {
|
||||
psdout(20) << "shard " << i->shard << " pwlc="
|
||||
<< ps->info.partial_writes_last_complete.at(i->shard)
|
||||
<< " last_complete=" << ps->info.last_complete
|
||||
<< " last_update=" << pi.last_update
|
||||
<< dendl;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (pi.last_update == pi.last_complete && // peer has no missing
|
||||
pi.last_update == ps->info.last_update) { // peer is up to date
|
||||
// replica has no missing and identical log as us. no need to
|
||||
|
||||
@ -1585,6 +1585,25 @@ public:
|
||||
|
||||
void update_heartbeat_peers();
|
||||
void query_unfound(Formatter *f, std::string state);
|
||||
void apply_pwlc(const std::pair<eversion_t, eversion_t> pwlc,
|
||||
const pg_shard_t &shard,
|
||||
pg_info_t &info,
|
||||
pg_log_t *log1,
|
||||
PGLog *log2);
|
||||
void apply_pwlc(const std::pair<eversion_t, eversion_t> pwlc,
|
||||
const pg_shard_t &shard,
|
||||
pg_info_t &info,
|
||||
pg_log_t *log)
|
||||
{
|
||||
apply_pwlc(pwlc, shard, info, log, nullptr);
|
||||
}
|
||||
void apply_pwlc(const std::pair<eversion_t, eversion_t> pwlc,
|
||||
const pg_shard_t &shard,
|
||||
pg_info_t &info,
|
||||
PGLog *log = nullptr)
|
||||
{
|
||||
apply_pwlc(pwlc, shard, info, nullptr, log);
|
||||
}
|
||||
void update_peer_info(const pg_shard_t &from, const pg_info_t &oinfo);
|
||||
bool proc_replica_notify(const pg_shard_t &from, const pg_notify_t ¬ify);
|
||||
void remove_down_peer_info(const OSDMapRef &osdmap);
|
||||
@ -1764,11 +1783,12 @@ private:
|
||||
pg_log_t&& olog, pg_shard_t from);
|
||||
|
||||
void proc_primary_info(ObjectStore::Transaction &t, const pg_info_t &info);
|
||||
void consider_adjusting_pwlc(eversion_t last_complete);
|
||||
void consider_rollback_pwlc(eversion_t last_complete);
|
||||
void proc_master_log(ObjectStore::Transaction& t, pg_info_t &oinfo,
|
||||
pg_log_t&& olog, pg_missing_t&& omissing,
|
||||
pg_shard_t from);
|
||||
void proc_replica_log(pg_info_t &oinfo, const pg_log_t &olog,
|
||||
void proc_replica_log(pg_info_t &oinfo, pg_log_t &olog,
|
||||
pg_missing_t&& omissing, pg_shard_t from);
|
||||
|
||||
void calc_min_last_complete_ondisk();
|
||||
|
||||
@ -4738,7 +4738,7 @@ public:
|
||||
std::move(childdups));
|
||||
}
|
||||
|
||||
mempool::osd_pglog::list<pg_log_entry_t> rewind_from_head(eversion_t newhead) {
|
||||
mempool::osd_pglog::list<pg_log_entry_t> rewind_from_head(eversion_t newhead, bool *dirty_log = nullptr) {
|
||||
ceph_assert(newhead >= tail);
|
||||
|
||||
mempool::osd_pglog::list<pg_log_entry_t>::iterator p = log.end();
|
||||
@ -4768,11 +4768,19 @@ public:
|
||||
}
|
||||
head = newhead;
|
||||
|
||||
if (can_rollback_to > newhead)
|
||||
if (can_rollback_to > newhead) {
|
||||
can_rollback_to = newhead;
|
||||
if (dirty_log) {
|
||||
*dirty_log = true;
|
||||
}
|
||||
}
|
||||
|
||||
if (rollback_info_trimmed_to > newhead)
|
||||
if (rollback_info_trimmed_to > newhead) {
|
||||
rollback_info_trimmed_to = newhead;
|
||||
if (dirty_log) {
|
||||
*dirty_log = true;
|
||||
}
|
||||
}
|
||||
|
||||
return divergent;
|
||||
}
|
||||
|
||||
@ -3075,8 +3075,16 @@ int Objecter::_calc_target(op_target_t *t, Connection *con, bool any_change)
|
||||
t->pg_num_pending = pg_num_pending;
|
||||
spg_t spgid(actual_pgid);
|
||||
if (pi->is_erasure()) {
|
||||
// Optimized EC pools need to be careful when calculating the shard
|
||||
// because an OSD may have multiple shards and the primary shard
|
||||
// might not be the first one in the acting set. The lookup
|
||||
// therefoere has to be done in primaryfirst order.
|
||||
std::vector<int> pg_temp = t->acting;
|
||||
if (osdmap->has_pgtemp(actual_pgid)) {
|
||||
pg_temp = osdmap->pgtemp_primaryfirst(*pi, t->acting);
|
||||
}
|
||||
for (uint8_t i = 0; i < t->acting.size(); ++i) {
|
||||
if (t->acting[i] == acting_primary) {
|
||||
if (pg_temp[i] == acting_primary) {
|
||||
spgid.reset_shard(osdmap->pgtemp_undo_primaryfirst(*pi, actual_pgid, shard_id_t(i)));
|
||||
break;
|
||||
}
|
||||
|
||||
@ -158,12 +158,12 @@ public:
|
||||
// for each missing shard.
|
||||
for (auto a : available) {
|
||||
minimum_set.insert(a);
|
||||
if (minimum_set.size() == data_chunk_count) {
|
||||
if (std::cmp_equal(minimum_set.size(), data_chunk_count)) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (minimum_set.size() != data_chunk_count) {
|
||||
if (std::cmp_not_equal(minimum_set.size(), data_chunk_count)) {
|
||||
minimum_set.clear();
|
||||
return -EIO; // Cannot recover.
|
||||
}
|
||||
@ -174,7 +174,6 @@ public:
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
[[deprecated]]
|
||||
int minimum_to_decode(const std::set<int> &want_to_read,
|
||||
const std::set<int> &available,
|
||||
@ -1320,7 +1319,8 @@ void test_decode(unsigned int k, unsigned int m, uint64_t chunk_size, uint64_t o
|
||||
}
|
||||
}
|
||||
|
||||
ASSERT_EQ(0, semap.decode(ec_impl, want, object_size, nullptr, true));
|
||||
semap.add_zero_padding_for_decode(read_request.zeros_for_decode);
|
||||
ASSERT_EQ(0, semap.decode(ec_impl, want, object_size));
|
||||
}
|
||||
|
||||
TEST(ECCommon, decode) {
|
||||
@ -1476,4 +1476,4 @@ TEST(ECCommon, decode7) {
|
||||
acting_set.insert_range(shard_id_t(0), 3);
|
||||
|
||||
test_decode(k, m, chunk_size, object_size, want, acting_set);
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
