--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2020 Red Hat <contact@redhat.com>
+ * Author: Adam C. Emerson
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#include <algorithm>
+#include <cstdint>
+#include <numeric>
+#include <optional>
+#include <string_view>
+
+#include <fmt/format.h>
+
+#include "include/rados/librados.hpp"
+
+#include "include/buffer.h"
+
+#include "common/async/yield_context.h"
+#include "common/random_string.h"
+
+#include "cls/fifo/cls_fifo_types.h"
+#include "cls/fifo/cls_fifo_ops.h"
+
+#include "cls_fifo_legacy.h"
+
+namespace rgw::cls::fifo {
+namespace cb = ceph::buffer;
+namespace fifo = rados::cls::fifo;
+
+using ceph::from_error_code;
+
+inline constexpr auto MAX_RACE_RETRIES = 10;
+
+void create_meta(lr::ObjectWriteOperation* op,
+ std::string_view id,
+ std::optional<fifo::objv> objv,
+ std::optional<std::string_view> oid_prefix,
+ bool exclusive,
+ std::uint64_t max_part_size,
+ std::uint64_t max_entry_size)
+{
+ fifo::op::create_meta cm;
+
+ cm.id = id;
+ cm.version = objv;
+ cm.oid_prefix = oid_prefix;
+ cm.max_part_size = max_part_size;
+ cm.max_entry_size = max_entry_size;
+ cm.exclusive = exclusive;
+
+ cb::list in;
+ encode(cm, in);
+ op->exec(fifo::op::CLASS, fifo::op::CREATE_META, in);
+}
+
+int get_meta(const DoutPrefixProvider *dpp, lr::IoCtx& ioctx, const std::string& oid,
+ std::optional<fifo::objv> objv, fifo::info* info,
+ std::uint32_t* part_header_size,
+ std::uint32_t* part_entry_overhead,
+ uint64_t tid, optional_yield y,
+ bool probe)
+{
+ lr::ObjectReadOperation op;
+ fifo::op::get_meta gm;
+ gm.version = objv;
+ cb::list in;
+ encode(gm, in);
+ cb::list bl;
+
+ op.exec(fifo::op::CLASS, fifo::op::GET_META, in,
+ &bl, nullptr);
+ auto r = rgw_rados_operate(dpp, ioctx, oid, &op, nullptr, y);
+ if (r >= 0) try {
+ fifo::op::get_meta_reply reply;
+ auto iter = bl.cbegin();
+ decode(reply, iter);
+ if (info) *info = std::move(reply.info);
+ if (part_header_size) *part_header_size = reply.part_header_size;
+ if (part_entry_overhead)
+ *part_entry_overhead = reply.part_entry_overhead;
+ } catch (const cb::error& err) {
+ ldpp_dout(dpp, -1)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " decode failed: " << err.what()
+ << " tid=" << tid << dendl;
+ r = from_error_code(err.code());
+ } else if (!(probe && (r == -ENOENT || r == -ENODATA))) {
+ ldpp_dout(dpp, -1)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " fifo::op::GET_META failed r=" << r << " tid=" << tid
+ << dendl;
+ }
+ return r;
+};
+
+namespace {
+void update_meta(lr::ObjectWriteOperation* op, const fifo::objv& objv,
+ const fifo::update& update)
+{
+ fifo::op::update_meta um;
+
+ um.version = objv;
+ um.tail_part_num = update.tail_part_num();
+ um.head_part_num = update.head_part_num();
+ um.min_push_part_num = update.min_push_part_num();
+ um.max_push_part_num = update.max_push_part_num();
+ um.journal_entries_add = std::move(update).journal_entries_add();
+ um.journal_entries_rm = std::move(update).journal_entries_rm();
+
+ cb::list in;
+ encode(um, in);
+ op->exec(fifo::op::CLASS, fifo::op::UPDATE_META, in);
+}
+
+void part_init(lr::ObjectWriteOperation* op, fifo::data_params params)
+{
+ fifo::op::init_part ip;
+
+ ip.params = params;
+
+ cb::list in;
+ encode(ip, in);
+ op->exec(fifo::op::CLASS, fifo::op::INIT_PART, in);
+}
+
+int push_part(const DoutPrefixProvider *dpp, lr::IoCtx& ioctx, const std::string& oid,
+ std::deque<cb::list> data_bufs, std::uint64_t tid,
+ optional_yield y)
+{
+ lr::ObjectWriteOperation op;
+ fifo::op::push_part pp;
+
+ op.assert_exists();
+
+ pp.data_bufs = data_bufs;
+ pp.total_len = 0;
+
+ for (const auto& bl : data_bufs)
+ pp.total_len += bl.length();
+
+ cb::list in;
+ encode(pp, in);
+ auto retval = 0;
+ op.exec(fifo::op::CLASS, fifo::op::PUSH_PART, in, nullptr, &retval);
+ auto r = rgw_rados_operate(dpp, ioctx, oid, &op, y, lr::OPERATION_RETURNVEC);
+ if (r < 0) {
+ ldpp_dout(dpp, -1)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " fifo::op::PUSH_PART failed r=" << r
+ << " tid=" << tid << dendl;
+ return r;
+ }
+ if (retval < 0) {
+ ldpp_dout(dpp, -1)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " error handling response retval=" << retval
+ << " tid=" << tid << dendl;
+ }
+ return retval;
+}
+
+void push_part(lr::IoCtx& ioctx, const std::string& oid,
+ std::deque<cb::list> data_bufs, std::uint64_t tid,
+ lr::AioCompletion* c)
+{
+ lr::ObjectWriteOperation op;
+ fifo::op::push_part pp;
+
+ pp.data_bufs = data_bufs;
+ pp.total_len = 0;
+
+ for (const auto& bl : data_bufs)
+ pp.total_len += bl.length();
+
+ cb::list in;
+ encode(pp, in);
+ op.exec(fifo::op::CLASS, fifo::op::PUSH_PART, in);
+ auto r = ioctx.aio_operate(oid, c, &op, lr::OPERATION_RETURNVEC);
+ ceph_assert(r >= 0);
+}
+
+void trim_part(lr::ObjectWriteOperation* op,
+ std::uint64_t ofs, bool exclusive)
+{
+ fifo::op::trim_part tp;
+
+ tp.ofs = ofs;
+ tp.exclusive = exclusive;
+
+ cb::list in;
+ encode(tp, in);
+ op->exec(fifo::op::CLASS, fifo::op::TRIM_PART, in);
+}
+
+int list_part(const DoutPrefixProvider *dpp, lr::IoCtx& ioctx, const std::string& oid,
+ std::uint64_t ofs, std::uint64_t max_entries,
+ std::vector<fifo::part_list_entry>* entries,
+ bool* more, bool* full_part,
+ std::uint64_t tid, optional_yield y)
+{
+ lr::ObjectReadOperation op;
+ fifo::op::list_part lp;
+
+ lp.ofs = ofs;
+ lp.max_entries = max_entries;
+
+ cb::list in;
+ encode(lp, in);
+ cb::list bl;
+ op.exec(fifo::op::CLASS, fifo::op::LIST_PART, in, &bl, nullptr);
+ auto r = rgw_rados_operate(dpp, ioctx, oid, &op, nullptr, y);
+ if (r >= 0) try {
+ fifo::op::list_part_reply reply;
+ auto iter = bl.cbegin();
+ decode(reply, iter);
+ if (entries) *entries = std::move(reply.entries);
+ if (more) *more = reply.more;
+ if (full_part) *full_part = reply.full_part;
+ } catch (const cb::error& err) {
+ ldpp_dout(dpp, -1)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " decode failed: " << err.what()
+ << " tid=" << tid << dendl;
+ r = from_error_code(err.code());
+ } else if (r != -ENOENT) {
+ ldpp_dout(dpp, -1)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " fifo::op::LIST_PART failed r=" << r << " tid=" << tid
+ << dendl;
+ }
+ return r;
+}
+
+struct list_entry_completion : public lr::ObjectOperationCompletion {
+ CephContext* cct;
+ int* r_out;
+ std::vector<fifo::part_list_entry>* entries;
+ bool* more;
+ bool* full_part;
+ std::uint64_t tid;
+
+ list_entry_completion(CephContext* cct, int* r_out, std::vector<fifo::part_list_entry>* entries,
+ bool* more, bool* full_part, std::uint64_t tid)
+ : cct(cct), r_out(r_out), entries(entries), more(more),
+ full_part(full_part), tid(tid) {}
+ virtual ~list_entry_completion() = default;
+ void handle_completion(int r, bufferlist& bl) override {
+ if (r >= 0) try {
+ fifo::op::list_part_reply reply;
+ auto iter = bl.cbegin();
+ decode(reply, iter);
+ if (entries) *entries = std::move(reply.entries);
+ if (more) *more = reply.more;
+ if (full_part) *full_part = reply.full_part;
+ } catch (const cb::error& err) {
+ lderr(cct)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " decode failed: " << err.what()
+ << " tid=" << tid << dendl;
+ r = from_error_code(err.code());
+ } else if (r < 0) {
+ lderr(cct)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " fifo::op::LIST_PART failed r=" << r << " tid=" << tid
+ << dendl;
+ }
+ if (r_out) *r_out = r;
+ }
+};
+
+lr::ObjectReadOperation list_part(CephContext* cct,
+ std::uint64_t ofs,
+ std::uint64_t max_entries,
+ int* r_out,
+ std::vector<fifo::part_list_entry>* entries,
+ bool* more, bool* full_part,
+ std::uint64_t tid)
+{
+ lr::ObjectReadOperation op;
+ fifo::op::list_part lp;
+
+ lp.ofs = ofs;
+ lp.max_entries = max_entries;
+
+ cb::list in;
+ encode(lp, in);
+ op.exec(fifo::op::CLASS, fifo::op::LIST_PART, in,
+ new list_entry_completion(cct, r_out, entries, more, full_part,
+ tid));
+ return op;
+}
+
+int get_part_info(const DoutPrefixProvider *dpp, lr::IoCtx& ioctx, const std::string& oid,
+ fifo::part_header* header,
+ std::uint64_t tid, optional_yield y)
+{
+ lr::ObjectReadOperation op;
+ fifo::op::get_part_info gpi;
+
+ cb::list in;
+ cb::list bl;
+ encode(gpi, in);
+ op.exec(fifo::op::CLASS, fifo::op::GET_PART_INFO, in, &bl, nullptr);
+ auto r = rgw_rados_operate(dpp, ioctx, oid, &op, nullptr, y);
+ if (r >= 0) try {
+ fifo::op::get_part_info_reply reply;
+ auto iter = bl.cbegin();
+ decode(reply, iter);
+ if (header) *header = std::move(reply.header);
+ } catch (const cb::error& err) {
+ ldpp_dout(dpp, -1)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " decode failed: " << err.what()
+ << " tid=" << tid << dendl;
+ r = from_error_code(err.code());
+ } else {
+ ldpp_dout(dpp, -1)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " fifo::op::GET_PART_INFO failed r=" << r << " tid=" << tid
+ << dendl;
+ }
+ return r;
+}
+
+struct partinfo_completion : public lr::ObjectOperationCompletion {
+ CephContext* cct;
+ int* rp;
+ fifo::part_header* h;
+ std::uint64_t tid;
+ partinfo_completion(CephContext* cct, int* rp, fifo::part_header* h,
+ std::uint64_t tid) :
+ cct(cct), rp(rp), h(h), tid(tid) {
+ }
+ virtual ~partinfo_completion() = default;
+ void handle_completion(int r, bufferlist& bl) override {
+ if (r >= 0) try {
+ fifo::op::get_part_info_reply reply;
+ auto iter = bl.cbegin();
+ decode(reply, iter);
+ if (h) *h = std::move(reply.header);
+ } catch (const cb::error& err) {
+ r = from_error_code(err.code());
+ lderr(cct) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " decode failed: " << err.what()
+ << " tid=" << tid << dendl;
+ } else {
+ lderr(cct) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " fifo::op::GET_PART_INFO failed r=" << r << " tid=" << tid
+ << dendl;
+ }
+ if (rp) {
+ *rp = r;
+ }
+ }
+};
+
+lr::ObjectReadOperation get_part_info(CephContext* cct,
+ fifo::part_header* header,
+ std::uint64_t tid, int* r = 0)
+{
+ lr::ObjectReadOperation op;
+ fifo::op::get_part_info gpi;
+
+ cb::list in;
+ cb::list bl;
+ encode(gpi, in);
+ op.exec(fifo::op::CLASS, fifo::op::GET_PART_INFO, in,
+ new partinfo_completion(cct, r, header, tid));
+ return op;
+}
+}
+
+std::optional<marker> FIFO::to_marker(std::string_view s)
+{
+ marker m;
+ if (s.empty()) {
+ m.num = info.tail_part_num;
+ m.ofs = 0;
+ return m;
+ }
+
+ auto pos = s.find(':');
+ if (pos == s.npos) {
+ return std::nullopt;
+ }
+
+ auto num = s.substr(0, pos);
+ auto ofs = s.substr(pos + 1);
+
+ auto n = ceph::parse<decltype(m.num)>(num);
+ if (!n) {
+ return std::nullopt;
+ }
+ m.num = *n;
+ auto o = ceph::parse<decltype(m.ofs)>(ofs);
+ if (!o) {
+ return std::nullopt;
+ }
+ m.ofs = *o;
+ return m;
+}
+
+int FIFO::apply_update(const DoutPrefixProvider *dpp,
+ fifo::info* info,
+ const fifo::objv& objv,
+ const fifo::update& update,
+ std::uint64_t tid)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ std::unique_lock l(m);
+ if (objv != info->version) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " version mismatch, canceling: tid=" << tid << dendl;
+ return -ECANCELED;
+ }
+
+ info->apply_update(update);
+ return {};
+}
+
+int FIFO::_update_meta(const DoutPrefixProvider *dpp, const fifo::update& update,
+ fifo::objv version, bool* pcanceled,
+ std::uint64_t tid, optional_yield y)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ lr::ObjectWriteOperation op;
+ bool canceled = false;
+ update_meta(&op, version, update);
+ auto r = rgw_rados_operate(dpp, ioctx, oid, &op, y);
+ if (r >= 0 || r == -ECANCELED) {
+ canceled = (r == -ECANCELED);
+ if (!canceled) {
+ r = apply_update(dpp, &info, version, update, tid);
+ if (r < 0) canceled = true;
+ }
+ if (canceled) {
+ r = read_meta(dpp, tid, y);
+ canceled = r < 0 ? false : true;
+ }
+ }
+ if (pcanceled) *pcanceled = canceled;
+ if (canceled) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled: tid=" << tid << dendl;
+ }
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " returning error: r=" << r << " tid=" << tid << dendl;
+ }
+ return r;
+}
+
+struct Updater : public Completion<Updater> {
+ FIFO* fifo;
+ fifo::update update;
+ fifo::objv version;
+ bool reread = false;
+ bool* pcanceled = nullptr;
+ std::uint64_t tid;
+ Updater(const DoutPrefixProvider *dpp, FIFO* fifo, lr::AioCompletion* super,
+ const fifo::update& update, fifo::objv version,
+ bool* pcanceled, std::uint64_t tid)
+ : Completion(dpp, super), fifo(fifo), update(update), version(version),
+ pcanceled(pcanceled) {}
+
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ if (reread)
+ handle_reread(dpp, std::move(p), r);
+ else
+ handle_update(dpp, std::move(p), r);
+ }
+
+ void handle_update(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " handling async update_meta: tid="
+ << tid << dendl;
+ if (r < 0 && r != -ECANCELED) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " update failed: r=" << r << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+ bool canceled = (r == -ECANCELED);
+ if (!canceled) {
+ int r = fifo->apply_update(dpp, &fifo->info, version, update, tid);
+ if (r < 0) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " update failed, marking canceled: r=" << r
+ << " tid=" << tid << dendl;
+ canceled = true;
+ }
+ }
+ if (canceled) {
+ reread = true;
+ fifo->read_meta(dpp, tid, call(std::move(p)));
+ return;
+ }
+ if (pcanceled)
+ *pcanceled = false;
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " completing: tid=" << tid << dendl;
+ complete(std::move(p), 0);
+ }
+
+ void handle_reread(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " handling async read_meta: tid="
+ << tid << dendl;
+ if (r < 0 && pcanceled) {
+ *pcanceled = false;
+ } else if (r >= 0 && pcanceled) {
+ *pcanceled = true;
+ }
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " failed dispatching read_meta: r=" << r << " tid="
+ << tid << dendl;
+ } else {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " completing: tid=" << tid << dendl;
+ }
+ complete(std::move(p), r);
+ }
+};
+
+void FIFO::_update_meta(const DoutPrefixProvider *dpp, const fifo::update& update,
+ fifo::objv version, bool* pcanceled,
+ std::uint64_t tid, lr::AioCompletion* c)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ lr::ObjectWriteOperation op;
+ update_meta(&op, info.version, update);
+ auto updater = std::make_unique<Updater>(dpp, this, c, update, version, pcanceled,
+ tid);
+ auto r = ioctx.aio_operate(oid, Updater::call(std::move(updater)), &op);
+ assert(r >= 0);
+}
+
+int FIFO::create_part(const DoutPrefixProvider *dpp, int64_t part_num, std::uint64_t tid,
+ optional_yield y)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ lr::ObjectWriteOperation op;
+ op.create(false); /* We don't need exclusivity, part_init ensures
+ we're creating from the same journal entry. */
+ std::unique_lock l(m);
+ part_init(&op, info.params);
+ auto oid = info.part_oid(part_num);
+ l.unlock();
+ auto r = rgw_rados_operate(dpp, ioctx, oid, &op, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " part_init failed: r=" << r << " tid="
+ << tid << dendl;
+ }
+ return r;
+}
+
+int FIFO::remove_part(const DoutPrefixProvider *dpp, int64_t part_num, std::uint64_t tid,
+ optional_yield y)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ lr::ObjectWriteOperation op;
+ op.remove();
+ std::unique_lock l(m);
+ auto oid = info.part_oid(part_num);
+ l.unlock();
+ auto r = rgw_rados_operate(dpp, ioctx, oid, &op, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " remove failed: r=" << r << " tid="
+ << tid << dendl;
+ }
+ return r;
+}
+
+int FIFO::process_journal(const DoutPrefixProvider *dpp, std::uint64_t tid, optional_yield y)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ std::vector<fifo::journal_entry> processed;
+
+ std::unique_lock l(m);
+ auto tmpjournal = info.journal;
+ auto new_tail = info.tail_part_num;
+ auto new_head = info.head_part_num;
+ auto new_max = info.max_push_part_num;
+ l.unlock();
+
+ int r = 0;
+ for (auto& entry : tmpjournal) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " processing entry: entry=" << entry << " tid=" << tid
+ << dendl;
+ switch (entry.op) {
+ using enum fifo::journal_entry::Op;
+ case create:
+ r = create_part(dpp, entry.part_num, tid, y);
+ if (entry.part_num > new_max) {
+ new_max = entry.part_num;
+ }
+ break;
+ case set_head:
+ r = 0;
+ if (entry.part_num > new_head) {
+ new_head = entry.part_num;
+ }
+ break;
+ case remove:
+ r = remove_part(dpp, entry.part_num, tid, y);
+ if (r == -ENOENT) r = 0;
+ if (entry.part_num >= new_tail) {
+ new_tail = entry.part_num + 1;
+ }
+ break;
+ default:
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " unknown journaled op: entry=" << entry << " tid="
+ << tid << dendl;
+ return -EIO;
+ }
+
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " processing entry failed: entry=" << entry
+ << " r=" << r << " tid=" << tid << dendl;
+ return -r;
+ }
+
+ processed.push_back(std::move(entry));
+ }
+
+ // Postprocess
+ bool canceled = true;
+
+ for (auto i = 0; canceled && i < MAX_RACE_RETRIES; ++i) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " postprocessing: i=" << i << " tid=" << tid << dendl;
+
+ std::optional<int64_t> tail_part_num;
+ std::optional<int64_t> head_part_num;
+ std::optional<int64_t> max_part_num;
+
+ std::unique_lock l(m);
+ auto objv = info.version;
+ if (new_tail > tail_part_num) tail_part_num = new_tail;
+ if (new_head > info.head_part_num) head_part_num = new_head;
+ if (new_max > info.max_push_part_num) max_part_num = new_max;
+ l.unlock();
+
+ if (processed.empty() &&
+ !tail_part_num &&
+ !max_part_num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " nothing to update any more: i=" << i << " tid="
+ << tid << dendl;
+ canceled = false;
+ break;
+ }
+ auto u = fifo::update().tail_part_num(tail_part_num)
+ .head_part_num(head_part_num).max_push_part_num(max_part_num)
+ .journal_entries_rm(processed);
+ r = _update_meta(dpp, u, objv, &canceled, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _update_meta failed: update=" << u
+ << " r=" << r << " tid=" << tid << dendl;
+ break;
+ }
+
+ if (canceled) {
+ std::vector<fifo::journal_entry> new_processed;
+ std::unique_lock l(m);
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " update canceled, retrying: i=" << i << " tid="
+ << tid << dendl;
+ for (auto& e : processed) {
+ if (info.journal.contains(e)) {
+ new_processed.push_back(e);
+ }
+ }
+ processed = std::move(new_processed);
+ }
+ }
+ if (r == 0 && canceled) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled too many times, giving up: tid=" << tid << dendl;
+ r = -ECANCELED;
+ }
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " failed, r=: " << r << " tid=" << tid << dendl;
+ }
+ return r;
+}
+
+int FIFO::_prepare_new_part(const DoutPrefixProvider *dpp,
+ std::int64_t new_part_num, bool is_head,
+ std::uint64_t tid, optional_yield y)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ std::unique_lock l(m);
+ using enum fifo::journal_entry::Op;
+ std::vector<fifo::journal_entry> jentries{{ create, new_part_num }};
+ if (info.journal.contains({create, new_part_num}) &&
+ (!is_head || info.journal.contains({set_head, new_part_num}))) {
+ l.unlock();
+ ldpp_dout(dpp, 5) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " new part journaled, but not processed: tid="
+ << tid << dendl;
+ auto r = process_journal(dpp, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " process_journal failed: r=" << r << " tid=" << tid << dendl;
+ }
+ return r;
+ }
+ auto version = info.version;
+
+ if (is_head) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " needs new head: tid=" << tid << dendl;
+ jentries.push_back({ set_head, new_part_num });
+ }
+ l.unlock();
+
+ int r = 0;
+ bool canceled = true;
+ for (auto i = 0; canceled && i < MAX_RACE_RETRIES; ++i) {
+ canceled = false;
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " updating metadata: i=" << i << " tid=" << tid << dendl;
+ auto u = fifo::update{}.journal_entries_add(jentries);
+ r = _update_meta(dpp, u, version, &canceled, tid, y);
+ if (r >= 0 && canceled) {
+ std::unique_lock l(m);
+ version = info.version;
+ auto found = (info.journal.contains({create, new_part_num}) ||
+ info.journal.contains({set_head, new_part_num}));
+ if ((info.max_push_part_num >= new_part_num &&
+ info.head_part_num >= new_part_num)) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced, but journaled and processed: i=" << i
+ << " tid=" << tid << dendl;
+ return 0;
+ }
+ if (found) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced, journaled but not processed: i=" << i
+ << " tid=" << tid << dendl;
+ canceled = false;
+ }
+ l.unlock();
+ }
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _update_meta failed: update=" << u << " r=" << r
+ << " tid=" << tid << dendl;
+ return r;
+ }
+ }
+ if (canceled) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled too many times, giving up: tid=" << tid << dendl;
+ return -ECANCELED;
+ }
+ r = process_journal(dpp, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " process_journal failed: r=" << r << " tid=" << tid << dendl;
+ }
+ return r;
+}
+
+int FIFO::_prepare_new_head(const DoutPrefixProvider *dpp,
+ std::int64_t new_head_part_num,
+ std::uint64_t tid, optional_yield y)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ std::unique_lock l(m);
+ auto max_push_part_num = info.max_push_part_num;
+ auto version = info.version;
+ l.unlock();
+
+ int r = 0;
+ if (max_push_part_num < new_head_part_num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " need new part: tid=" << tid << dendl;
+ r = _prepare_new_part(dpp, new_head_part_num, true, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _prepare_new_part failed: r=" << r
+ << " tid=" << tid << dendl;
+ return r;
+ }
+ std::unique_lock l(m);
+ if (info.max_push_part_num < new_head_part_num) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " inconsistency, push part less than head part: "
+ << " tid=" << tid << dendl;
+ return -EIO;
+ }
+ l.unlock();
+ return 0;
+ }
+
+ using enum fifo::journal_entry::Op;
+ fifo::journal_entry jentry;
+ jentry.op = set_head;
+ jentry.part_num = new_head_part_num;
+
+ r = 0;
+ bool canceled = true;
+ for (auto i = 0; canceled && i < MAX_RACE_RETRIES; ++i) {
+ canceled = false;
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " updating metadata: i=" << i << " tid=" << tid << dendl;
+ auto u = fifo::update{}.journal_entries_add({{ jentry }});
+ r = _update_meta(dpp, u, version, &canceled, tid, y);
+ if (r >= 0 && canceled) {
+ std::unique_lock l(m);
+ auto found = (info.journal.contains({create, new_head_part_num}) ||
+ info.journal.contains({set_head, new_head_part_num}));
+ version = info.version;
+ if ((info.head_part_num >= new_head_part_num)) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced, but journaled and processed: i=" << i
+ << " tid=" << tid << dendl;
+ return 0;
+ }
+ if (found) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced, journaled but not processed: i=" << i
+ << " tid=" << tid << dendl;
+ canceled = false;
+ }
+ l.unlock();
+ }
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _update_meta failed: update=" << u << " r=" << r
+ << " tid=" << tid << dendl;
+ return r;
+ }
+ }
+ if (canceled) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled too many times, giving up: tid=" << tid << dendl;
+ return -ECANCELED;
+ }
+ r = process_journal(dpp, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " process_journal failed: r=" << r << " tid=" << tid << dendl;
+ }
+ return r;
+}
+
+struct NewPartPreparer : public Completion<NewPartPreparer> {
+ FIFO* f;
+ std::vector<fifo::journal_entry> jentries;
+ int i = 0;
+ std::int64_t new_part_num;
+ bool canceled = false;
+ uint64_t tid;
+
+ NewPartPreparer(const DoutPrefixProvider *dpp, FIFO* f, lr::AioCompletion* super,
+ std::vector<fifo::journal_entry> jentries,
+ std::int64_t new_part_num,
+ std::uint64_t tid)
+ : Completion(dpp, super), f(f), jentries(std::move(jentries)),
+ new_part_num(new_part_num), tid(tid) {}
+
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _update_meta failed: r=" << r
+ << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+
+ if (canceled) {
+ using enum fifo::journal_entry::Op;
+ std::unique_lock l(f->m);
+ auto found = (f->info.journal.contains({create, new_part_num}) ||
+ f->info.journal.contains({set_head, new_part_num}));
+ auto max_push_part_num = f->info.max_push_part_num;
+ auto head_part_num = f->info.head_part_num;
+ auto version = f->info.version;
+ l.unlock();
+ if ((max_push_part_num >= new_part_num &&
+ head_part_num >= new_part_num)) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced, but journaled and processed: i=" << i
+ << " tid=" << tid << dendl;
+ complete(std::move(p), 0);
+ return;
+ }
+ if (i >= MAX_RACE_RETRIES) {
+ complete(std::move(p), -ECANCELED);
+ return;
+ }
+ if (!found) {
+ ++i;
+ f->_update_meta(dpp, fifo::update{}
+ .journal_entries_add(jentries),
+ version, &canceled, tid, call(std::move(p)));
+ return;
+ } else {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced, journaled but not processed: i=" << i
+ << " tid=" << tid << dendl;
+ canceled = false;
+ }
+ // Fall through. We still need to process the journal.
+ }
+ f->process_journal(dpp, tid, super());
+ return;
+ }
+};
+
+void FIFO::_prepare_new_part(const DoutPrefixProvider *dpp, std::int64_t new_part_num,
+ bool is_head, std::uint64_t tid, lr::AioCompletion* c)
+{
+ std::unique_lock l(m);
+ using enum fifo::journal_entry::Op;
+ std::vector<fifo::journal_entry> jentries{{create, new_part_num}};
+ if (info.journal.contains({create, new_part_num}) &&
+ (!is_head || info.journal.contains({set_head, new_part_num}))) {
+ l.unlock();
+ ldpp_dout(dpp, 5) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " new part journaled, but not processed: tid="
+ << tid << dendl;
+ process_journal(dpp, tid, c);
+ return;
+ }
+ auto version = info.version;
+
+ if (is_head) {
+ jentries.push_back({ set_head, new_part_num });
+ }
+ l.unlock();
+
+ auto n = std::make_unique<NewPartPreparer>(dpp, this, c, jentries,
+ new_part_num, tid);
+ auto np = n.get();
+ _update_meta(dpp, fifo::update{}.journal_entries_add(jentries), version,
+ &np->canceled, tid, NewPartPreparer::call(std::move(n)));
+}
+
+struct NewHeadPreparer : public Completion<NewHeadPreparer> {
+ FIFO* f;
+ int i = 0;
+ bool newpart;
+ std::int64_t new_head_part_num;
+ bool canceled = false;
+ std::uint64_t tid;
+
+ NewHeadPreparer(const DoutPrefixProvider *dpp, FIFO* f, lr::AioCompletion* super,
+ bool newpart, std::int64_t new_head_part_num,
+ std::uint64_t tid)
+ : Completion(dpp, super), f(f), newpart(newpart),
+ new_head_part_num(new_head_part_num), tid(tid) {}
+
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ if (newpart)
+ handle_newpart(std::move(p), r);
+ else
+ handle_update(dpp, std::move(p), r);
+ }
+
+ void handle_newpart(Ptr&& p, int r) {
+ if (r < 0) {
+ lderr(f->cct) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _prepare_new_part failed: r=" << r
+ << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+ std::unique_lock l(f->m);
+ if (f->info.max_push_part_num < new_head_part_num) {
+ l.unlock();
+ lderr(f->cct) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _prepare_new_part failed: r=" << r
+ << " tid=" << tid << dendl;
+ complete(std::move(p), -EIO);
+ } else {
+ l.unlock();
+ complete(std::move(p), 0);
+ }
+ }
+
+ void handle_update(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _update_meta failed: r=" << r
+ << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+
+ if (canceled) {
+ using enum fifo::journal_entry::Op;
+ std::unique_lock l(f->m);
+ auto found = (f->info.journal.contains({create, new_head_part_num }) ||
+ f->info.journal.contains({set_head, new_head_part_num }));
+ auto head_part_num = f->info.head_part_num;
+ auto version = f->info.version;
+
+ l.unlock();
+ if ((head_part_num >= new_head_part_num)) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced, but journaled and processed: i=" << i
+ << " tid=" << tid << dendl;
+ complete(std::move(p), 0);
+ return;
+ }
+ if (i >= MAX_RACE_RETRIES) {
+ complete(std::move(p), -ECANCELED);
+ return;
+ }
+ if (!found) {
+ ++i;
+ fifo::journal_entry jentry;
+ jentry.op = set_head;
+ jentry.part_num = new_head_part_num;
+ f->_update_meta(dpp, fifo::update{}
+ .journal_entries_add({{jentry}}),
+ version, &canceled, tid, call(std::move(p)));
+ return;
+ } else {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced, journaled but not processed: i=" << i
+ << " tid=" << tid << dendl;
+ canceled = false;
+ }
+ // Fall through. We still need to process the journal.
+ }
+ f->process_journal(dpp, tid, super());
+ return;
+ }
+};
+
+void FIFO::_prepare_new_head(const DoutPrefixProvider *dpp, std::int64_t new_head_part_num,
+ std::uint64_t tid, lr::AioCompletion* c)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ std::unique_lock l(m);
+ auto max_push_part_num = info.max_push_part_num;
+ auto version = info.version;
+ l.unlock();
+
+ if (max_push_part_num < new_head_part_num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " need new part: tid=" << tid << dendl;
+ auto n = std::make_unique<NewHeadPreparer>(dpp, this, c, true, new_head_part_num,
+ tid);
+ _prepare_new_part(dpp, new_head_part_num, true, tid,
+ NewHeadPreparer::call(std::move(n)));
+ } else {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " updating head: tid=" << tid << dendl;
+ auto n = std::make_unique<NewHeadPreparer>(dpp, this, c, false, new_head_part_num,
+ tid);
+ auto np = n.get();
+ using enum fifo::journal_entry::Op;
+ fifo::journal_entry jentry;
+ jentry.op = set_head;
+ jentry.part_num = new_head_part_num;
+ _update_meta(dpp, fifo::update{}.journal_entries_add({{jentry}}), version,
+ &np->canceled, tid, NewHeadPreparer::call(std::move(n)));
+ }
+}
+
+int FIFO::push_entries(const DoutPrefixProvider *dpp, const std::deque<cb::list>& data_bufs,
+ std::uint64_t tid, optional_yield y)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ std::unique_lock l(m);
+ auto head_part_num = info.head_part_num;
+ const auto part_oid = info.part_oid(head_part_num);
+ l.unlock();
+
+ auto r = push_part(dpp, ioctx, part_oid, data_bufs, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " push_part failed: r=" << r << " tid=" << tid << dendl;
+ }
+ return r;
+}
+
+void FIFO::push_entries(const std::deque<cb::list>& data_bufs,
+ std::uint64_t tid, lr::AioCompletion* c)
+{
+ std::unique_lock l(m);
+ auto head_part_num = info.head_part_num;
+ const auto part_oid = info.part_oid(head_part_num);
+ l.unlock();
+
+ push_part(ioctx, part_oid, data_bufs, tid, c);
+}
+
+int FIFO::trim_part(const DoutPrefixProvider *dpp, int64_t part_num, uint64_t ofs,
+ bool exclusive, std::uint64_t tid,
+ optional_yield y)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ lr::ObjectWriteOperation op;
+ std::unique_lock l(m);
+ const auto part_oid = info.part_oid(part_num);
+ l.unlock();
+ rgw::cls::fifo::trim_part(&op, ofs, exclusive);
+ auto r = rgw_rados_operate(dpp, ioctx, part_oid, &op, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " trim_part failed: r=" << r << " tid=" << tid << dendl;
+ }
+ return 0;
+}
+
+void FIFO::trim_part(const DoutPrefixProvider *dpp, int64_t part_num, uint64_t ofs,
+ bool exclusive, std::uint64_t tid,
+ lr::AioCompletion* c)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ lr::ObjectWriteOperation op;
+ std::unique_lock l(m);
+ const auto part_oid = info.part_oid(part_num);
+ l.unlock();
+ rgw::cls::fifo::trim_part(&op, ofs, exclusive);
+ auto r = ioctx.aio_operate(part_oid, c, &op);
+ ceph_assert(r >= 0);
+}
+
+int FIFO::open(const DoutPrefixProvider *dpp, lr::IoCtx ioctx, std::string oid, std::unique_ptr<FIFO>* fifo,
+ optional_yield y, std::optional<fifo::objv> objv,
+ bool probe)
+{
+ ldpp_dout(dpp, 20)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering" << dendl;
+ fifo::info info;
+ std::uint32_t size;
+ std::uint32_t over;
+ int r = get_meta(dpp, ioctx, std::move(oid), objv, &info, &size, &over, 0, y,
+ probe);
+ if (r < 0) {
+ if (!(probe && (r == -ENOENT || r == -ENODATA))) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " get_meta failed: r=" << r << dendl;
+ }
+ return r;
+ }
+ std::unique_ptr<FIFO> f(new FIFO(std::move(ioctx), oid));
+ f->info = info;
+ f->part_header_size = size;
+ f->part_entry_overhead = over;
+ // If there are journal entries, process them, in case
+ // someone crashed mid-transaction.
+ if (!info.journal.empty()) {
+ ldpp_dout(dpp, 20)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " processing leftover journal" << dendl;
+ r = f->process_journal(dpp, 0, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " process_journal failed: r=" << r << dendl;
+ return r;
+ }
+ }
+ *fifo = std::move(f);
+ return 0;
+}
+
+int FIFO::create(const DoutPrefixProvider *dpp, lr::IoCtx ioctx, std::string oid, std::unique_ptr<FIFO>* fifo,
+ optional_yield y, std::optional<fifo::objv> objv,
+ std::optional<std::string_view> oid_prefix,
+ bool exclusive, std::uint64_t max_part_size,
+ std::uint64_t max_entry_size)
+{
+ ldpp_dout(dpp, 20)
+ << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering" << dendl;
+ lr::ObjectWriteOperation op;
+ create_meta(&op, oid, objv, oid_prefix, exclusive, max_part_size,
+ max_entry_size);
+ auto r = rgw_rados_operate(dpp, ioctx, oid, &op, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " create_meta failed: r=" << r << dendl;
+ return r;
+ }
+ r = open(dpp, std::move(ioctx), std::move(oid), fifo, y, objv);
+ return r;
+}
+
+int FIFO::read_meta(const DoutPrefixProvider *dpp, std::uint64_t tid, optional_yield y) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ fifo::info _info;
+ std::uint32_t _phs;
+ std::uint32_t _peo;
+
+ auto r = get_meta(dpp, ioctx, oid, std::nullopt, &_info, &_phs, &_peo, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " get_meta failed: r=" << r << " tid=" << tid << dendl;
+ return r;
+ }
+ std::unique_lock l(m);
+ // We have a newer version already!
+ if (_info.version.same_or_later(this->info.version)) {
+ info = std::move(_info);
+ part_header_size = _phs;
+ part_entry_overhead = _peo;
+ }
+ return 0;
+}
+
+int FIFO::read_meta(const DoutPrefixProvider *dpp, optional_yield y) {
+ std::unique_lock l(m);
+ auto tid = ++next_tid;
+ l.unlock();
+ return read_meta(dpp, tid, y);
+}
+
+struct Reader : public Completion<Reader> {
+ FIFO* fifo;
+ cb::list bl;
+ std::uint64_t tid;
+ Reader(const DoutPrefixProvider *dpp, FIFO* fifo, lr::AioCompletion* super, std::uint64_t tid)
+ : Completion(dpp, super), fifo(fifo), tid(tid) {}
+
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ if (r >= 0) try {
+ fifo::op::get_meta_reply reply;
+ auto iter = bl.cbegin();
+ decode(reply, iter);
+ std::unique_lock l(fifo->m);
+ if (reply.info.version.same_or_later(fifo->info.version)) {
+ fifo->info = std::move(reply.info);
+ fifo->part_header_size = reply.part_header_size;
+ fifo->part_entry_overhead = reply.part_entry_overhead;
+ }
+ } catch (const cb::error& err) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " failed to decode response err=" << err.what()
+ << " tid=" << tid << dendl;
+ r = from_error_code(err.code());
+ } else {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " read_meta failed r=" << r
+ << " tid=" << tid << dendl;
+ }
+ complete(std::move(p), r);
+ }
+};
+
+void FIFO::read_meta(const DoutPrefixProvider *dpp, std::uint64_t tid, lr::AioCompletion* c)
+{
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ lr::ObjectReadOperation op;
+ fifo::op::get_meta gm;
+ cb::list in;
+ encode(gm, in);
+ auto reader = std::make_unique<Reader>(dpp, this, c, tid);
+ auto rp = reader.get();
+ auto r = ioctx.aio_exec(oid, Reader::call(std::move(reader)), fifo::op::CLASS,
+ fifo::op::GET_META, in, &rp->bl);
+ assert(r >= 0);
+}
+
+const fifo::info& FIFO::meta() const {
+ return info;
+}
+
+std::pair<std::uint32_t, std::uint32_t> FIFO::get_part_layout_info() const {
+ return {part_header_size, part_entry_overhead};
+}
+
+int FIFO::push(const DoutPrefixProvider *dpp, const cb::list& bl, optional_yield y) {
+ return push(dpp, std::vector{ bl }, y);
+}
+
+void FIFO::push(const DoutPrefixProvider *dpp, const cb::list& bl, lr::AioCompletion* c) {
+ push(dpp, std::vector{ bl }, c);
+}
+
+int FIFO::push(const DoutPrefixProvider *dpp, const std::vector<cb::list>& data_bufs, optional_yield y)
+{
+ std::unique_lock l(m);
+ auto tid = ++next_tid;
+ auto max_entry_size = info.params.max_entry_size;
+ auto need_new_head = info.need_new_head();
+ auto head_part_num = info.head_part_num;
+ l.unlock();
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ if (data_bufs.empty()) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " empty push, returning success tid=" << tid << dendl;
+ return 0;
+ }
+
+ // Validate sizes
+ for (const auto& bl : data_bufs) {
+ if (bl.length() > max_entry_size) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entry bigger than max_entry_size tid=" << tid << dendl;
+ return -E2BIG;
+ }
+ }
+
+ int r = 0;
+ if (need_new_head) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " need new head tid=" << tid << dendl;
+ r = _prepare_new_head(dpp, head_part_num + 1, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _prepare_new_head failed: r=" << r
+ << " tid=" << tid << dendl;
+ return r;
+ }
+ }
+
+ std::deque<cb::list> remaining(data_bufs.begin(), data_bufs.end());
+ std::deque<cb::list> batch;
+
+ uint64_t batch_len = 0;
+ auto retries = 0;
+ bool canceled = true;
+ while ((!remaining.empty() || !batch.empty()) &&
+ (retries <= MAX_RACE_RETRIES)) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " preparing push: remaining=" << remaining.size()
+ << " batch=" << batch.size() << " retries=" << retries
+ << " tid=" << tid << dendl;
+ std::unique_lock l(m);
+ head_part_num = info.head_part_num;
+ auto max_part_size = info.params.max_part_size;
+ auto overhead = part_entry_overhead;
+ l.unlock();
+
+ while (!remaining.empty() &&
+ (remaining.front().length() + batch_len <= max_part_size)) {
+ /* We can send entries with data_len up to max_entry_size,
+ however, we want to also account the overhead when
+ dealing with multiple entries. Previous check doesn't
+ account for overhead on purpose. */
+ batch_len += remaining.front().length() + overhead;
+ batch.push_back(std::move(remaining.front()));
+ remaining.pop_front();
+ }
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " prepared push: remaining=" << remaining.size()
+ << " batch=" << batch.size() << " retries=" << retries
+ << " batch_len=" << batch_len
+ << " tid=" << tid << dendl;
+
+ auto r = push_entries(dpp, batch, tid, y);
+ if (r == -ERANGE) {
+ canceled = true;
+ ++retries;
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " need new head tid=" << tid << dendl;
+ r = _prepare_new_head(dpp, head_part_num + 1, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " prepare_new_head failed: r=" << r
+ << " tid=" << tid << dendl;
+ return r;
+ }
+ r = 0;
+ continue;
+ }
+ if (r == -ENOENT) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " racing client trimmed part, rereading metadata "
+ << "tid=" << tid << dendl;
+ canceled = true;
+ ++retries;
+ r = read_meta(dpp, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " read_meta failed: r=" << r
+ << " tid=" << tid << dendl;
+ return r;
+ }
+ r = 0;
+ continue;
+ }
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " push_entries failed: r=" << r
+ << " tid=" << tid << dendl;
+ return r;
+ }
+ // Made forward progress!
+ canceled = false;
+ retries = 0;
+ batch_len = 0;
+ if (r == ssize(batch)) {
+ batch.clear();
+ } else {
+ batch.erase(batch.begin(), batch.begin() + r);
+ for (const auto& b : batch) {
+ batch_len += b.length() + part_entry_overhead;
+ }
+ }
+ }
+ if (canceled) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled too many times, giving up: tid=" << tid << dendl;
+ return -ECANCELED;
+ }
+ return 0;
+}
+
+struct Pusher : public Completion<Pusher> {
+ FIFO* f;
+ std::deque<cb::list> remaining;
+ std::deque<cb::list> batch;
+ int i = 0;
+ std::int64_t head_part_num;
+ std::uint64_t tid;
+ enum { pushing, new_heading, meta_reading } state = pushing;
+
+ void prep_then_push(const DoutPrefixProvider *dpp, Ptr&& p, const unsigned successes) {
+ std::unique_lock l(f->m);
+ auto max_part_size = f->info.params.max_part_size;
+ auto part_entry_overhead = f->part_entry_overhead;
+ head_part_num = f->info.head_part_num;
+ l.unlock();
+
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " preparing push: remaining=" << remaining.size()
+ << " batch=" << batch.size() << " i=" << i
+ << " tid=" << tid << dendl;
+
+ uint64_t batch_len = 0;
+ if (successes > 0) {
+ if (successes == batch.size()) {
+ batch.clear();
+ } else {
+ batch.erase(batch.begin(), batch.begin() + successes);
+ for (const auto& b : batch) {
+ batch_len += b.length() + part_entry_overhead;
+ }
+ }
+ }
+
+ if (batch.empty() && remaining.empty()) {
+ complete(std::move(p), 0);
+ return;
+ }
+
+ while (!remaining.empty() &&
+ (remaining.front().length() + batch_len <= max_part_size)) {
+
+ /* We can send entries with data_len up to max_entry_size,
+ however, we want to also account the overhead when
+ dealing with multiple entries. Previous check doesn't
+ account for overhead on purpose. */
+ batch_len += remaining.front().length() + part_entry_overhead;
+ batch.push_back(std::move(remaining.front()));
+ remaining.pop_front();
+ }
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " prepared push: remaining=" << remaining.size()
+ << " batch=" << batch.size() << " i=" << i
+ << " batch_len=" << batch_len
+ << " tid=" << tid << dendl;
+ push(std::move(p));
+ }
+
+ void push(Ptr&& p) {
+ f->push_entries(batch, tid, call(std::move(p)));
+ }
+
+ void new_head(const DoutPrefixProvider *dpp, Ptr&& p) {
+ state = new_heading;
+ f->_prepare_new_head(dpp, head_part_num + 1, tid, call(std::move(p)));
+ }
+
+ void read_meta(const DoutPrefixProvider *dpp, Ptr&& p) {
+ ++i;
+ state = meta_reading;
+ f->read_meta(dpp, tid, call(std::move(p)));
+ }
+
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ switch (state) {
+ case pushing:
+ if (r == -ERANGE) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " need new head tid=" << tid << dendl;
+ new_head(dpp, std::move(p));
+ return;
+ }
+ if (r == -ENOENT) {
+ if (i > MAX_RACE_RETRIES) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " racing client deleted part, but we're out"
+ << " of retries: tid=" << tid << dendl;
+ complete(std::move(p), r);
+ }
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " racing client deleted part: tid=" << tid << dendl;
+ read_meta(dpp, std::move(p));
+ return;
+ }
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " push_entries failed: r=" << r
+ << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+ i = 0; // We've made forward progress, so reset the race counter!
+ prep_then_push(dpp, std::move(p), r);
+ break;
+
+ case new_heading:
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " prepare_new_head failed: r=" << r
+ << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+ state = pushing;
+ handle_new_head(dpp, std::move(p), r);
+ break;
+
+ case meta_reading:
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " read_meta failed: r=" << r
+ << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+ state = pushing;
+ prep_then_push(dpp, std::move(p), r);
+ break;
+ }
+ }
+
+ void handle_new_head(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ if (r == -ECANCELED) {
+ if (p->i == MAX_RACE_RETRIES) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled too many times, giving up: tid=" << tid << dendl;
+ complete(std::move(p), -ECANCELED);
+ return;
+ }
+ ++p->i;
+ } else if (r) {
+ complete(std::move(p), r);
+ return;
+ }
+
+ if (p->batch.empty()) {
+ prep_then_push(dpp, std::move(p), 0);
+ return;
+ } else {
+ push(std::move(p));
+ return;
+ }
+ }
+
+ Pusher(const DoutPrefixProvider *dpp, FIFO* f, std::deque<cb::list>&& remaining,
+ std::int64_t head_part_num, std::uint64_t tid,
+ lr::AioCompletion* super)
+ : Completion(dpp, super), f(f), remaining(std::move(remaining)),
+ head_part_num(head_part_num), tid(tid) {}
+};
+
+void FIFO::push(const DoutPrefixProvider *dpp, const std::vector<cb::list>& data_bufs,
+ lr::AioCompletion* c)
+{
+ std::unique_lock l(m);
+ auto tid = ++next_tid;
+ auto max_entry_size = info.params.max_entry_size;
+ auto need_new_head = info.need_new_head();
+ auto head_part_num = info.head_part_num;
+ l.unlock();
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ auto p = std::make_unique<Pusher>(dpp, this, std::deque<cb::list>(data_bufs.begin(), data_bufs.end()),
+ head_part_num, tid, c);
+ // Validate sizes
+ for (const auto& bl : data_bufs) {
+ if (bl.length() > max_entry_size) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entry bigger than max_entry_size tid=" << tid << dendl;
+ Pusher::complete(std::move(p), -E2BIG);
+ return;
+ }
+ }
+
+ if (data_bufs.empty() ) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " empty push, returning success tid=" << tid << dendl;
+ Pusher::complete(std::move(p), 0);
+ return;
+ }
+
+ if (need_new_head) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " need new head tid=" << tid << dendl;
+ p->new_head(dpp, std::move(p));
+ } else {
+ p->prep_then_push(dpp, std::move(p), 0);
+ }
+}
+
+int FIFO::list(const DoutPrefixProvider *dpp, int max_entries,
+ std::optional<std::string_view> markstr,
+ std::vector<list_entry>* presult, bool* pmore,
+ optional_yield y)
+{
+ std::unique_lock l(m);
+ auto tid = ++next_tid;
+ std::int64_t part_num = info.tail_part_num;
+ l.unlock();
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ std::uint64_t ofs = 0;
+ if (markstr) {
+ auto marker = to_marker(*markstr);
+ if (!marker) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " invalid marker string: " << markstr
+ << " tid= "<< tid << dendl;
+ return -EINVAL;
+ }
+ part_num = marker->num;
+ ofs = marker->ofs;
+ }
+
+ std::vector<list_entry> result;
+ result.reserve(max_entries);
+ bool more = false;
+
+ std::vector<fifo::part_list_entry> entries;
+ int r = 0;
+ while (max_entries > 0) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " max_entries=" << max_entries << " tid=" << tid << dendl;
+ bool part_more = false;
+ bool part_full = false;
+
+ std::unique_lock l(m);
+ auto part_oid = info.part_oid(part_num);
+ l.unlock();
+
+ r = list_part(dpp, ioctx, part_oid, ofs, max_entries, &entries,
+ &part_more, &part_full, tid, y);
+ if (r == -ENOENT) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " missing part, rereading metadata"
+ << " tid= "<< tid << dendl;
+ r = read_meta(dpp, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " read_meta failed: r=" << r
+ << " tid= "<< tid << dendl;
+ return r;
+ }
+ if (part_num < info.tail_part_num) {
+ /* raced with trim? restart */
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " raced with trim, restarting: tid=" << tid << dendl;
+ max_entries += result.size();
+ result.clear();
+ std::unique_lock l(m);
+ part_num = info.tail_part_num;
+ l.unlock();
+ ofs = 0;
+ continue;
+ }
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " assuming part was not written yet, so end of data: "
+ << "tid=" << tid << dendl;
+ more = false;
+ r = 0;
+ break;
+ }
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " list_entries failed: r=" << r
+ << " tid= "<< tid << dendl;
+ return r;
+ }
+ more = part_full || part_more;
+ for (auto& entry : entries) {
+ list_entry e;
+ e.data = std::move(entry.data);
+ e.marker = marker{part_num, entry.ofs}.to_string();
+ e.mtime = entry.mtime;
+ result.push_back(std::move(e));
+ --max_entries;
+ if (max_entries == 0)
+ break;
+ }
+ entries.clear();
+ if (max_entries > 0 &&
+ part_more) {
+ }
+
+ if (!part_full) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " head part is not full, so we can assume we're done: "
+ << "tid=" << tid << dendl;
+ break;
+ }
+ if (!part_more) {
+ ++part_num;
+ ofs = 0;
+ }
+ }
+ if (presult)
+ *presult = std::move(result);
+ if (pmore)
+ *pmore = more;
+ return 0;
+}
+
+int FIFO::trim(const DoutPrefixProvider *dpp, std::string_view markstr, bool exclusive, optional_yield y)
+{
+ bool overshoot = false;
+ auto marker = to_marker(markstr);
+ if (!marker) {
+ return -EINVAL;
+ }
+ auto part_num = marker->num;
+ auto ofs = marker->ofs;
+ std::unique_lock l(m);
+ auto tid = ++next_tid;
+ auto hn = info.head_part_num;
+ const auto max_part_size = info.params.max_part_size;
+ if (part_num > hn) {
+ l.unlock();
+ auto r = read_meta(dpp, tid, y);
+ if (r < 0) {
+ return r;
+ }
+ l.lock();
+ auto hn = info.head_part_num;
+ if (part_num > hn) {
+ overshoot = true;
+ part_num = hn;
+ ofs = max_part_size;
+ }
+ }
+ if (part_num < info.tail_part_num) {
+ return -ENODATA;
+ }
+ auto pn = info.tail_part_num;
+ l.unlock();
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+
+ int r = 0;
+ while (pn < part_num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " pn=" << pn << " tid=" << tid << dendl;
+ std::unique_lock l(m);
+ l.unlock();
+ r = trim_part(dpp, pn, max_part_size, false, tid, y);
+ if (r < 0 && r == -ENOENT) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " trim_part failed: r=" << r
+ << " tid= "<< tid << dendl;
+ return r;
+ }
+ ++pn;
+ }
+ r = trim_part(dpp, part_num, ofs, exclusive, tid, y);
+ if (r < 0 && r != -ENOENT) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " trim_part failed: r=" << r
+ << " tid= "<< tid << dendl;
+ return r;
+ }
+
+ l.lock();
+ auto tail_part_num = info.tail_part_num;
+ auto objv = info.version;
+ l.unlock();
+ bool canceled = tail_part_num < part_num;
+ int retries = 0;
+ while ((tail_part_num < part_num) &&
+ canceled &&
+ (retries <= MAX_RACE_RETRIES)) {
+ r = _update_meta(dpp, fifo::update{}.tail_part_num(part_num), objv, &canceled,
+ tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " _update_meta failed: r=" << r
+ << " tid= "<< tid << dendl;
+ return r;
+ }
+ if (canceled) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled: retries=" << retries
+ << " tid=" << tid << dendl;
+ l.lock();
+ tail_part_num = info.tail_part_num;
+ objv = info.version;
+ l.unlock();
+ ++retries;
+ }
+ }
+ if (canceled) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled too many times, giving up: tid=" << tid << dendl;
+ return -EIO;
+ }
+ return overshoot ? -ENODATA : 0;
+}
+
+struct Trimmer : public Completion<Trimmer> {
+ FIFO* fifo;
+ std::int64_t part_num;
+ std::uint64_t ofs;
+ std::int64_t pn;
+ bool exclusive;
+ std::uint64_t tid;
+ bool update = false;
+ bool reread = false;
+ bool canceled = false;
+ bool overshoot = false;
+ int retries = 0;
+
+ Trimmer(const DoutPrefixProvider *dpp, FIFO* fifo, std::int64_t part_num, std::uint64_t ofs, std::int64_t pn,
+ bool exclusive, lr::AioCompletion* super, std::uint64_t tid)
+ : Completion(dpp, super), fifo(fifo), part_num(part_num), ofs(ofs), pn(pn),
+ exclusive(exclusive), tid(tid) {}
+
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+
+ if (reread) {
+ reread = false;
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " read_meta failed: r="
+ << r << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+ std::unique_lock l(fifo->m);
+ auto hn = fifo->info.head_part_num;
+ const auto max_part_size = fifo->info.params.max_part_size;
+ const auto tail_part_num = fifo->info.tail_part_num;
+ l.unlock();
+ if (part_num > hn) {
+ part_num = hn;
+ ofs = max_part_size;
+ overshoot = true;
+ }
+ if (part_num < tail_part_num) {
+ complete(std::move(p), -ENODATA);
+ return;
+ }
+ pn = tail_part_num;
+ if (pn < part_num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " pn=" << pn << " tid=" << tid << dendl;
+ fifo->trim_part(dpp, pn++, max_part_size, false, tid,
+ call(std::move(p)));
+ } else {
+ update = true;
+ canceled = tail_part_num < part_num;
+ fifo->trim_part(dpp, part_num, ofs, exclusive, tid, call(std::move(p)));
+ }
+ return;
+ }
+
+ if (r == -ENOENT) {
+ r = 0;
+ }
+
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << (update ? " update_meta " : " trim ") << "failed: r="
+ << r << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ }
+
+ if (!update) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " handling preceding trim callback: tid=" << tid << dendl;
+ retries = 0;
+ if (pn < part_num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " pn=" << pn << " tid=" << tid << dendl;
+ std::unique_lock l(fifo->m);
+ const auto max_part_size = fifo->info.params.max_part_size;
+ l.unlock();
+ fifo->trim_part(dpp, pn++, max_part_size, false, tid,
+ call(std::move(p)));
+ return;
+ }
+
+ std::unique_lock l(fifo->m);
+ const auto tail_part_num = fifo->info.tail_part_num;
+ l.unlock();
+ update = true;
+ canceled = tail_part_num < part_num;
+ fifo->trim_part(dpp, part_num, ofs, exclusive, tid, call(std::move(p)));
+ return;
+ }
+
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " handling update-needed callback: tid=" << tid << dendl;
+ std::unique_lock l(fifo->m);
+ auto tail_part_num = fifo->info.tail_part_num;
+ auto objv = fifo->info.version;
+ l.unlock();
+ if ((tail_part_num < part_num) &&
+ canceled) {
+ if (retries > MAX_RACE_RETRIES) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled too many times, giving up: tid=" << tid << dendl;
+ complete(std::move(p), -EIO);
+ return;
+ }
+ ++retries;
+ fifo->_update_meta(dpp, fifo::update{}
+ .tail_part_num(part_num), objv, &canceled,
+ tid, call(std::move(p)));
+ } else {
+ complete(std::move(p), overshoot ? -ENODATA : 0);
+ }
+ }
+};
+
+void FIFO::trim(const DoutPrefixProvider *dpp, std::string_view markstr, bool exclusive,
+ lr::AioCompletion* c) {
+ auto marker = to_marker(markstr);
+ auto realmark = marker.value_or(::rgw::cls::fifo::marker{});
+ std::unique_lock l(m);
+ const auto hn = info.head_part_num;
+ const auto max_part_size = info.params.max_part_size;
+ const auto pn = info.tail_part_num;
+ const auto part_oid = info.part_oid(pn);
+ auto tid = ++next_tid;
+ l.unlock();
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ auto trimmer = std::make_unique<Trimmer>(dpp, this, realmark.num, realmark.ofs,
+ pn, exclusive, c, tid);
+ if (!marker) {
+ Trimmer::complete(std::move(trimmer), -EINVAL);
+ return;
+ }
+ ++trimmer->pn;
+ auto ofs = marker->ofs;
+ if (marker->num > hn) {
+ trimmer->reread = true;
+ read_meta(dpp, tid, Trimmer::call(std::move(trimmer)));
+ return;
+ }
+ if (pn < marker->num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " pn=" << pn << " tid=" << tid << dendl;
+ ofs = max_part_size;
+ } else {
+ trimmer->update = true;
+ }
+ trim_part(dpp, pn, ofs, exclusive, tid, Trimmer::call(std::move(trimmer)));
+}
+
+int FIFO::get_part_info(const DoutPrefixProvider *dpp, int64_t part_num,
+ fifo::part_header* header,
+ optional_yield y)
+{
+ std::unique_lock l(m);
+ const auto part_oid = info.part_oid(part_num);
+ auto tid = ++next_tid;
+ l.unlock();
+ auto r = rgw::cls::fifo::get_part_info(dpp, ioctx, part_oid, header, tid, y);
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " get_part_info failed: r="
+ << r << " tid=" << tid << dendl;
+ }
+ return r;
+}
+
+void FIFO::get_part_info(int64_t part_num,
+ fifo::part_header* header,
+ lr::AioCompletion* c)
+{
+ std::unique_lock l(m);
+ const auto part_oid = info.part_oid(part_num);
+ auto tid = ++next_tid;
+ l.unlock();
+ auto op = rgw::cls::fifo::get_part_info(cct, header, tid);
+ auto r = ioctx.aio_operate(part_oid, c, &op, nullptr);
+ ceph_assert(r >= 0);
+}
+
+struct InfoGetter : Completion<InfoGetter> {
+ FIFO* fifo;
+ fifo::part_header header;
+ fu2::function<void(int r, fifo::part_header&&)> f;
+ std::uint64_t tid;
+ bool headerread = false;
+
+ InfoGetter(const DoutPrefixProvider *dpp, FIFO* fifo, fu2::function<void(int r, fifo::part_header&&)> f,
+ std::uint64_t tid, lr::AioCompletion* super)
+ : Completion(dpp, super), fifo(fifo), f(std::move(f)), tid(tid) {}
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ if (!headerread) {
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " read_meta failed: r="
+ << r << " tid=" << tid << dendl;
+ if (f)
+ f(r, {});
+ complete(std::move(p), r);
+ return;
+ }
+
+ auto info = fifo->meta();
+ auto hpn = info.head_part_num;
+ if (hpn < 0) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " no head, returning empty partinfo r="
+ << r << " tid=" << tid << dendl;
+ if (f)
+ f(0, {});
+ complete(std::move(p), r);
+ return;
+ }
+ headerread = true;
+ auto op = rgw::cls::fifo::get_part_info(fifo->cct, &header, tid);
+ std::unique_lock l(fifo->m);
+ auto oid = fifo->info.part_oid(hpn);
+ l.unlock();
+ r = fifo->ioctx.aio_operate(oid, call(std::move(p)), &op,
+ nullptr);
+ ceph_assert(r >= 0);
+ return;
+ }
+
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " get_part_info failed: r="
+ << r << " tid=" << tid << dendl;
+ }
+
+ if (f)
+ f(r, std::move(header));
+ complete(std::move(p), r);
+ return;
+ }
+};
+
+void FIFO::get_head_info(const DoutPrefixProvider *dpp, fu2::unique_function<void(int r,
+ fifo::part_header&&)> f,
+ lr::AioCompletion* c)
+{
+ std::unique_lock l(m);
+ auto tid = ++next_tid;
+ l.unlock();
+ auto ig = std::make_unique<InfoGetter>(dpp, this, std::move(f), tid, c);
+ read_meta(dpp, tid, InfoGetter::call(std::move(ig)));
+}
+
+struct JournalProcessor : public Completion<JournalProcessor> {
+private:
+ FIFO* const fifo;
+
+ std::vector<fifo::journal_entry> processed;
+ decltype(fifo->info.journal) journal;
+ decltype(journal)::iterator iter;
+ std::int64_t new_tail;
+ std::int64_t new_head;
+ std::int64_t new_max;
+ int race_retries = 0;
+ bool first_pp = true;
+ bool canceled = false;
+ std::uint64_t tid;
+
+ enum {
+ entry_callback,
+ pp_callback,
+ } state;
+
+ void create_part(const DoutPrefixProvider *dpp, Ptr&& p, int64_t part_num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ state = entry_callback;
+ lr::ObjectWriteOperation op;
+ op.create(false); /* We don't need exclusivity, part_init ensures
+ we're creating from the same journal entry. */
+ std::unique_lock l(fifo->m);
+ part_init(&op, fifo->info.params);
+ auto oid = fifo->info.part_oid(part_num);
+ l.unlock();
+ auto r = fifo->ioctx.aio_operate(oid, call(std::move(p)), &op);
+ ceph_assert(r >= 0);
+ return;
+ }
+
+ void remove_part(const DoutPrefixProvider *dpp, Ptr&& p, int64_t part_num) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ state = entry_callback;
+ lr::ObjectWriteOperation op;
+ op.remove();
+ std::unique_lock l(fifo->m);
+ auto oid = fifo->info.part_oid(part_num);
+ l.unlock();
+ auto r = fifo->ioctx.aio_operate(oid, call(std::move(p)), &op);
+ ceph_assert(r >= 0);
+ return;
+ }
+
+ void finish_je(const DoutPrefixProvider *dpp, Ptr&& p, int r,
+ const fifo::journal_entry& entry) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " finishing entry: entry=" << entry
+ << " tid=" << tid << dendl;
+
+ using enum fifo::journal_entry::Op;
+ if (entry.op == remove && r == -ENOENT)
+ r = 0;
+
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " processing entry failed: entry=" << entry
+ << " r=" << r << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ return;
+ } else {
+ switch (entry.op) {
+ case unknown:
+ case set_head:
+ // Can't happen. Filtered out in process.
+ complete(std::move(p), -EIO);
+ return;
+
+ case create:
+ if (entry.part_num > new_max) {
+ new_max = entry.part_num;
+ }
+ break;
+ case remove:
+ if (entry.part_num >= new_tail) {
+ new_tail = entry.part_num + 1;
+ }
+ break;
+ }
+ processed.push_back(entry);
+ }
+ ++iter;
+ process(dpp, std::move(p));
+ }
+
+ void postprocess(const DoutPrefixProvider *dpp, Ptr&& p) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ if (processed.empty()) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " nothing to update any more: race_retries="
+ << race_retries << " tid=" << tid << dendl;
+ complete(std::move(p), 0);
+ return;
+ }
+ pp_run(dpp, std::move(p), 0, false);
+ }
+
+public:
+
+ JournalProcessor(const DoutPrefixProvider *dpp, FIFO* fifo, std::uint64_t tid, lr::AioCompletion* super)
+ : Completion(dpp, super), fifo(fifo), tid(tid) {
+ std::unique_lock l(fifo->m);
+ journal = fifo->info.journal;
+ iter = journal.begin();
+ new_tail = fifo->info.tail_part_num;
+ new_head = fifo->info.head_part_num;
+ new_max = fifo->info.max_push_part_num;
+ }
+
+ void pp_run(const DoutPrefixProvider *dpp, Ptr&& p, int r, bool canceled) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ std::optional<int64_t> tail_part_num;
+ std::optional<int64_t> head_part_num;
+ std::optional<int64_t> max_part_num;
+
+ if (r < 0) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " failed, r=: " << r << " tid=" << tid << dendl;
+ complete(std::move(p), r);
+ }
+
+
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " postprocessing: race_retries="
+ << race_retries << " tid=" << tid << dendl;
+
+ if (!first_pp && r == 0 && !canceled) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " nothing to update any more: race_retries="
+ << race_retries << " tid=" << tid << dendl;
+ complete(std::move(p), 0);
+ return;
+ }
+
+ first_pp = false;
+
+ if (canceled) {
+ if (race_retries >= MAX_RACE_RETRIES) {
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " canceled too many times, giving up: tid="
+ << tid << dendl;
+ complete(std::move(p), -ECANCELED);
+ return;
+ }
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " update canceled, retrying: race_retries="
+ << race_retries << " tid=" << tid << dendl;
+
+ ++race_retries;
+
+ std::vector<fifo::journal_entry> new_processed;
+ std::unique_lock l(fifo->m);
+ for (auto& e : processed) {
+ if (fifo->info.journal.contains(e)) {
+ new_processed.push_back(e);
+ }
+ }
+ processed = std::move(new_processed);
+ }
+
+ std::unique_lock l(fifo->m);
+ auto objv = fifo->info.version;
+ if (new_tail > fifo->info.tail_part_num) {
+ tail_part_num = new_tail;
+ }
+
+ if (new_head > fifo->info.head_part_num) {
+ head_part_num = new_head;
+ }
+
+ if (new_max > fifo->info.max_push_part_num) {
+ max_part_num = new_max;
+ }
+ l.unlock();
+
+ if (processed.empty() &&
+ !tail_part_num &&
+ !max_part_num) {
+ /* nothing to update anymore */
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " nothing to update any more: race_retries="
+ << race_retries << " tid=" << tid << dendl;
+ complete(std::move(p), 0);
+ return;
+ }
+ state = pp_callback;
+ fifo->_update_meta(dpp, fifo::update{}
+ .tail_part_num(tail_part_num)
+ .head_part_num(head_part_num)
+ .max_push_part_num(max_part_num)
+ .journal_entries_rm(processed),
+ objv, &this->canceled, tid, call(std::move(p)));
+ return;
+ }
+
+ JournalProcessor(const JournalProcessor&) = delete;
+ JournalProcessor& operator =(const JournalProcessor&) = delete;
+ JournalProcessor(JournalProcessor&&) = delete;
+ JournalProcessor& operator =(JournalProcessor&&) = delete;
+
+ void process(const DoutPrefixProvider *dpp, Ptr&& p) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ while (iter != journal.end()) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " processing entry: entry=" << *iter
+ << " tid=" << tid << dendl;
+ const auto entry = *iter;
+ switch (entry.op) {
+ using enum fifo::journal_entry::Op;
+ case create:
+ create_part(dpp, std::move(p), entry.part_num);
+ return;
+ case set_head:
+ if (entry.part_num > new_head) {
+ new_head = entry.part_num;
+ }
+ processed.push_back(entry);
+ ++iter;
+ continue;
+ case remove:
+ remove_part(dpp, std::move(p), entry.part_num);
+ return;
+ default:
+ ldpp_dout(dpp, -1) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " unknown journaled op: entry=" << entry << " tid="
+ << tid << dendl;
+ complete(std::move(p), -EIO);
+ return;
+ }
+ }
+ postprocess(dpp, std::move(p));
+ return;
+ }
+
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ ldpp_dout(dpp, 20) << __PRETTY_FUNCTION__ << ":" << __LINE__
+ << " entering: tid=" << tid << dendl;
+ switch (state) {
+ case entry_callback:
+ finish_je(dpp, std::move(p), r, *iter);
+ return;
+ case pp_callback:
+ auto c = canceled;
+ canceled = false;
+ pp_run(dpp, std::move(p), r, c);
+ return;
+ }
+
+ abort();
+ }
+
+};
+
+void FIFO::process_journal(const DoutPrefixProvider *dpp, std::uint64_t tid, lr::AioCompletion* c) {
+ auto p = std::make_unique<JournalProcessor>(dpp, this, tid, c);
+ p->process(dpp, std::move(p));
+}
+
+struct Lister : Completion<Lister> {
+ FIFO* f;
+ std::vector<list_entry> result;
+ bool more = false;
+ std::int64_t part_num;
+ std::uint64_t ofs;
+ int max_entries;
+ int r_out = 0;
+ std::vector<fifo::part_list_entry> entries;
+ bool part_more = false;
+ bool part_full = false;
+ std::vector<list_entry>* entries_out;
+ bool* more_out;
+ std::uint64_t tid;
+
+ bool read = false;
+
+ void complete(Ptr&& p, int r) {
+ if (r >= 0) {
+ if (more_out) *more_out = more;
+ if (entries_out) *entries_out = std::move(result);
+ }
+ Completion::complete(std::move(p), r);
+ }
+
+public:
+ Lister(const DoutPrefixProvider *dpp, FIFO* f, std::int64_t part_num, std::uint64_t ofs, int max_entries,
+ std::vector<list_entry>* entries_out, bool* more_out,
+ std::uint64_t tid, lr::AioCompletion* super)
+ : Completion(dpp, super), f(f), part_num(part_num), ofs(ofs), max_entries(max_entries),
+ entries_out(entries_out), more_out(more_out), tid(tid) {
+ result.reserve(max_entries);
+ }
+
+ Lister(const Lister&) = delete;
+ Lister& operator =(const Lister&) = delete;
+ Lister(Lister&&) = delete;
+ Lister& operator =(Lister&&) = delete;
+
+ void handle(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ if (read)
+ handle_read(std::move(p), r);
+ else
+ handle_list(dpp, std::move(p), r);
+ }
+
+ void list(Ptr&& p) {
+ if (max_entries > 0) {
+ part_more = false;
+ part_full = false;
+ entries.clear();
+
+ std::unique_lock l(f->m);
+ auto part_oid = f->info.part_oid(part_num);
+ l.unlock();
+
+ read = false;
+ auto op = list_part(f->cct, ofs, max_entries, &r_out,
+ &entries, &part_more, &part_full, tid);
+ f->ioctx.aio_operate(part_oid, call(std::move(p)), &op, nullptr);
+ } else {
+ complete(std::move(p), 0);
+ }
+ }
+
+ void handle_read(Ptr&& p, int r) {
+ read = false;
+ if (r >= 0) r = r_out;
+ r_out = 0;
+
+ if (r < 0) {
+ complete(std::move(p), r);
+ return;
+ }
+
+ if (part_num < f->info.tail_part_num) {
+ /* raced with trim? restart */
+ max_entries += result.size();
+ result.clear();
+ part_num = f->info.tail_part_num;
+ ofs = 0;
+ list(std::move(p));
+ return;
+ }
+ /* assuming part was not written yet, so end of data */
+ more = false;
+ complete(std::move(p), 0);
+ return;
+ }
+
+ void handle_list(const DoutPrefixProvider *dpp, Ptr&& p, int r) {
+ if (r >= 0) r = r_out;
+ r_out = 0;
+ std::unique_lock l(f->m);
+ auto part_oid = f->info.part_oid(part_num);
+ l.unlock();
+ if (r == -ENOENT) {
+ read = true;
+ f->read_meta(dpp, tid, call(std::move(p)));
+ return;
+ }
+ if (r < 0) {
+ complete(std::move(p), r);
+ return;
+ }
+
+ more = part_full || part_more;
+ for (auto& entry : entries) {
+ list_entry e;
+ e.data = std::move(entry.data);
+ e.marker = marker{part_num, entry.ofs}.to_string();
+ e.mtime = entry.mtime;
+ result.push_back(std::move(e));
+ }
+ max_entries -= entries.size();
+ entries.clear();
+ if (max_entries > 0 && part_more) {
+ list(std::move(p));
+ return;
+ }
+
+ if (!part_full) { /* head part is not full */
+ complete(std::move(p), 0);
+ return;
+ }
+ ++part_num;
+ ofs = 0;
+ list(std::move(p));
+ }
+};
+
+void FIFO::list(const DoutPrefixProvider *dpp, int max_entries,
+ std::optional<std::string_view> markstr,
+ std::vector<list_entry>* out,
+ bool* more,
+ lr::AioCompletion* c) {
+ std::unique_lock l(m);
+ auto tid = ++next_tid;
+ std::int64_t part_num = info.tail_part_num;
+ l.unlock();
+ std::uint64_t ofs = 0;
+ std::optional<::rgw::cls::fifo::marker> marker;
+
+ if (markstr) {
+ marker = to_marker(*markstr);
+ if (marker) {
+ part_num = marker->num;
+ ofs = marker->ofs;
+ }
+ }
+
+ auto ls = std::make_unique<Lister>(dpp, this, part_num, ofs, max_entries, out,
+ more, tid, c);
+ if (markstr && !marker) {
+ auto l = ls.get();
+ l->complete(std::move(ls), -EINVAL);
+ } else {
+ ls->list(std::move(ls));
+ }
+}
+}
projects / ceph.git / blobdiff