librados: new style pgls

Signed-off-by: John Spray <john.spray@redhat.com>
Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/src/include/rados/librados.h b/src/include/rados/librados.h
index bda583a44f7565db95651d81b27ace74438bd4cc..c7e279a40b7f7ab1d2d66e276ea34abb01f58d58 100644 (file)
--- a/src/include/rados/librados.h
+++ b/src/include/rados/librados.h
@@ -189,6 +189,25 @@ typedef void *rados_ioctx_t;
  */
 typedef void *rados_list_ctx_t;
 
+/**
+ * @typedef rados_enumerate_cursor
+ *
+ * The cursor used with rados_enumerate_objects
+ * and accompanying methods.
+ */
+typedef void * rados_enumerate_cursor;
+
+typedef struct rados_enumerate_item {
+  size_t oid_length;
+  char *oid;
+
+  size_t nspace_length;
+  char *nspace;
+
+  size_t locator_length;
+  char *locator;
+} rados_enumerate_item;
+
 /**
  * @typedef rados_snap_t
  * The id of a snapshot.
@@ -956,6 +975,41 @@ CEPH_RADOS_API int rados_nobjects_list_next(rados_list_ctx_t ctx,
  */
 CEPH_RADOS_API void rados_nobjects_list_close(rados_list_ctx_t ctx);
 
+CEPH_RADOS_API rados_enumerate_cursor rados_enumerate_objects_begin(rados_ioctx_t io);
+CEPH_RADOS_API rados_enumerate_cursor rados_enumerate_objects_end(rados_ioctx_t io);
+
+CEPH_RADOS_API int rados_enumerate_objects_is_end(rados_ioctx_t io,
+    rados_enumerate_cursor cur);
+
+CEPH_RADOS_API void rados_enumerate_cursor_free(rados_ioctx_t io,
+    rados_enumerate_cursor cur);
+
+CEPH_RADOS_API int rados_enumerate_cursor_cmp(rados_ioctx_t io,
+    rados_enumerate_cursor lhs, rados_enumerate_cursor rhs);
+
+/**
+ * @return the number of items set in the result array
+ */
+CEPH_RADOS_API int rados_enumerate_objects(rados_ioctx_t io,
+    const rados_enumerate_cursor start,
+    const rados_enumerate_cursor finish,
+    const size_t result_size,
+    rados_enumerate_item *results,
+    rados_enumerate_cursor *next);
+
+CEPH_RADOS_API void rados_enumerate_objects_free(
+    const size_t result_size,
+    rados_enumerate_item *results);
+
+CEPH_RADOS_API void rados_enumerate_objects_split(rados_ioctx_t io,
+    const rados_enumerate_cursor start,
+    const rados_enumerate_cursor finish,
+    const size_t n,
+    const size_t m,
+    rados_enumerate_cursor *split_start,
+    rados_enumerate_cursor *split_finish);
+
+
 /** @} New Listing Objects */
 
 /**

diff --git a/src/librados/librados.cc b/src/librados/librados.cc
index 403b5b03a0a0d35b46e446a17247274bdcf53ba8..593545aa2ddb6b5fdbebd5a4eb0ad8a765ddb71a 100644 (file)
--- a/src/librados/librados.cc
+++ b/src/librados/librados.cc
@@ -19,6 +19,7 @@
 #include "common/ceph_argparse.h"
 #include "common/common_init.h"
 #include "common/TracepointProvider.h"
+#include "common/hobject.h"
 #include "include/rados/librados.h"
 #include "include/rados/librados.hpp"
 #include "include/types.h"
@@ -3578,6 +3579,113 @@ extern "C" int rados_exec(rados_ioctx_t io, const char *o, const char *cls, cons
   return ret;
 }
 
+extern "C" rados_enumerate_cursor rados_enumerate_objects_begin(rados_ioctx_t io)
+{
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+
+  hobject_t *result = new hobject_t(ctx->objecter->enumerate_objects_begin());
+  return (rados_enumerate_cursor)result;
+}
+
+extern "C" rados_enumerate_cursor rados_enumerate_objects_end(rados_ioctx_t io)
+{
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+
+  hobject_t *result = new hobject_t(ctx->objecter->enumerate_objects_end());
+  return (rados_enumerate_cursor)result;
+}
+
+extern "C" int rados_enumerate_objects_is_end(
+    rados_ioctx_t io, rados_enumerate_cursor cur)
+{
+  hobject_t *hobj = (hobject_t*)cur;
+  return hobj->is_max();
+}
+
+extern "C" void rados_enumerate_cursor_free(
+    rados_ioctx_t io, rados_enumerate_cursor cur)
+{
+  hobject_t *hobj = (hobject_t*)cur;
+  delete hobj;
+}
+
+extern "C" int rados_enumerate_cursor_cmp(
+    rados_ioctx_t io,
+    rados_enumerate_cursor lhs_cur,
+    rados_enumerate_cursor rhs_cur)
+{
+  hobject_t *lhs = (hobject_t*)lhs_cur;
+  hobject_t *rhs = (hobject_t*)rhs_cur;
+  return cmp_bitwise(*lhs, *rhs);
+}
+
+extern "C" int rados_enumerate_objects(rados_ioctx_t io,
+    const rados_enumerate_cursor start,
+    const rados_enumerate_cursor finish,
+    const size_t result_item_count,
+    rados_enumerate_item *result_items,
+    rados_enumerate_cursor *next)
+{
+  assert(next);
+
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+
+  // Zero out items so that they will be safe to free later
+  //memset(result_items, 0, sizeof(rados_enumerate_item) * result_item_count);
+
+  std::list<librados::ListObjectImpl> result;
+  hobject_t next_hash;
+
+  C_SaferCond cond;
+  ctx->objecter->enumerate_objects(
+      ctx->poolid,
+      ctx->oloc.nspace,
+      *((hobject_t*)start),
+      *((hobject_t*)finish),
+      result_item_count,
+      &result,
+      &next_hash,
+      &cond);
+
+  hobject_t *next_hobj = (hobject_t*)(*next);
+  assert(next_hobj);
+
+  int r = cond.wait();
+  if (r < 0) {
+    std::cerr << "objecter returned " << r << std::endl;
+    *next_hobj = hobject_t::get_max();
+    return r;
+  }
+
+  assert(result.size() <= result_item_count);  // Don't overflow!
+
+  int k = 0;
+  for (std::list<librados::ListObjectImpl>::iterator i = result.begin();
+       i != result.end(); ++i) {
+    rados_enumerate_item &item = result_items[k++];
+    do_out_buffer(i->oid, &item.oid, &item.oid_length);
+    do_out_buffer(i->nspace, &item.nspace, &item.nspace_length);
+    do_out_buffer(i->locator, &item.locator, &item.locator_length);
+  }
+
+  *next_hobj = next_hash;
+
+  return result.size();
+}
+
+extern "C" void rados_enumerate_objects_free(
+    const size_t result_size,
+    rados_enumerate_item *results)
+{
+  assert(results);
+
+  for (unsigned int i = 0; i < result_size; ++i) {
+    rados_buffer_free(results[i].oid);
+    rados_buffer_free(results[i].locator);
+    rados_buffer_free(results[i].nspace);
+  }
+}
+
 /* list objects */
 
 extern "C" int rados_nobjects_list_open(rados_ioctx_t io, rados_list_ctx_t *listh)
