--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "include/types.h"
+
+#include "cls/2pc_queue/cls_2pc_queue_types.h"
+#include "cls/2pc_queue/cls_2pc_queue_ops.h"
+#include "cls/2pc_queue/cls_2pc_queue_const.h"
+#include "cls/queue/cls_queue_ops.h"
+#include "cls/queue/cls_queue_src.h"
+#include "objclass/objclass.h"
+
+CLS_VER(1,0)
+CLS_NAME(2pc_queue)
+
+static int cls_2pc_queue_init(cls_method_context_t hctx, bufferlist *in, bufferlist *out) {
+ auto in_iter = in->cbegin();
+
+ cls_queue_init_op op;
+ try {
+ decode(op, in_iter);
+ } catch (buffer::error& err) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_init: failed to decode entry: %s", err.what());
+ return -EINVAL;
+ }
+
+ cls_2pc_urgent_data urgent_data;
+
+ cls_queue_init_op init_op;
+
+ CLS_LOG(20, "INFO: cls_2pc_queue_init: max size is %lu (bytes)", op.queue_size);
+
+ init_op.queue_size = op.queue_size;
+ init_op.max_urgent_data_size = 23552; // overall head is 24KB ~ pending 1K reservations ops
+ encode(urgent_data, init_op.bl_urgent_data);
+
+ return queue_init(hctx, init_op);
+}
+
+static int cls_2pc_queue_reserve(cls_method_context_t hctx, bufferlist *in, bufferlist *out) {
+ cls_2pc_queue_reserve_op res_op;
+ try {
+ auto in_iter = in->cbegin();
+ decode(res_op, in_iter);
+ } catch (buffer::error& err) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_reserve: failed to decode entry: %s", err.what());
+ return -EINVAL;
+ }
+
+ if (res_op.size == 0) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_reserve: cannot reserve zero bytes");
+ return -EINVAL;
+ }
+ if (res_op.entries == 0) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_reserve: cannot reserve zero entries");
+ return -EINVAL;
+ }
+
+ // get head
+ cls_queue_head head;
+ int ret = queue_read_head(hctx, head);
+ if (ret < 0) {
+ return ret;
+ }
+
+ cls_2pc_urgent_data urgent_data;
+ try {
+ auto in_iter = head.bl_urgent_data.cbegin();
+ decode(urgent_data, in_iter);
+ } catch (buffer::error& err) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_reserve: failed to decode entry: %s", err.what());
+ return -EINVAL;
+ }
+
+ const auto overhead = res_op.entries*QUEUE_ENTRY_OVERHEAD;
+ const auto remaining_size = (head.tail.offset >= head.front.offset) ?
+ (head.queue_size - head.tail.offset) + (head.front.offset - head.max_head_size) :
+ head.front.offset - head.tail.offset;
+
+
+ if (res_op.size + urgent_data.reserved_size + overhead > remaining_size) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_reserve: reservations exceeded maximum capacity");
+ CLS_LOG(10, "INFO: cls_2pc_queue_reserve: remaining size: %lu (bytes)", remaining_size);
+ CLS_LOG(10, "INFO: cls_2pc_queue_reserve: current reservations: %lu (bytes)", urgent_data.reserved_size);
+ CLS_LOG(10, "INFO: cls_2pc_queue_reserve: requested size: %lu (bytes)", res_op.size);
+ return -ENOSPC;
+ }
+
+ urgent_data.reserved_size += res_op.size + overhead;
+ bool result;
+ std::tie(std::ignore, result) = urgent_data.reservations.emplace(std::piecewise_construct,
+ std::forward_as_tuple(++urgent_data.last_id),
+ std::forward_as_tuple(res_op.size, ceph::real_clock::now()));
+ if (!result) {
+ // an old reservation that was never committed or aborted is in the map
+ // caller should try again assuming other IDs are ok
+ CLS_LOG(1, "ERROR: cls_2pc_queue_reserve: reservation id conflict after rollover: %u", urgent_data.last_id);
+ return -EAGAIN;
+ }
+
+ // write back head
+ head.bl_urgent_data.clear();
+ encode(urgent_data, head.bl_urgent_data);
+
+ const uint64_t urgent_data_length = head.bl_urgent_data.length();
+
+ if (head.max_urgent_data_size < urgent_data_length) {
+ // TODO: use objects xattr for spillover
+ CLS_LOG(1, "ERROR: cls_2pc_queue_reserve: urgent data size: %lu exceeded maximum: %lu", urgent_data_length, head.max_urgent_data_size);
+ return -ENOSPC;
+ }
+
+ ret = queue_write_head(hctx, head);
+ if (ret < 0) {
+ return ret;
+ }
+
+ CLS_LOG(20, "INFO: cls_2pc_queue_reserve: remaining size: %lu (bytes)", remaining_size);
+ CLS_LOG(20, "INFO: cls_2pc_queue_reserve: current reservations: %lu (bytes)", urgent_data.reserved_size);
+ CLS_LOG(20, "INFO: cls_2pc_queue_reserve: requested size: %lu (bytes)", res_op.size);
+ CLS_LOG(20, "INFO: cls_2pc_queue_reserve: urgent data size: %lu (bytes)", urgent_data_length);
+ CLS_LOG(20, "INFO: cls_2pc_queue_reserve: current reservation ops: %lu", urgent_data.reservations.size());
+
+ cls_2pc_queue_reserve_ret op_ret;
+ op_ret.id = urgent_data.last_id;
+ encode(op_ret, *out);
+
+ return 0;
+}
+
+static int cls_2pc_queue_commit(cls_method_context_t hctx, bufferlist *in, bufferlist *out) {
+ cls_2pc_queue_commit_op commit_op;
+ try {
+ auto in_iter = in->cbegin();
+ decode(commit_op, in_iter);
+ } catch (buffer::error& err) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_commit: failed to decode entry: %s", err.what());
+ return -EINVAL;
+ }
+
+ // get head
+ cls_queue_head head;
+ int ret = queue_read_head(hctx, head);
+ if (ret < 0) {
+ return ret;
+ }
+
+ cls_2pc_urgent_data urgent_data;
+ try {
+ auto in_iter = head.bl_urgent_data.cbegin();
+ decode(urgent_data, in_iter);
+ } catch (buffer::error& err) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_commit: failed to decode entry: %s", err.what());
+ return -EINVAL;
+ }
+
+ auto it = urgent_data.reservations.find(commit_op.id);
+ if (it == urgent_data.reservations.end()) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_commit: reservation does not exist: %u", commit_op.id);
+ return -ENOENT;
+ }
+
+ auto& res = it->second;
+ const auto actual_size = std::accumulate(commit_op.bl_data_vec.begin(),
+ commit_op.bl_data_vec.end(), 0UL, [] (uint64_t sum, const bufferlist& bl) {
+ return sum + bl.length();
+ });
+
+ if (res.size < actual_size) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_commit: trying to commit %lu bytes to a %lu bytes reservation",
+ actual_size,
+ res.size);
+ return -EINVAL;
+ }
+
+ // commit the data to the queue
+ cls_queue_enqueue_op enqueue_op;
+ enqueue_op.bl_data_vec = std::move(commit_op.bl_data_vec);
+ ret = queue_enqueue(hctx, enqueue_op, head);
+ if (ret < 0) {
+ return ret;
+ }
+
+ // remove the reservation
+ urgent_data.reserved_size -= res.size;
+ urgent_data.reservations.erase(it);
+
+ CLS_LOG(20, "INFO: cls_2pc_queue_commit: current reservations: %lu (bytes)", urgent_data.reserved_size);
+ CLS_LOG(20, "INFO: cls_2pc_queue_reserve: current reservation ops: %lu", urgent_data.reservations.size());
+ // write back head
+ head.bl_urgent_data.clear();
+ encode(urgent_data, head.bl_urgent_data);
+ return queue_write_head(hctx, head);
+}
+
+static int cls_2pc_queue_abort(cls_method_context_t hctx, bufferlist *in, bufferlist *out) {
+ cls_2pc_queue_abort_op abort_op;
+ try {
+ auto in_iter = in->cbegin();
+ decode(abort_op, in_iter);
+ } catch (buffer::error& err) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_abort: failed to decode entry: %s", err.what());
+ return -EINVAL;
+ }
+
+ // get head
+ cls_queue_head head;
+ int ret = queue_read_head(hctx, head);
+ if (ret < 0) {
+ return ret;
+ }
+
+ cls_2pc_urgent_data urgent_data;
+ try {
+ auto in_iter = head.bl_urgent_data.cbegin();
+ decode(urgent_data, in_iter);
+ } catch (buffer::error& err) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_abort: failed to decode entry: %s", err.what());
+ return -EINVAL;
+ }
+
+ auto it = urgent_data.reservations.find(abort_op.id);
+ if (it == urgent_data.reservations.end()) {
+ CLS_LOG(10, "INFO: cls_2pc_queue_abort: reservation does not exist: %u", abort_op.id);
+ return 0;
+ }
+
+ // remove the reservation
+ urgent_data.reserved_size -= it->second.size;
+ urgent_data.reservations.erase(it);
+
+ CLS_LOG(20, "INFO: cls_2pc_queue_abort: current reservations: %lu (bytes)", urgent_data.reserved_size);
+ CLS_LOG(20, "INFO: cls_2pc_queue_reserve: current reservation ops: %lu", urgent_data.reservations.size());
+
+ // write back head
+ head.bl_urgent_data.clear();
+ encode(urgent_data, head.bl_urgent_data);
+ return queue_write_head(hctx, head);
+}
+
+static int cls_2pc_queue_list_reservations(cls_method_context_t hctx, bufferlist *in, bufferlist *out) {
+ //get head
+ cls_queue_head head;
+ int ret = queue_read_head(hctx, head);
+ if (ret < 0) {
+ return ret;
+ }
+
+ cls_2pc_urgent_data urgent_data;
+ try {
+ auto in_iter = head.bl_urgent_data.cbegin();
+ decode(urgent_data, in_iter);
+ } catch (buffer::error& err) {
+ CLS_LOG(1, "ERROR: cls_2pc_queue_list_reservations: failed to decode entry: %s", err.what());
+ return -EINVAL;
+ }
+
+ cls_2pc_queue_reservations_ret op_ret;
+ op_ret.reservations = std::move(urgent_data.reservations);
+
+ encode(op_ret, *out);
+
+ return 0;
+}
+
+CLS_INIT(2pc_queue)
+{
+ CLS_LOG(1, "Loaded 2pc queue class!");
+
+ cls_handle_t h_class;
+ cls_method_handle_t h_2pc_queue_init;
+ cls_method_handle_t h_2pc_queue_reserve;
+ cls_method_handle_t h_2pc_queue_commit;
+ cls_method_handle_t h_2pc_queue_abort;
+ cls_method_handle_t h_2pc_queue_list_reservations;
+
+ cls_register(TPC_QUEUE_CLASS, &h_class);
+
+ cls_register_cxx_method(h_class, TPC_QUEUE_INIT, CLS_METHOD_RD | CLS_METHOD_WR, cls_2pc_queue_init, &h_2pc_queue_init);
+ cls_register_cxx_method(h_class, TPC_QUEUE_RESERVE, CLS_METHOD_RD | CLS_METHOD_WR, cls_2pc_queue_reserve, &h_2pc_queue_reserve);
+ cls_register_cxx_method(h_class, TPC_QUEUE_COMMIT, CLS_METHOD_RD | CLS_METHOD_WR, cls_2pc_queue_commit, &h_2pc_queue_commit);
+ cls_register_cxx_method(h_class, TPC_QUEUE_ABORT, CLS_METHOD_RD | CLS_METHOD_WR, cls_2pc_queue_abort, &h_2pc_queue_abort);
+ cls_register_cxx_method(h_class, TPC_QUEUE_LIST_RESERVATIONS, CLS_METHOD_RD, cls_2pc_queue_list_reservations, &h_2pc_queue_list_reservations);
+
+ return;
+}
+
--- /dev/null
+// -*- mode:C; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "include/types.h"
+
+#include "cls/2pc_queue/cls_2pc_queue_types.h"
+#include "cls/2pc_queue/cls_2pc_queue_client.h"
+#include "cls/queue/cls_queue_client.h"
+#include "cls/2pc_queue/cls_2pc_queue_types.h"
+
+#include "gtest/gtest.h"
+#include "test/librados/test_cxx.h"
+#include "global/global_context.h"
+
+#include <string>
+#include <vector>
+#include <algorithm>
+#include <thread>
+#include <chrono>
+#include <atomic>
+
+class TestCls2PCQueue : public ::testing::Test {
+protected:
+ librados::Rados rados;
+ std::string pool_name;
+ librados::IoCtx ioctx;
+
+ void SetUp() override {
+ pool_name = get_temp_pool_name();
+ ASSERT_EQ("", create_one_pool_pp(pool_name, rados));
+ ASSERT_EQ(0, rados.ioctx_create(pool_name.c_str(), ioctx));
+ }
+
+ void TearDown() override {
+ ioctx.close();
+ ASSERT_EQ(0, destroy_one_pool_pp(pool_name, rados));
+ }
+};
+
+TEST_F(TestCls2PCQueue, GetCapacity)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 8*1024;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ uint64_t size;
+
+ const int ret = cls_queue_get_capacity(ioctx, queue_name, size);
+ ASSERT_EQ(0, ret);
+ ASSERT_EQ(max_size, size);
+}
+
+TEST_F(TestCls2PCQueue, Reserve)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024U*1024U;
+ const auto number_of_ops = 10U;
+ const auto number_of_elements = 23U;
+ const auto size_to_reserve = 250U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ cls_2pc_reservation::id_t res_id;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ }
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), number_of_ops);
+ for (const auto& r : reservations) {
+ ASSERT_NE(r.first, cls_2pc_reservation::NO_ID);
+ ASSERT_GT(r.second.timestamp.time_since_epoch().count(), 0);
+ }
+}
+
+TEST_F(TestCls2PCQueue, Commit)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024*128;
+ const auto number_of_ops = 200U;
+ const auto number_of_elements = 23U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ const std::string element_prefix("op-" +to_string(i) + "-element-");
+ auto total_size = 0UL;
+ std::vector<bufferlist> data(number_of_elements);
+ // create vector of buffer lists
+ std::generate(data.begin(), data.end(), [j = 0, &element_prefix, &total_size] () mutable {
+ bufferlist bl;
+ bl.append(element_prefix + to_string(j++));
+ total_size += bl.length();
+ return bl;
+ });
+
+ cls_2pc_reservation::id_t res_id;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, total_size, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ cls_2pc_queue_commit(op, data, res_id);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), 0);
+}
+
+TEST_F(TestCls2PCQueue, Abort)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024U*1024U;
+ const auto number_of_ops = 17U;
+ const auto number_of_elements = 23U;
+ const auto size_to_reserve = 250U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ cls_2pc_reservation::id_t res_id;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ cls_2pc_queue_abort(op, res_id);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), 0);
+}
+
+TEST_F(TestCls2PCQueue, ReserveError)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 256U*1024U;
+ const auto number_of_ops = 254U;
+ const auto number_of_elements = 1U;
+ const auto size_to_reserve = 1024U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ cls_2pc_reservation::id_t res_id;
+ for (auto i = 0U; i < number_of_ops-1; ++i) {
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ }
+ res_id = cls_2pc_reservation::NO_ID;
+ // this one is failing because it exceeds the queue size
+ ASSERT_NE(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_EQ(res_id, cls_2pc_reservation::NO_ID);
+
+ // this one is failing because it tries to reserve 0 entries
+ ASSERT_NE(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, 0, res_id), 0);
+ // this one is failing because it tries to reserve 0 bytes
+ ASSERT_NE(cls_2pc_queue_reserve(ioctx, queue_name, op, 0, number_of_elements, res_id), 0);
+
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), number_of_ops-1);
+ for (const auto& r : reservations) {
+ ASSERT_NE(r.first, cls_2pc_reservation::NO_ID);
+ ASSERT_GT(r.second.timestamp.time_since_epoch().count(), 0);
+ }
+}
+
+TEST_F(TestCls2PCQueue, CommitError)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024;
+ const auto number_of_ops = 17U;
+ const auto number_of_elements = 23U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ const auto invalid_reservation_op = 8;
+ const auto invalid_elements_op = 11;
+ std::vector<bufferlist> invalid_data(number_of_elements+3);
+ // create vector of buffer lists
+ std::generate(invalid_data.begin(), invalid_data.end(), [j = 0] () mutable {
+ bufferlist bl;
+ bl.append("invalid data is larger that regular data" + to_string(j++));
+ return bl;
+ });
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ const std::string element_prefix("op-" +to_string(i) + "-element-");
+ std::vector<bufferlist> data(number_of_elements);
+ auto total_size = 0UL;
+ // create vector of buffer lists
+ std::generate(data.begin(), data.end(), [j = 0, &element_prefix, &total_size] () mutable {
+ bufferlist bl;
+ bl.append(element_prefix + to_string(j++));
+ total_size += bl.length();
+ return bl;
+ });
+
+ cls_2pc_reservation::id_t res_id;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, total_size, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ if (i == invalid_reservation_op) {
+ // fail on a commits with invalid reservation id
+ cls_2pc_queue_commit(op, data, res_id+999);
+ ASSERT_NE(0, ioctx.operate(queue_name, &op));
+ } else if (i == invalid_elements_op) {
+ // fail on a commits when data size is larger than the reserved one
+ cls_2pc_queue_commit(op, invalid_data, res_id);
+ ASSERT_NE(0, ioctx.operate(queue_name, &op));
+ } else {
+ cls_2pc_queue_commit(op, data, res_id);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ }
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ // 2 reservations were not comitted
+ ASSERT_EQ(reservations.size(), 2);
+}
+
+TEST_F(TestCls2PCQueue, AbortError)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024;
+ const auto number_of_ops = 17U;
+ const auto number_of_elements = 23U;
+ const auto size_to_reserve = 250U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ const auto invalid_reservation_op = 8;
+
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ cls_2pc_reservation::id_t res_id;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ if (i == invalid_reservation_op) {
+ // aborting a reservation which does not exists
+ // is a no-op, not an error
+ cls_2pc_queue_abort(op, res_id+999);
+ } else {
+ cls_2pc_queue_abort(op, res_id);
+ }
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ // 1 reservation was not aborted
+ ASSERT_EQ(reservations.size(), 1);
+}
+
+TEST_F(TestCls2PCQueue, MultiReserve)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024;
+ const auto number_of_ops = 11U;
+ const auto number_of_elements = 23U;
+ const auto max_producer_count = 10U;
+ const auto size_to_reserve = 250U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ std::vector<std::thread> producers(max_producer_count);
+ for (auto& p : producers) {
+ p = std::thread([this, &queue_name] {
+ librados::ObjectWriteOperation op;
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ cls_2pc_reservation::id_t res_id = cls_2pc_reservation::NO_ID;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, 0);
+ }
+ });
+ }
+
+ std::for_each(producers.begin(), producers.end(), [](auto& p) { p.join(); });
+
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), number_of_ops*max_producer_count);
+ auto total_reservations = 0U;
+ for (const auto& r : reservations) {
+ total_reservations += r.second.size;
+ }
+ ASSERT_EQ(total_reservations, number_of_ops*max_producer_count*size_to_reserve);
+}
+
+TEST_F(TestCls2PCQueue, MultiCommit)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024;
+ const auto number_of_ops = 11U;
+ const auto number_of_elements = 23U;
+ const auto max_producer_count = 10U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ std::vector<std::thread> producers(max_producer_count);
+ for (auto& p : producers) {
+ p = std::thread([this, &queue_name] {
+ librados::ObjectWriteOperation op;
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ const std::string element_prefix("op-" +to_string(i) + "-element-");
+ std::vector<bufferlist> data(number_of_elements);
+ auto total_size = 0UL;
+ // create vector of buffer lists
+ std::generate(data.begin(), data.end(), [j = 0, &element_prefix, &total_size] () mutable {
+ bufferlist bl;
+ bl.append(element_prefix + to_string(j++));
+ total_size += bl.length();
+ return bl;
+ });
+ cls_2pc_reservation::id_t res_id = cls_2pc_reservation::NO_ID;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, total_size, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, 0);
+ cls_2pc_queue_commit(op, data, res_id);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ });
+ }
+
+ std::for_each(producers.begin(), producers.end(), [](auto& p) { p.join(); });
+
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), 0);
+}
+
+TEST_F(TestCls2PCQueue, MultiAbort)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024;
+ const auto number_of_ops = 11U;
+ const auto number_of_elements = 23U;
+ const auto max_producer_count = 10U;
+ const auto size_to_reserve = 250U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ std::vector<std::thread> producers(max_producer_count);
+ for (auto& p : producers) {
+ p = std::thread([this, &queue_name] {
+ librados::ObjectWriteOperation op;
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ cls_2pc_reservation::id_t res_id = cls_2pc_reservation::NO_ID;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, 0);
+ cls_2pc_queue_abort(op, res_id);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ });
+ }
+
+ std::for_each(producers.begin(), producers.end(), [](auto& p) { p.join(); });
+
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), 0);
+}
+
+TEST_F(TestCls2PCQueue, ReserveCommit)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024;
+ const auto number_of_ops = 11U;
+ const auto number_of_elements = 23U;
+ const auto max_workers = 10U;
+ const auto size_to_reserve = 512U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ std::vector<std::thread> reservers(max_workers);
+ for (auto& r : reservers) {
+ r = std::thread([this, &queue_name] {
+ librados::ObjectWriteOperation op;
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ cls_2pc_reservation::id_t res_id = cls_2pc_reservation::NO_ID;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ }
+ });
+ }
+
+ auto committer = std::thread([this, &queue_name] {
+ librados::ObjectWriteOperation op;
+ int remaining_ops = number_of_ops*max_workers;
+ while (remaining_ops > 0) {
+ const std::string element_prefix("op-" +to_string(remaining_ops) + "-element-");
+ std::vector<bufferlist> data(number_of_elements);
+ // create vector of buffer lists
+ std::generate(data.begin(), data.end(), [j = 0, &element_prefix] () mutable {
+ bufferlist bl;
+ bl.append(element_prefix + to_string(j++));
+ return bl;
+ });
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ for (const auto& r : reservations) {
+ cls_2pc_queue_commit(op, data, r.first);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ --remaining_ops;
+ }
+ }
+ });
+
+ std::for_each(reservers.begin(), reservers.end(), [](auto& r) { r.join(); });
+ committer.join();
+
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), 0);
+}
+
+TEST_F(TestCls2PCQueue, ReserveAbort)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024;
+ const auto number_of_ops = 17U;
+ const auto number_of_elements = 23U;
+ const auto max_workers = 10U;
+ const auto size_to_reserve = 250U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ std::vector<std::thread> reservers(max_workers);
+ for (auto& r : reservers) {
+ r = std::thread([this, &queue_name] {
+ librados::ObjectWriteOperation op;
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ cls_2pc_reservation::id_t res_id = cls_2pc_reservation::NO_ID;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ }
+ });
+ }
+
+ auto aborter = std::thread([this, &queue_name] {
+ librados::ObjectWriteOperation op;
+ int remaining_ops = number_of_ops*max_workers;
+ while (remaining_ops > 0) {
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ for (const auto& r : reservations) {
+ cls_2pc_queue_abort(op, r.first);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ --remaining_ops;
+ }
+ }
+ });
+
+ std::for_each(reservers.begin(), reservers.end(), [](auto& r) { r.join(); });
+ aborter.join();
+
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), 0);
+}
+
+TEST_F(TestCls2PCQueue, Cleanup)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 128*1024*1024;
+ const auto number_of_ops = 17U;
+ const auto number_of_elements = 23U;
+ const auto max_workers = 10U;
+ const auto size_to_reserve = 512U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ // anything older than 100ms is cosidered stale
+ ceph::real_time stale_time = ceph::real_clock::now() + std::chrono::milliseconds(100);
+
+ std::vector<std::thread> reservers(max_workers);
+ for (auto& r : reservers) {
+ r = std::thread([this, &queue_name] {
+ librados::ObjectWriteOperation op;
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ cls_2pc_reservation::id_t res_id = cls_2pc_reservation::NO_ID;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, size_to_reserve, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, cls_2pc_reservation::NO_ID);
+ // wait for 10ms between each reservation to make sure at least some are stale
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+ });
+ }
+
+ auto cleaned_reservations = 0U;
+ auto committed_reservations = 0U;
+ auto aborter = std::thread([this, &queue_name, &stale_time, &cleaned_reservations, &committed_reservations] {
+ librados::ObjectWriteOperation op;
+ int remaining_ops = number_of_ops*max_workers;
+ while (remaining_ops > 0) {
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ for (const auto& r : reservations) {
+ if (r.second.timestamp > stale_time) {
+ // abort stale reservations
+ cls_2pc_queue_abort(op, r.first);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ ++cleaned_reservations;
+ } else {
+ // commit good reservations
+ const std::string element_prefix("op-" +to_string(remaining_ops) + "-element-");
+ std::vector<bufferlist> data(number_of_elements);
+ // create vector of buffer lists
+ std::generate(data.begin(), data.end(), [j = 0, &element_prefix] () mutable {
+ bufferlist bl;
+ bl.append(element_prefix + to_string(j++));
+ return bl;
+ });
+ cls_2pc_queue_commit(op, data, r.first);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ ++committed_reservations;
+ }
+ --remaining_ops;
+ }
+ }
+ });
+
+
+ std::for_each(reservers.begin(), reservers.end(), [](auto& r) { r.join(); });
+ aborter.join();
+
+ ASSERT_GT(cleaned_reservations, 0);
+ ASSERT_GT(committed_reservations, 0);
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), 0);
+}
+
+TEST_F(TestCls2PCQueue, MultiProducer)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 128*1024*1024;
+ const auto number_of_ops = 300U;
+ const auto number_of_elements = 23U;
+ const auto max_producer_count = 10U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ auto producer_count = max_producer_count;
+
+ std::vector<std::thread> producers(max_producer_count);
+ for (auto& p : producers) {
+ p = std::thread([this, &queue_name, &producer_count] {
+ librados::ObjectWriteOperation op;
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ const std::string element_prefix("op-" +to_string(i) + "-element-");
+ std::vector<bufferlist> data(number_of_elements);
+ auto total_size = 0UL;
+ // create vector of buffer lists
+ std::generate(data.begin(), data.end(), [j = 0, &element_prefix, &total_size] () mutable {
+ bufferlist bl;
+ bl.append(element_prefix + to_string(j++));
+ total_size += bl.length();
+ return bl;
+ });
+ cls_2pc_reservation::id_t res_id = cls_2pc_reservation::NO_ID;
+ ASSERT_EQ(cls_2pc_queue_reserve(ioctx, queue_name, op, total_size, number_of_elements, res_id), 0);
+ ASSERT_NE(res_id, 0);
+ cls_2pc_queue_commit(op, data, res_id);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ --producer_count;
+ });
+ }
+
+ auto consume_count = 0U;
+ std::thread consumer([this, &queue_name, &consume_count, &producer_count] {
+ librados::ObjectWriteOperation op;
+ const auto max_elements = 42;
+ const std::string marker;
+ bool truncated = false;
+ std::string end_marker;
+ std::vector<cls_queue_entry> entries;
+ while (producer_count > 0 || truncated) {
+ const auto ret = cls_2pc_queue_list_entries(ioctx, queue_name, marker, max_elements, entries, &truncated, end_marker);
+ ASSERT_EQ(0, ret);
+ consume_count += entries.size();
+ cls_2pc_queue_remove_entries(op, end_marker);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ });
+
+ std::for_each(producers.begin(), producers.end(), [](auto& p) { p.join(); });
+ consumer.join();
+ ASSERT_EQ(consume_count, number_of_ops*number_of_elements*max_producer_count);
+}
+
+TEST_F(TestCls2PCQueue, MultiProducerConsumer)
+{
+ const std::string queue_name = "my-queue";
+ const auto max_size = 1024*1024;
+ const auto number_of_ops = 300U;
+ const auto number_of_elements = 23U;
+ const auto max_workers = 10U;
+ librados::ObjectWriteOperation op;
+ op.create(true);
+ cls_2pc_queue_init(op, queue_name, max_size);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+
+ auto producer_count = max_workers;
+
+ auto retry_happened = false;
+
+ std::vector<std::thread> producers(max_workers);
+ for (auto& p : producers) {
+ p = std::thread([this, &queue_name, &producer_count, &retry_happened] {
+ librados::ObjectWriteOperation op;
+ for (auto i = 0U; i < number_of_ops; ++i) {
+ const std::string element_prefix("op-" +to_string(i) + "-element-");
+ std::vector<bufferlist> data(number_of_elements);
+ auto total_size = 0UL;
+ // create vector of buffer lists
+ std::generate(data.begin(), data.end(), [j = 0, &element_prefix, &total_size] () mutable {
+ bufferlist bl;
+ bl.append(element_prefix + to_string(j++));
+ total_size += bl.length();
+ return bl;
+ });
+ cls_2pc_reservation::id_t res_id = cls_2pc_reservation::NO_ID;
+ auto rc = cls_2pc_queue_reserve(ioctx, queue_name, op, total_size, number_of_elements, res_id);
+ while (rc != 0) {
+ // other errors should cause test to fail
+ ASSERT_EQ(rc, -ENOSPC);
+ ASSERT_EQ(res_id, 0);
+ // queue is full, sleep and retry
+ retry_happened = true;
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ rc = cls_2pc_queue_reserve(ioctx, queue_name, op, total_size, number_of_elements, res_id);
+ };
+ ASSERT_NE(res_id, 0);
+ cls_2pc_queue_commit(op, data, res_id);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ --producer_count;
+ });
+ }
+
+ const auto max_elements = 128;
+ std::vector<std::thread> consumers(max_workers/2);
+ for (auto& c : consumers) {
+ c = std::thread([this, &queue_name, &producer_count] {
+ librados::ObjectWriteOperation op;
+ const std::string marker;
+ bool truncated = false;
+ std::string end_marker;
+ std::vector<cls_queue_entry> entries;
+ while (producer_count > 0 || truncated) {
+ const auto ret = cls_2pc_queue_list_entries(ioctx, queue_name, marker, max_elements, entries, &truncated, end_marker);
+ ASSERT_EQ(0, ret);
+ if (entries.empty()) {
+ // queue is empty, let it fill
+ std::this_thread::sleep_for(std::chrono::milliseconds(100));
+ } else {
+ cls_2pc_queue_remove_entries(op, end_marker);
+ ASSERT_EQ(0, ioctx.operate(queue_name, &op));
+ }
+ }
+ });
+ }
+
+ std::for_each(producers.begin(), producers.end(), [](auto& p) { p.join(); });
+ std::for_each(consumers.begin(), consumers.end(), [](auto& c) { c.join(); });
+ if (!retry_happened) {
+ std::cerr << "Queue was never full - all reservations were sucessfull." <<
+ "Please decrease the amount of consumer threads" << std::endl;
+ }
+ // make sure that queue is empty and no reservations remain
+ cls_2pc_reservations reservations;
+ ASSERT_EQ(0, cls_2pc_queue_list_reservations(ioctx, queue_name, reservations));
+ ASSERT_EQ(reservations.size(), 0);
+ const std::string marker;
+ bool truncated = false;
+ std::string end_marker;
+ std::vector<cls_queue_entry> entries;
+ ASSERT_EQ(0, cls_2pc_queue_list_entries(ioctx, queue_name, marker, max_elements, entries, &truncated, end_marker));
+ ASSERT_EQ(entries.size(), 0);
+}
+
projects / ceph.git / commitdiff