api_service api_service_pp \
api_c_write_operations \
api_c_read_operations \
+ api_cls_remote_reads \
list_parallel \
open_pools_parallel \
delete_pools_parallel
INSTALL_RPATH ""
CXX_VISIBILITY_PRESET hidden)
install(TARGETS cls_fifo DESTINATION ${cls_dir})
+
+# cls_test_remote_reads
+set(cls_test_remote_reads_srcs test_remote_reads/cls_test_remote_reads.cc)
+add_library(cls_test_remote_reads SHARED ${cls_test_remote_reads_srcs})
+set_target_properties(cls_test_remote_reads PROPERTIES
+ VERSION "1.0.0"
+ SOVERSION "1"
+ INSTALL_RPATH ""
+ CXX_VISIBILITY_PRESET hidden)
+install(TARGETS cls_test_remote_reads DESTINATION ${cls_dir})
--- /dev/null
+/*
+ * This is an example RADOS object class that shows how to use remote reads.
+ */
+
+#include "common/ceph_json.h"
+#include "objclass/objclass.h"
+
+CLS_VER(1,0)
+CLS_NAME(test_remote_reads)
+
+cls_handle_t h_class;
+cls_method_handle_t h_test_read;
+cls_method_handle_t h_test_gather;
+
+/**
+ * read data
+ */
+static int test_read(cls_method_context_t hctx, bufferlist *in, bufferlist *out) {
+ int r = cls_cxx_read(hctx, 0, 0, out);
+ if (r < 0) {
+ CLS_ERR("%s: error reading data", __PRETTY_FUNCTION__);
+ return r;
+ }
+ return 0;
+}
+
+/**
+ * gather data from other objects using remote reads
+ */
+static int test_gather(cls_method_context_t hctx, bufferlist *in, bufferlist *out) {
+ std::map<std::string, bufferlist> src_obj_buffs;
+ int r = cls_cxx_get_gathered_data(hctx, &src_obj_buffs);
+ if (src_obj_buffs.empty()) {
+ // start remote reads
+ JSONParser parser;
+ bool b = parser.parse(in->c_str(), in->length());
+ if (!b) {
+ CLS_ERR("%s: failed to parse json", __PRETTY_FUNCTION__);
+ return -EBADMSG;
+ }
+ auto *o_cls = parser.find_obj("cls");
+ ceph_assert(o_cls);
+ std::string cls = o_cls->get_data_val().str;
+
+ auto *o_method = parser.find_obj("method");
+ ceph_assert(o_method);
+ std::string method = o_method->get_data_val().str;
+
+ auto *o_pool = parser.find_obj("pool");
+ ceph_assert(o_pool);
+ std::string pool = o_pool->get_data_val().str;
+
+ auto *o_src_objects = parser.find_obj("src_objects");
+ ceph_assert(o_src_objects);
+ auto src_objects_v = o_src_objects->get_array_elements();
+ std::set<std::string> src_objects;
+ for (auto it = src_objects_v.begin(); it != src_objects_v.end(); it++) {
+ std::string oid_without_double_quotes = it->substr(1, it->size()-2);
+ src_objects.insert(oid_without_double_quotes);
+ }
+ r = cls_cxx_gather(hctx, src_objects, pool, cls.c_str(), method.c_str(), *in);
+ } else {
+ // write data gathered using remote reads
+ int offset = 0;
+ for (std::map<std::string, bufferlist>::iterator it = src_obj_buffs.begin(); it != src_obj_buffs.end(); it++) {
+ bufferlist bl= it->second;
+ r = cls_cxx_write(hctx, offset, bl.length(), &bl);
+ offset += bl.length();
+ }
+ }
+ return r;
+}
+
+CLS_INIT(test_remote_reads)
+{
+ CLS_LOG(0, "loading cls_test_remote_reads");
+
+ cls_register("test_remote_reads", &h_class);
+
+ cls_register_cxx_method(h_class, "test_read",
+ CLS_METHOD_RD,
+ test_read, &h_test_read);
+
+ cls_register_cxx_method(h_class, "test_gather",
+ CLS_METHOD_RD | CLS_METHOD_WR,
+ test_gather, &h_test_gather);
+}
target_link_libraries(ceph_test_rados_api_snapshots_pp
librados ${UNITTEST_LIBS} radostest-cxx)
+add_executable(ceph_test_rados_api_cls_remote_reads
+ cls_remote_reads.cc
+ $<TARGET_OBJECTS:unit-main>)
+target_link_libraries(ceph_test_rados_api_cls_remote_reads
+ librados global ${UNITTEST_LIBS} radostest-cxx)
+
install(TARGETS
ceph_test_rados_api_aio
ceph_test_rados_api_aio_pp
ceph_test_rados_api_tier_pp
ceph_test_rados_api_watch_notify
ceph_test_rados_api_watch_notify_pp
+ ceph_test_rados_api_cls_remote_reads
DESTINATION ${CMAKE_INSTALL_BINDIR})
# unittest_librados
--- /dev/null
+#include <set>
+#include <string>
+
+#include "common/ceph_json.h"
+#include "gtest/gtest.h"
+#include "test/librados/test_cxx.h"
+
+using namespace librados;
+
+TEST(ClsTestRemoteReads, TestGather) {
+ Rados cluster;
+ std::string pool_name = get_temp_pool_name();
+ ASSERT_EQ("", create_one_pool_pp(pool_name, cluster));
+ IoCtx ioctx;
+ cluster.ioctx_create(pool_name.c_str(), ioctx);
+
+ bufferlist in, out;
+ int object_size = 4096;
+ char buf[object_size];
+ memset(buf, 1, sizeof(buf));
+
+ // create source objects from which data are gathered
+ in.append(buf, sizeof(buf));
+ ASSERT_EQ(0, ioctx.write_full("src_object.1", in));
+ in.append(buf, sizeof(buf));
+ ASSERT_EQ(0, ioctx.write_full("src_object.2", in));
+ in.append(buf, sizeof(buf));
+ ASSERT_EQ(0, ioctx.write_full("src_object.3", in));
+
+ // construct JSON request passed to "test_gather" method, and in turn, to "test_read" method
+ JSONFormatter *formatter = new JSONFormatter(true);
+ formatter->open_object_section("foo");
+ std::set<std::string> src_objects;
+ src_objects.insert("src_object.1");
+ src_objects.insert("src_object.2");
+ src_objects.insert("src_object.3");
+ encode_json("src_objects", src_objects, formatter);
+ encode_json("cls", "test_remote_reads", formatter);
+ encode_json("method", "test_read", formatter);
+ encode_json("pool", pool_name, formatter);
+ formatter->close_section();
+ in.clear();
+ formatter->flush(in);
+
+ // create target object by combining data gathered from source objects using "test_read" method
+ ASSERT_EQ(0, ioctx.exec("tgt_object", "test_remote_reads", "test_gather", in, out));
+
+ // read target object and check its size
+ ASSERT_EQ(3*object_size, ioctx.read("tgt_object", out, 0, 0));
+
+ ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
+}
projects / ceph.git / commitdiff