@@ -4970,3 +5078,54 @@ std::ostream& librados::operator<<(std::ostream& out, const librados::ListObject
   out << *(lop.impl);
   return out;
 }
+
+CEPH_RADOS_API void rados_enumerate_objects_split(
+    rados_ioctx_t io,
+    const rados_enumerate_cursor start,
+    const rados_enumerate_cursor finish,
+    const size_t n,
+    const size_t m,
+    rados_enumerate_cursor *split_start,
+    rados_enumerate_cursor *split_finish)
+{
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+
+  assert(split_start);
+  assert(split_finish);
+  hobject_t *split_start_hobj = (hobject_t*)(*split_start);
+  hobject_t *split_finish_hobj = (hobject_t*)(*split_finish);
+  assert(split_start_hobj);
+  assert(split_finish);
+  hobject_t *start_hobj = (hobject_t*)(start);
+  hobject_t *finish_hobj = (hobject_t*)(finish);
+
+  if (start_hobj->is_max()) {
+    *split_start_hobj = hobject_t::get_max();
+    *split_finish_hobj = hobject_t::get_max();
+    return;
+  }
+
+  uint64_t start_hash = hobject_t::_reverse_bits(start_hobj->get_hash());
+  uint64_t finish_hash =
+    finish_hobj->is_max() ? 0x100000000 :
+    hobject_t::_reverse_bits(finish_hobj->get_hash());
+
+  uint64_t diff = finish_hash - start_hash;
+  uint64_t rev_start = start_hash + (diff * n / m);
+  uint64_t rev_finish = start_hash + (diff * (n + 1) / m);
+  if (n == 0)
+    *split_start_hobj = *start_hobj;
+  else
+    *split_start_hobj = hobject_t(
+      object_t(), string(), CEPH_NOSNAP,
+      hobject_t::_reverse_bits(rev_start), ctx->poolid, string());
+  if (n == m - 1)
+    *split_finish_hobj = *finish_hobj;
+  else if (rev_finish >= 0x100000000)
+    *split_finish_hobj = hobject_t::get_max();
+  else
+    *split_finish_hobj = hobject_t(
+      object_t(), string(), CEPH_NOSNAP,
+      hobject_t::_reverse_bits(rev_finish), ctx->poolid, string());
+}
+