ceph/src/test/librados/list.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 #include "include/rados/librados.h"
   4 #include "include/rados/librados.hpp"
   5 #include "include/stringify.h"
   6 #include "test/librados/test.h"
   7 #include "test/librados/test_common.h"
   8 #include "test/librados/TestCase.h"
   9 #include "global/global_context.h"
  10
  11 #include "include/types.h"
  12 #include "common/hobject.h"
  13 #include "gtest/gtest.h"
  14 #include <errno.h>
  15 #include <string>
  16 #include <stdexcept>
  17
  18 using namespace librados;
  19
  20 typedef RadosTestNSCleanup LibRadosList;
  21 typedef RadosTestECNSCleanup LibRadosListEC;
  22 typedef RadosTestNP LibRadosListNP;
  23
  24
  25 TEST_F(LibRadosList, ListObjects) {
  26   char buf[128];
  27   memset(buf, 0xcc, sizeof(buf));
  28   ASSERT_EQ(0, rados_write(ioctx, "foo", buf, sizeof(buf), 0));
  29   rados_list_ctx_t ctx;
  30   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
  31   const char *entry;
  32   bool foundit = false;
  33   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) != -ENOENT) {
  34     foundit = true;
  35     ASSERT_EQ(std::string(entry), "foo");
  36   }
  37   ASSERT_TRUE(foundit);
  38   rados_nobjects_list_close(ctx);
  39 }
  40
  41
  42 static void check_list(
  43   std::set<std::string>& myset,
  44   rados_list_ctx_t& ctx,
  45   const std::string &check_nspace)
  46 {
  47   const char *entry, *nspace;
  48   cout << "myset " << myset << std::endl;
  49   // we should see every item exactly once.
  50   int ret;
  51   while ((ret = rados_nobjects_list_next(ctx, &entry, NULL, &nspace)) == 0) {
  52     std::string test_name;
  53     if (check_nspace == all_nspaces) {
  54       test_name = std::string(nspace) + ":" + std::string(entry);
  55     } else {
  56       ASSERT_TRUE(std::string(nspace) == check_nspace);
  57       test_name = std::string(entry);
  58     }
  59     cout << test_name << std::endl;
  60
  61     ASSERT_TRUE(myset.end() != myset.find(test_name));
  62     myset.erase(test_name);
  63   }
  64   ASSERT_EQ(-ENOENT, ret);
  65   ASSERT_TRUE(myset.empty());
  66 }
  67
  68 TEST_F(LibRadosList, ListObjectsNS) {
  69   char buf[128];
  70   memset(buf, 0xcc, sizeof(buf));
  71   // Create :foo1, :foo2, :foo3, n1:foo1, ns1:foo4, ns1:foo5, ns2:foo6, n2:foo7
  72   rados_ioctx_set_namespace(ioctx, "");
  73   ASSERT_EQ(0, rados_write(ioctx, "foo1", buf, sizeof(buf), 0));
  74   rados_ioctx_set_namespace(ioctx, "ns1");
  75   ASSERT_EQ(0, rados_write(ioctx, "foo1", buf, sizeof(buf), 0));
  76   rados_ioctx_set_namespace(ioctx, "");
  77   ASSERT_EQ(0, rados_write(ioctx, "foo2", buf, sizeof(buf), 0));
  78   ASSERT_EQ(0, rados_write(ioctx, "foo3", buf, sizeof(buf), 0));
  79   rados_ioctx_set_namespace(ioctx, "ns1");
  80   ASSERT_EQ(0, rados_write(ioctx, "foo4", buf, sizeof(buf), 0));
  81   ASSERT_EQ(0, rados_write(ioctx, "foo5", buf, sizeof(buf), 0));
  82   rados_ioctx_set_namespace(ioctx, "ns2");
  83   ASSERT_EQ(0, rados_write(ioctx, "foo6", buf, sizeof(buf), 0));
  84   ASSERT_EQ(0, rados_write(ioctx, "foo7", buf, sizeof(buf), 0));
  85
  86   char nspace[4];
  87   ASSERT_EQ(-ERANGE, rados_ioctx_get_namespace(ioctx, nspace, 3));
  88   ASSERT_EQ(static_cast<int>(strlen("ns2")),
  89             rados_ioctx_get_namespace(ioctx, nspace, sizeof(nspace)));
  90   ASSERT_EQ(0, strcmp("ns2", nspace));
  91
  92   std::set<std::string> def, ns1, ns2, all;
  93   def.insert(std::string("foo1"));
  94   def.insert(std::string("foo2"));
  95   def.insert(std::string("foo3"));
  96   ns1.insert(std::string("foo1"));
  97   ns1.insert(std::string("foo4"));
  98   ns1.insert(std::string("foo5"));
  99   ns2.insert(std::string("foo6"));
 100   ns2.insert(std::string("foo7"));
 101   all.insert(std::string(":foo1"));
 102   all.insert(std::string(":foo2"));
 103   all.insert(std::string(":foo3"));
 104   all.insert(std::string("ns1:foo1"));
 105   all.insert(std::string("ns1:foo4"));
 106   all.insert(std::string("ns1:foo5"));
 107   all.insert(std::string("ns2:foo6"));
 108   all.insert(std::string("ns2:foo7"));
 109
 110   rados_list_ctx_t ctx;
 111   // Check default namespace ""
 112   rados_ioctx_set_namespace(ioctx, "");
 113   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 114   check_list(def, ctx, "");
 115   rados_nobjects_list_close(ctx);
 116
 117   // Check namespace "ns1"
 118   rados_ioctx_set_namespace(ioctx, "ns1");
 119   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 120   check_list(ns1, ctx, "ns1");
 121   rados_nobjects_list_close(ctx);
 122
 123   // Check namespace "ns2"
 124   rados_ioctx_set_namespace(ioctx, "ns2");
 125   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 126   check_list(ns2, ctx, "ns2");
 127   rados_nobjects_list_close(ctx);
 128
 129   // Check ALL namespaces
 130   rados_ioctx_set_namespace(ioctx, LIBRADOS_ALL_NSPACES);
 131   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 132   check_list(all, ctx, all_nspaces);
 133   rados_nobjects_list_close(ctx);
 134 }
 135
 136
 137 TEST_F(LibRadosList, ListObjectsStart) {
 138   char buf[128];
 139   memset(buf, 0xcc, sizeof(buf));
 140
 141   for (int i=0; i<16; ++i) {
 142     string n = stringify(i);
 143     ASSERT_EQ(0, rados_write(ioctx, n.c_str(), buf, sizeof(buf), 0));
 144   }
 145
 146   rados_list_ctx_t ctx;
 147   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 148   std::map<int, std::set<std::string> > pg_to_obj;
 149   const char *entry;
 150   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) == 0) {
 151     uint32_t pos = rados_nobjects_list_get_pg_hash_position(ctx);
 152     std::cout << entry << " " << pos << std::endl;
 153     pg_to_obj[pos].insert(entry);
 154   }
 155   rados_nobjects_list_close(ctx);
 156
 157   std::map<int, std::set<std::string> >::reverse_iterator p =
 158     pg_to_obj.rbegin();
 159   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 160   while (p != pg_to_obj.rend()) {
 161     ASSERT_EQ((uint32_t)p->first, rados_nobjects_list_seek(ctx, p->first));
 162     ASSERT_EQ(0, rados_nobjects_list_next(ctx, &entry, NULL, NULL));
 163     std::cout << "have " << entry << " expect one of " << p->second << std::endl;
 164     ASSERT_TRUE(p->second.count(entry));
 165     ++p;
 166   }
 167   rados_nobjects_list_close(ctx);
 168 }
 169
 170 // this function replicates
 171 // librados::operator<<(std::ostream& os, const librados::ObjectCursor& oc)
 172 // because we don't want to use librados in librados client.
 173 std::ostream& operator<<(std::ostream&os, const rados_object_list_cursor& oc)
 174 {
 175   if (oc) {
 176     os << *(hobject_t *)oc;
 177   } else {
 178     os << hobject_t{};
 179   }
 180   return os;
 181 }
 182
 183 TEST_F(LibRadosList, ListObjectsCursor) {
 184   char buf[128];
 185   memset(buf, 0xcc, sizeof(buf));
 186
 187   const int max_objs = 16;
 188
 189   for (int i=0; i<max_objs; ++i) {
 190     string n = stringify(i);
 191     ASSERT_EQ(0, rados_write(ioctx, n.c_str(), buf, sizeof(buf), 0));
 192   }
 193
 194   {
 195     rados_list_ctx_t ctx;
 196     const char *entry;
 197     rados_object_list_cursor cursor;
 198     ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 199     ASSERT_EQ(rados_nobjects_list_get_cursor(ctx, &cursor), 0);
 200     rados_object_list_cursor first_cursor = cursor;
 201     cout << "x cursor=" << cursor << std::endl;
 202     while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) == 0) {
 203       string oid = entry;
 204       ASSERT_EQ(rados_nobjects_list_get_cursor(ctx, &cursor), 0);
 205       cout << "> oid=" << oid << " cursor=" << cursor << std::endl;
 206     }
 207     rados_nobjects_list_seek_cursor(ctx, first_cursor);
 208     ASSERT_EQ(rados_nobjects_list_next(ctx, &entry, NULL, NULL), 0);
 209     cout << "FIRST> seek to " << first_cursor << " oid=" << string(entry) << std::endl;
 210   }
 211   rados_list_ctx_t ctx;
 212   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 213
 214   std::map<rados_object_list_cursor, string> cursor_to_obj;
 215   int count = 0;
 216
 217   const char *entry;
 218   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) == 0) {
 219     rados_object_list_cursor cursor;
 220     ASSERT_EQ(rados_nobjects_list_get_cursor(ctx, &cursor), 0);
 221     string oid = entry;
 222     cout << ": oid=" << oid << " cursor=" << cursor << std::endl;
 223     cursor_to_obj[cursor] = oid;
 224
 225     rados_nobjects_list_seek_cursor(ctx, cursor);
 226     cout << ": seek to " << cursor << std::endl;
 227     ASSERT_EQ(rados_nobjects_list_next(ctx, &entry, NULL, NULL), 0);
 228     cout << "> " << cursor << " -> " << entry << std::endl;
 229     ASSERT_EQ(string(entry), oid);
 230     ASSERT_LT(count, max_objs); /* avoid infinite loops due to bad seek */
 231
 232     ++count;
 233   }
 234
 235   ASSERT_EQ(count, max_objs);
 236
 237   auto p = cursor_to_obj.rbegin();
 238   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 239   while (p != cursor_to_obj.rend()) {
 240     cout << ": seek to " << p->first << std::endl;
 241     rados_object_list_cursor cursor;
 242     rados_object_list_cursor oid(p->first);
 243     rados_nobjects_list_seek_cursor(ctx, oid);
 244     ASSERT_EQ(rados_nobjects_list_get_cursor(ctx, &cursor), 0);
 245     cout << ": cursor()=" << cursor << " expected=" << oid << std::endl;
 246     // ASSERT_EQ(ObjectCursor(oid), ObjectCursor(cursor));
 247     ASSERT_EQ(rados_nobjects_list_next(ctx, &entry, NULL, NULL), 0);
 248     cout << "> " << cursor << " -> " << entry << std::endl;
 249     cout << ": entry=" << entry << " expected=" << p->second << std::endl;
 250     ASSERT_EQ(p->second, string(entry));
 251
 252     ++p;
 253
 254     rados_object_list_cursor_free(ctx, cursor);
 255   }
 256 }
 257
 258 TEST_F(LibRadosListEC, ListObjects) {
 259   char buf[128];
 260   memset(buf, 0xcc, sizeof(buf));
 261   ASSERT_EQ(0, rados_write(ioctx, "foo", buf, sizeof(buf), 0));
 262   rados_list_ctx_t ctx;
 263   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 264   const char *entry;
 265   bool foundit = false;
 266   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) != -ENOENT) {
 267     foundit = true;
 268     ASSERT_EQ(std::string(entry), "foo");
 269   }
 270   ASSERT_TRUE(foundit);
 271   rados_nobjects_list_close(ctx);
 272 }
 273
 274 TEST_F(LibRadosListEC, ListObjectsNS) {
 275   char buf[128];
 276   memset(buf, 0xcc, sizeof(buf));
 277   // Create :foo1, :foo2, :foo3, n1:foo1, ns1:foo4, ns1:foo5, ns2:foo6, n2:foo7
 278   rados_ioctx_set_namespace(ioctx, "");
 279   ASSERT_EQ(0, rados_write(ioctx, "foo1", buf, sizeof(buf), 0));
 280   rados_ioctx_set_namespace(ioctx, "ns1");
 281   ASSERT_EQ(0, rados_write(ioctx, "foo1", buf, sizeof(buf), 0));
 282   rados_ioctx_set_namespace(ioctx, "");
 283   ASSERT_EQ(0, rados_write(ioctx, "foo2", buf, sizeof(buf), 0));
 284   ASSERT_EQ(0, rados_write(ioctx, "foo3", buf, sizeof(buf), 0));
 285   rados_ioctx_set_namespace(ioctx, "ns1");
 286   ASSERT_EQ(0, rados_write(ioctx, "foo4", buf, sizeof(buf), 0));
 287   ASSERT_EQ(0, rados_write(ioctx, "foo5", buf, sizeof(buf), 0));
 288   rados_ioctx_set_namespace(ioctx, "ns2");
 289   ASSERT_EQ(0, rados_write(ioctx, "foo6", buf, sizeof(buf), 0));
 290   ASSERT_EQ(0, rados_write(ioctx, "foo7", buf, sizeof(buf), 0));
 291
 292   std::set<std::string> def, ns1, ns2, all;
 293   def.insert(std::string("foo1"));
 294   def.insert(std::string("foo2"));
 295   def.insert(std::string("foo3"));
 296   ns1.insert(std::string("foo1"));
 297   ns1.insert(std::string("foo4"));
 298   ns1.insert(std::string("foo5"));
 299   ns2.insert(std::string("foo6"));
 300   ns2.insert(std::string("foo7"));
 301   all.insert(std::string(":foo1"));
 302   all.insert(std::string(":foo2"));
 303   all.insert(std::string(":foo3"));
 304   all.insert(std::string("ns1:foo1"));
 305   all.insert(std::string("ns1:foo4"));
 306   all.insert(std::string("ns1:foo5"));
 307   all.insert(std::string("ns2:foo6"));
 308   all.insert(std::string("ns2:foo7"));
 309
 310   rados_list_ctx_t ctx;
 311   // Check default namespace ""
 312   rados_ioctx_set_namespace(ioctx, "");
 313   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 314   check_list(def, ctx, "");
 315   rados_nobjects_list_close(ctx);
 316
 317   // Check default namespace "ns1"
 318   rados_ioctx_set_namespace(ioctx, "ns1");
 319   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 320   check_list(ns1, ctx, "ns1");
 321   rados_nobjects_list_close(ctx);
 322
 323   // Check default namespace "ns2"
 324   rados_ioctx_set_namespace(ioctx, "ns2");
 325   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 326   check_list(ns2, ctx, "ns2");
 327   rados_nobjects_list_close(ctx);
 328
 329   // Check all namespaces
 330   rados_ioctx_set_namespace(ioctx, LIBRADOS_ALL_NSPACES);
 331   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 332   check_list(all, ctx, all_nspaces);
 333   rados_nobjects_list_close(ctx);
 334 }
 335
 336
 337 TEST_F(LibRadosListEC, ListObjectsStart) {
 338   char buf[128];
 339   memset(buf, 0xcc, sizeof(buf));
 340
 341   for (int i=0; i<16; ++i) {
 342     string n = stringify(i);
 343     ASSERT_EQ(0, rados_write(ioctx, n.c_str(), buf, sizeof(buf), 0));
 344   }
 345
 346   rados_list_ctx_t ctx;
 347   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 348   std::map<int, std::set<std::string> > pg_to_obj;
 349   const char *entry;
 350   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) == 0) {
 351     uint32_t pos = rados_nobjects_list_get_pg_hash_position(ctx);
 352     std::cout << entry << " " << pos << std::endl;
 353     pg_to_obj[pos].insert(entry);
 354   }
 355   rados_nobjects_list_close(ctx);
 356
 357   std::map<int, std::set<std::string> >::reverse_iterator p =
 358     pg_to_obj.rbegin();
 359   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 360   while (p != pg_to_obj.rend()) {
 361     ASSERT_EQ((uint32_t)p->first, rados_nobjects_list_seek(ctx, p->first));
 362     ASSERT_EQ(0, rados_nobjects_list_next(ctx, &entry, NULL, NULL));
 363     std::cout << "have " << entry << " expect one of " << p->second << std::endl;
 364     ASSERT_TRUE(p->second.count(entry));
 365     ++p;
 366   }
 367   rados_nobjects_list_close(ctx);
 368 }
 369
 370 TEST_F(LibRadosListNP, ListObjectsError) {
 371   std::string pool_name;
 372   rados_t cluster;
 373   rados_ioctx_t ioctx;
 374   pool_name = get_temp_pool_name();
 375   ASSERT_EQ("", create_one_pool(pool_name, &cluster));
 376   ASSERT_EQ(0, rados_ioctx_create(cluster, pool_name.c_str(), &ioctx));
 377   char buf[128];
 378   memset(buf, 0xcc, sizeof(buf));
 379   rados_ioctx_set_namespace(ioctx, "");
 380   ASSERT_EQ(0, rados_write(ioctx, "foo", buf, sizeof(buf), 0));
 381
 382   //ASSERT_EQ(0, rados_pool_delete(cluster, pool_name.c_str()));
 383   {
 384     char *buf, *st;
 385     size_t buflen, stlen;
 386     string c = "{\"prefix\":\"osd pool rm\",\"pool\": \"" + pool_name +
 387       "\",\"pool2\":\"" + pool_name +
 388       "\",\"yes_i_really_really_mean_it_not_faking\": true}";
 389     const char *cmd[2] = { c.c_str(), 0 };
 390     ASSERT_EQ(0, rados_mon_command(cluster, (const char **)cmd, 1, "", 0, &buf, &buflen, &st, &stlen));
 391     ASSERT_EQ(0, rados_wait_for_latest_osdmap(cluster));
 392   }
 393
 394   rados_list_ctx_t ctx;
 395   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 396   const char *entry;
 397   ASSERT_EQ(-ENOENT, rados_nobjects_list_next(ctx, &entry, NULL, NULL));
 398   rados_nobjects_list_close(ctx);
 399   rados_ioctx_destroy(ioctx);
 400   rados_shutdown(cluster);
 401 }
 402
 403
 404
 405 // ---------------------------------------------
 406
 407 TEST_F(LibRadosList, EnumerateObjects) {
 408   char buf[128];
 409   memset(buf, 0xcc, sizeof(buf));
 410
 411   const uint32_t n_objects = 16;
 412   for (unsigned i=0; i<n_objects; ++i) {
 413     ASSERT_EQ(0, rados_write(ioctx, stringify(i).c_str(), buf, sizeof(buf), 0));
 414   }
 415
 416   // Ensure a non-power-of-two PG count to avoid only
 417   // touching the easy path.
 418   ASSERT_TRUE(set_pg_num(&s_cluster, pool_name, 11).empty());
 419   ASSERT_TRUE(set_pgp_num(&s_cluster, pool_name, 11).empty());
 420
 421   std::set<std::string> saw_obj;
 422   rados_object_list_cursor c = rados_object_list_begin(ioctx);
 423   rados_object_list_cursor end = rados_object_list_end(ioctx);
 424   while(!rados_object_list_is_end(ioctx, c))
 425   {
 426     rados_object_list_item results[12];
 427     memset(results, 0, sizeof(rados_object_list_item) * 12);
 428     rados_object_list_cursor temp_end = rados_object_list_end(ioctx);
 429     int r = rados_object_list(ioctx, c, temp_end,
 430             12, NULL, 0, results, &c);
 431     rados_object_list_cursor_free(ioctx, temp_end);
 432     ASSERT_GE(r, 0);
 433     for (int i = 0; i < r; ++i) {
 434       std::string oid(results[i].oid, results[i].oid_length);
 435       if (saw_obj.count(oid)) {
 436           std::cerr << "duplicate obj " << oid << std::endl;
 437       }
 438       ASSERT_FALSE(saw_obj.count(oid));
 439       saw_obj.insert(oid);
 440     }
 441     rados_object_list_free(12, results);
 442   }
 443   rados_object_list_cursor_free(ioctx, c);
 444   rados_object_list_cursor_free(ioctx, end);
 445
 446   for (unsigned i=0; i<n_objects; ++i) {
 447     if (!saw_obj.count(stringify(i))) {
 448         std::cerr << "missing object " << i << std::endl;
 449     }
 450     ASSERT_TRUE(saw_obj.count(stringify(i)));
 451   }
 452   ASSERT_EQ(n_objects, saw_obj.size());
 453 }
 454
 455 TEST_F(LibRadosList, EnumerateObjectsSplit) {
 456   char buf[128];
 457   memset(buf, 0xcc, sizeof(buf));
 458
 459   const uint32_t n_objects = 16;
 460   for (unsigned i=0; i<n_objects; ++i) {
 461     ASSERT_EQ(0, rados_write(ioctx, stringify(i).c_str(), buf, sizeof(buf), 0));
 462   }
 463
 464   // Ensure a non-power-of-two PG count to avoid only
 465   // touching the easy path.
 466   ASSERT_TRUE(set_pg_num(&s_cluster, pool_name, 11).empty());
 467   ASSERT_TRUE(set_pgp_num(&s_cluster, pool_name, 11).empty());
 468
 469   rados_object_list_cursor begin = rados_object_list_begin(ioctx);
 470   rados_object_list_cursor end = rados_object_list_end(ioctx);
 471
 472   // Step through an odd number of shards
 473   unsigned m = 5;
 474   std::set<std::string> saw_obj;
 475   for (unsigned n = 0; n < m; ++n) {
 476       rados_object_list_cursor shard_start = rados_object_list_begin(ioctx);;
 477       rados_object_list_cursor shard_end = rados_object_list_end(ioctx);;
 478
 479       rados_object_list_slice(
 480         ioctx,
 481         begin,
 482         end,
 483         n,
 484         m,
 485         &shard_start,
 486         &shard_end);
 487       std::cout << "split " << n << "/" << m << " -> "
 488                 << *(hobject_t*)shard_start << " "
 489                 << *(hobject_t*)shard_end << std::endl;
 490
 491       rados_object_list_cursor c = shard_start;
 492       //while(c < shard_end)
 493       while(rados_object_list_cursor_cmp(ioctx, c, shard_end) == -1)
 494       {
 495         rados_object_list_item results[12];
 496         memset(results, 0, sizeof(rados_object_list_item) * 12);
 497         int r = rados_object_list(ioctx,
 498                 c, shard_end,
 499                 12, NULL, 0, results, &c);
 500         ASSERT_GE(r, 0);
 501         for (int i = 0; i < r; ++i) {
 502           std::string oid(results[i].oid, results[i].oid_length);
 503           if (saw_obj.count(oid)) {
 504               std::cerr << "duplicate obj " << oid << std::endl;
 505           }
 506           ASSERT_FALSE(saw_obj.count(oid));
 507           saw_obj.insert(oid);
 508         }
 509         rados_object_list_free(12, results);
 510       }
 511       rados_object_list_cursor_free(ioctx, shard_start);
 512       rados_object_list_cursor_free(ioctx, shard_end);
 513   }
 514
 515   rados_object_list_cursor_free(ioctx, begin);
 516   rados_object_list_cursor_free(ioctx, end);
 517
 518   for (unsigned i=0; i<n_objects; ++i) {
 519     if (!saw_obj.count(stringify(i))) {
 520         std::cerr << "missing object " << i << std::endl;
 521     }
 522     ASSERT_TRUE(saw_obj.count(stringify(i)));
 523   }
 524   ASSERT_EQ(n_objects, saw_obj.size());
 525 }