src/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 RadosTestPPNSCleanup LibRadosListPP;
  22 typedef RadosTestECNSCleanup LibRadosListEC;
  23 typedef RadosTestECPPNSCleanup LibRadosListECPP;
  24 typedef RadosTestNP LibRadosListNP;
  25
  26
  27 TEST_F(LibRadosList, ListObjects) {
  28   char buf[128];
  29   memset(buf, 0xcc, sizeof(buf));
  30   ASSERT_EQ(0, rados_write(ioctx, "foo", buf, sizeof(buf), 0));
  31   rados_list_ctx_t ctx;
  32   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
  33   const char *entry;
  34   bool foundit = false;
  35   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) != -ENOENT) {
  36     foundit = true;
  37     ASSERT_EQ(std::string(entry), "foo");
  38   }
  39   ASSERT_TRUE(foundit);
  40   rados_nobjects_list_close(ctx);
  41 }
  42
  43 TEST_F(LibRadosListPP, ListObjectsPP) {
  44   char buf[128];
  45   memset(buf, 0xcc, sizeof(buf));
  46   bufferlist bl1;
  47   bl1.append(buf, sizeof(buf));
  48   ASSERT_EQ(0, ioctx.write("foo", bl1, sizeof(buf), 0));
  49   NObjectIterator iter(ioctx.nobjects_begin());
  50   bool foundit = false;
  51   while (iter != ioctx.nobjects_end()) {
  52     foundit = true;
  53     ASSERT_EQ((*iter).get_oid(), "foo");
  54     ++iter;
  55   }
  56   ASSERT_TRUE(foundit);
  57 }
  58
  59 TEST_F(LibRadosListPP, ListObjectsTwicePP) {
  60   char buf[128];
  61   memset(buf, 0xcc, sizeof(buf));
  62   bufferlist bl1;
  63   bl1.append(buf, sizeof(buf));
  64   ASSERT_EQ(0, ioctx.write("foo", bl1, sizeof(buf), 0));
  65   NObjectIterator iter(ioctx.nobjects_begin());
  66   bool foundit = false;
  67   while (iter != ioctx.nobjects_end()) {
  68     foundit = true;
  69     ASSERT_EQ((*iter).get_oid(), "foo");
  70     ++iter;
  71   }
  72   ASSERT_TRUE(foundit);
  73   ++iter;
  74   ASSERT_TRUE(iter == ioctx.nobjects_end());
  75   foundit = false;
  76   iter.seek(0);
  77   while (iter != ioctx.nobjects_end()) {
  78     foundit = true;
  79     ASSERT_EQ((*iter).get_oid(), "foo");
  80     ++iter;
  81   }
  82   ASSERT_TRUE(foundit);
  83 }
  84
  85 TEST_F(LibRadosListPP, ListObjectsCopyIterPP) {
  86   char buf[128];
  87   memset(buf, 0xcc, sizeof(buf));
  88   bufferlist bl1;
  89   bl1.append(buf, sizeof(buf));
  90   ASSERT_EQ(0, ioctx.write("foo", bl1, sizeof(buf), 0));
  91
  92   // make sure this is still valid after the original iterators are gone
  93   NObjectIterator iter3;
  94   {
  95     NObjectIterator iter(ioctx.nobjects_begin());
  96     NObjectIterator iter2(iter);
  97     iter3 = iter2;
  98     ASSERT_EQ((*iter).get_oid(), "foo");
  99     ++iter;
 100     ASSERT_TRUE(iter == ioctx.nobjects_end());
 101     ++iter;
 102     ASSERT_TRUE(iter == ioctx.nobjects_end());
 103
 104     ASSERT_EQ(iter2->get_oid(), "foo");
 105     ASSERT_EQ(iter3->get_oid(), "foo");
 106     ++iter2;
 107     ASSERT_TRUE(iter2 == ioctx.nobjects_end());
 108   }
 109
 110   ASSERT_EQ(iter3->get_oid(), "foo");
 111   iter3 = iter3;
 112   ASSERT_EQ(iter3->get_oid(), "foo");
 113   ++iter3;
 114   ASSERT_TRUE(iter3 == ioctx.nobjects_end());
 115 }
 116
 117 TEST_F(LibRadosListPP, ListObjectsEndIter) {
 118   char buf[128];
 119   memset(buf, 0xcc, sizeof(buf));
 120   bufferlist bl1;
 121   bl1.append(buf, sizeof(buf));
 122   ASSERT_EQ(0, ioctx.write("foo", bl1, sizeof(buf), 0));
 123
 124   NObjectIterator iter(ioctx.nobjects_begin());
 125   NObjectIterator iter_end(ioctx.nobjects_end());
 126   NObjectIterator iter_end2 = ioctx.nobjects_end();
 127   ASSERT_TRUE(iter_end == iter_end2);
 128   ASSERT_TRUE(iter_end == ioctx.nobjects_end());
 129   ASSERT_TRUE(iter_end2 == ioctx.nobjects_end());
 130
 131   ASSERT_EQ(iter->get_oid(), "foo");
 132   ++iter;
 133   ASSERT_TRUE(iter == ioctx.nobjects_end());
 134   ASSERT_TRUE(iter == iter_end);
 135   ASSERT_TRUE(iter == iter_end2);
 136   NObjectIterator iter2 = iter;
 137   ASSERT_TRUE(iter2 == ioctx.nobjects_end());
 138   ASSERT_TRUE(iter2 == iter_end);
 139   ASSERT_TRUE(iter2 == iter_end2);
 140 }
 141
 142 static void check_list(
 143   std::set<std::string>& myset,
 144   rados_list_ctx_t& ctx,
 145   std::string check_nspace)
 146 {
 147   const char *entry, *nspace;
 148   cout << "myset " << myset << std::endl;
 149   // we should see every item exactly once.
 150   int ret;
 151   while ((ret = rados_nobjects_list_next(ctx, &entry, NULL, &nspace)) == 0) {
 152     std::string test_name;
 153     if (check_nspace == all_nspaces) {
 154       test_name = std::string(nspace) + ":" + std::string(entry);
 155     } else {
 156       ASSERT_TRUE(std::string(nspace) == check_nspace);
 157       test_name = std::string(entry);
 158     }
 159     cout << test_name << std::endl;
 160
 161     ASSERT_TRUE(myset.end() != myset.find(test_name));
 162     myset.erase(test_name);
 163   }
 164   ASSERT_EQ(-ENOENT, ret);
 165   ASSERT_TRUE(myset.empty());
 166 }
 167
 168 TEST_F(LibRadosList, ListObjectsNS) {
 169   char buf[128];
 170   memset(buf, 0xcc, sizeof(buf));
 171   // Create :foo1, :foo2, :foo3, n1:foo1, ns1:foo4, ns1:foo5, ns2:foo6, n2:foo7
 172   rados_ioctx_set_namespace(ioctx, "");
 173   ASSERT_EQ(0, rados_write(ioctx, "foo1", buf, sizeof(buf), 0));
 174   rados_ioctx_set_namespace(ioctx, "ns1");
 175   ASSERT_EQ(0, rados_write(ioctx, "foo1", buf, sizeof(buf), 0));
 176   rados_ioctx_set_namespace(ioctx, "");
 177   ASSERT_EQ(0, rados_write(ioctx, "foo2", buf, sizeof(buf), 0));
 178   ASSERT_EQ(0, rados_write(ioctx, "foo3", buf, sizeof(buf), 0));
 179   rados_ioctx_set_namespace(ioctx, "ns1");
 180   ASSERT_EQ(0, rados_write(ioctx, "foo4", buf, sizeof(buf), 0));
 181   ASSERT_EQ(0, rados_write(ioctx, "foo5", buf, sizeof(buf), 0));
 182   rados_ioctx_set_namespace(ioctx, "ns2");
 183   ASSERT_EQ(0, rados_write(ioctx, "foo6", buf, sizeof(buf), 0));
 184   ASSERT_EQ(0, rados_write(ioctx, "foo7", buf, sizeof(buf), 0));
 185
 186   std::set<std::string> def, ns1, ns2, all;
 187   def.insert(std::string("foo1"));
 188   def.insert(std::string("foo2"));
 189   def.insert(std::string("foo3"));
 190   ns1.insert(std::string("foo1"));
 191   ns1.insert(std::string("foo4"));
 192   ns1.insert(std::string("foo5"));
 193   ns2.insert(std::string("foo6"));
 194   ns2.insert(std::string("foo7"));
 195   all.insert(std::string(":foo1"));
 196   all.insert(std::string(":foo2"));
 197   all.insert(std::string(":foo3"));
 198   all.insert(std::string("ns1:foo1"));
 199   all.insert(std::string("ns1:foo4"));
 200   all.insert(std::string("ns1:foo5"));
 201   all.insert(std::string("ns2:foo6"));
 202   all.insert(std::string("ns2:foo7"));
 203
 204   rados_list_ctx_t ctx;
 205   // Check default namespace ""
 206   rados_ioctx_set_namespace(ioctx, "");
 207   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 208   check_list(def, ctx, "");
 209   rados_nobjects_list_close(ctx);
 210
 211   // Check namespace "ns1"
 212   rados_ioctx_set_namespace(ioctx, "ns1");
 213   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 214   check_list(ns1, ctx, "ns1");
 215   rados_nobjects_list_close(ctx);
 216
 217   // Check namespace "ns2"
 218   rados_ioctx_set_namespace(ioctx, "ns2");
 219   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 220   check_list(ns2, ctx, "ns2");
 221   rados_nobjects_list_close(ctx);
 222
 223   // Check ALL namespaces
 224   rados_ioctx_set_namespace(ioctx, LIBRADOS_ALL_NSPACES);
 225   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 226   check_list(all, ctx, all_nspaces);
 227   rados_nobjects_list_close(ctx);
 228 }
 229
 230 static void check_listpp(std::set<std::string>& myset, IoCtx& ioctx, std::string check_nspace)
 231 {
 232   NObjectIterator iter(ioctx.nobjects_begin());
 233   std::set<std::string> orig_set(myset);
 234   /**
 235    * During splitting, we might see duplicate items.
 236    * We assert that every object returned is in myset and that
 237    * we don't hit ENOENT until we have hit every item in myset
 238    * at least once.
 239    */
 240   while (iter != ioctx.nobjects_end()) {
 241     std::string test_name;
 242     if (check_nspace == all_nspaces) {
 243       test_name = iter->get_nspace() + ":" + iter->get_oid();
 244     } else {
 245       ASSERT_TRUE(iter->get_nspace() == check_nspace);
 246       test_name = iter->get_oid();
 247     }
 248     ASSERT_TRUE(orig_set.end() != orig_set.find(test_name));
 249     myset.erase(test_name);
 250     ++iter;
 251   }
 252   ASSERT_TRUE(myset.empty());
 253 }
 254
 255 TEST_F(LibRadosListPP, ListObjectsPPNS) {
 256   char buf[128];
 257   memset(buf, 0xcc, sizeof(buf));
 258   bufferlist bl1;
 259   bl1.append(buf, sizeof(buf));
 260   // Create :foo1, :foo2, :foo3, n1:foo1, ns1:foo4, ns1:foo5, ns2:foo6, n2:foo7
 261   ioctx.set_namespace("");
 262   ASSERT_EQ(0, ioctx.write("foo1", bl1, sizeof(buf), 0));
 263   ioctx.set_namespace("ns1");
 264   ASSERT_EQ(0, ioctx.write("foo1", bl1, sizeof(buf), 0));
 265   ioctx.set_namespace("");
 266   ASSERT_EQ(0, ioctx.write("foo2", bl1, sizeof(buf), 0));
 267   ASSERT_EQ(0, ioctx.write("foo3", bl1, sizeof(buf), 0));
 268   ioctx.set_namespace("ns1");
 269   ASSERT_EQ(0, ioctx.write("foo4", bl1, sizeof(buf), 0));
 270   ASSERT_EQ(0, ioctx.write("foo5", bl1, sizeof(buf), 0));
 271   ioctx.set_namespace("ns2");
 272   ASSERT_EQ(0, ioctx.write("foo6", bl1, sizeof(buf), 0));
 273   ASSERT_EQ(0, ioctx.write("foo7", bl1, sizeof(buf), 0));
 274
 275   std::set<std::string> def, ns1, ns2, all;
 276   def.insert(std::string("foo1"));
 277   def.insert(std::string("foo2"));
 278   def.insert(std::string("foo3"));
 279   ns1.insert(std::string("foo1"));
 280   ns1.insert(std::string("foo4"));
 281   ns1.insert(std::string("foo5"));
 282   ns2.insert(std::string("foo6"));
 283   ns2.insert(std::string("foo7"));
 284   all.insert(std::string(":foo1"));
 285   all.insert(std::string(":foo2"));
 286   all.insert(std::string(":foo3"));
 287   all.insert(std::string("ns1:foo1"));
 288   all.insert(std::string("ns1:foo4"));
 289   all.insert(std::string("ns1:foo5"));
 290   all.insert(std::string("ns2:foo6"));
 291   all.insert(std::string("ns2:foo7"));
 292
 293   ioctx.set_namespace("");
 294   check_listpp(def, ioctx, "");
 295
 296   ioctx.set_namespace("ns1");
 297   check_listpp(ns1, ioctx, "ns1");
 298
 299   ioctx.set_namespace("ns2");
 300   check_listpp(ns2, ioctx, "ns2");
 301
 302   ioctx.set_namespace(all_nspaces);
 303   check_listpp(all, ioctx, all_nspaces);
 304 }
 305
 306 TEST_F(LibRadosListPP, ListObjectsManyPP) {
 307   char buf[128];
 308   memset(buf, 0xcc, sizeof(buf));
 309   bufferlist bl;
 310   bl.append(buf, sizeof(buf));
 311
 312   for (int i=0; i<256; ++i) {
 313     ASSERT_EQ(0, ioctx.write(stringify(i), bl, bl.length(), 0));
 314   }
 315
 316   librados::NObjectIterator it = ioctx.nobjects_begin();
 317   std::set<std::string> saw_obj;
 318   std::set<int> saw_pg;
 319   for (; it != ioctx.nobjects_end(); ++it) {
 320     std::cout << it->get_oid()
 321               << " " << it.get_pg_hash_position() << std::endl;
 322     saw_obj.insert(it->get_oid());
 323     saw_pg.insert(it.get_pg_hash_position());
 324   }
 325   std::cout << "saw " << saw_pg.size() << " pgs " << std::endl;
 326
 327   // make sure they are 0..n
 328   for (unsigned i = 0; i < saw_pg.size(); ++i)
 329     ASSERT_TRUE(saw_pg.count(i));
 330 }
 331
 332 TEST_F(LibRadosList, ListObjectsStart) {
 333   char buf[128];
 334   memset(buf, 0xcc, sizeof(buf));
 335
 336   for (int i=0; i<16; ++i) {
 337     string n = stringify(i);
 338     ASSERT_EQ(0, rados_write(ioctx, n.c_str(), buf, sizeof(buf), 0));
 339   }
 340
 341   rados_list_ctx_t ctx;
 342   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 343   std::map<int, std::set<std::string> > pg_to_obj;
 344   const char *entry;
 345   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) == 0) {
 346     uint32_t pos = rados_nobjects_list_get_pg_hash_position(ctx);
 347     std::cout << entry << " " << pos << std::endl;
 348     pg_to_obj[pos].insert(entry);
 349   }
 350   rados_nobjects_list_close(ctx);
 351
 352   std::map<int, std::set<std::string> >::reverse_iterator p =
 353     pg_to_obj.rbegin();
 354   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 355   while (p != pg_to_obj.rend()) {
 356     ASSERT_EQ((uint32_t)p->first, rados_nobjects_list_seek(ctx, p->first));
 357     ASSERT_EQ(0, rados_nobjects_list_next(ctx, &entry, NULL, NULL));
 358     std::cout << "have " << entry << " expect one of " << p->second << std::endl;
 359     ASSERT_TRUE(p->second.count(entry));
 360     ++p;
 361   }
 362   rados_nobjects_list_close(ctx);
 363 }
 364
 365 TEST_F(LibRadosListPP, ListObjectsStartPP) {
 366   char buf[128];
 367   memset(buf, 0xcc, sizeof(buf));
 368   bufferlist bl;
 369   bl.append(buf, sizeof(buf));
 370
 371   for (int i=0; i<16; ++i) {
 372     ASSERT_EQ(0, ioctx.write(stringify(i), bl, bl.length(), 0));
 373   }
 374
 375   librados::NObjectIterator it = ioctx.nobjects_begin();
 376   std::map<int, std::set<std::string> > pg_to_obj;
 377   for (; it != ioctx.nobjects_end(); ++it) {
 378     std::cout << it->get_oid() << " " << it.get_pg_hash_position() << std::endl;
 379     pg_to_obj[it.get_pg_hash_position()].insert(it->get_oid());
 380   }
 381
 382   std::map<int, std::set<std::string> >::reverse_iterator p =
 383     pg_to_obj.rbegin();
 384   it = ioctx.nobjects_begin(p->first);
 385   while (p != pg_to_obj.rend()) {
 386     ASSERT_EQ((uint32_t)p->first, it.seek(p->first));
 387     std::cout << "have " << it->get_oid() << " expect one of " << p->second << std::endl;
 388     ASSERT_TRUE(p->second.count(it->get_oid()));
 389     ++p;
 390   }
 391 }
 392
 393 TEST_F(LibRadosListPP, ListObjectsCursorNSPP) {
 394   char buf[128];
 395   memset(buf, 0xcc, sizeof(buf));
 396   bufferlist bl;
 397   bl.append(buf, sizeof(buf));
 398
 399   const int max_objs = 16;
 400
 401   map<string, string> oid_to_ns;
 402
 403   for (int i=0; i<max_objs; ++i) {
 404     stringstream ss;
 405     ss << "ns" << i / 4;
 406     ioctx.set_namespace(ss.str());
 407     string oid = stringify(i);
 408     ASSERT_EQ(0, ioctx.write(oid, bl, bl.length(), 0));
 409
 410     oid_to_ns[oid] = ss.str();
 411   }
 412
 413   ioctx.set_namespace(all_nspaces);
 414
 415   librados::NObjectIterator it = ioctx.nobjects_begin();
 416   std::map<librados::ObjectCursor, string> cursor_to_obj;
 417
 418   int count = 0;
 419
 420   librados::ObjectCursor seek_cursor;
 421
 422   map<string, list<librados::ObjectCursor> > ns_to_cursors;
 423
 424   for (it = ioctx.nobjects_begin(); it != ioctx.nobjects_end(); ++it) {
 425     librados::ObjectCursor cursor = it.get_cursor();
 426     string oid = it->get_oid();
 427     cout << "> oid=" << oid << " cursor=" << it.get_cursor() << std::endl;
 428   }
 429
 430   vector<string> objs_order;
 431
 432   for (it = ioctx.nobjects_begin(); it != ioctx.nobjects_end(); ++it, ++count) {
 433     librados::ObjectCursor cursor = it.get_cursor();
 434     string oid = it->get_oid();
 435     std::cout << oid << " " << it.get_pg_hash_position() << std::endl;
 436     cout << ": oid=" << oid << " cursor=" << it.get_cursor() << std::endl;
 437     cursor_to_obj[cursor] = oid;
 438
 439     ASSERT_EQ(oid_to_ns[oid], it->get_nspace());
 440
 441     it.seek(cursor);
 442     cout << ": seek to " << cursor << " it.cursor=" << it.get_cursor() << std::endl;
 443     ASSERT_EQ(oid, it->get_oid());
 444     ASSERT_LT(count, max_objs); /* avoid infinite loops due to bad seek */
 445
 446     ns_to_cursors[it->get_nspace()].push_back(cursor);
 447
 448     if (count == max_objs/2) {
 449       seek_cursor = cursor;
 450     }
 451     objs_order.push_back(it->get_oid());
 452   }
 453
 454   ASSERT_EQ(count, max_objs);
 455
 456   /* check that reading past seek also works */
 457   cout << "seek_cursor=" << seek_cursor << std::endl;
 458   it.seek(seek_cursor);
 459   for (count = max_objs/2; count < max_objs; ++count, ++it) {
 460     ASSERT_EQ(objs_order[count], it->get_oid());
 461   }
 462
 463   /* seek to all cursors, check that we get expected obj */
 464   for (auto& niter : ns_to_cursors) {
 465     const string& ns = niter.first;
 466     list<librados::ObjectCursor>& cursors = niter.second;
 467
 468     for (auto& cursor : cursors) {
 469       cout << ": seek to " << cursor << std::endl;
 470       it.seek(cursor);
 471       ASSERT_EQ(cursor, it.get_cursor());
 472       string& expected_oid = cursor_to_obj[cursor];
 473       cout << ": it->get_cursor()=" << it.get_cursor() << " expected=" << cursor << std::endl;
 474       cout << ": it->get_oid()=" << it->get_oid() << " expected=" << expected_oid << std::endl;
 475       cout << ": it->get_nspace()=" << it->get_oid() << " expected=" << ns << std::endl;
 476       ASSERT_EQ(expected_oid, it->get_oid());
 477       ASSERT_EQ(it->get_nspace(), ns);
 478     }
 479   }
 480 }
 481
 482 TEST_F(LibRadosListPP, ListObjectsCursorPP) {
 483   char buf[128];
 484   memset(buf, 0xcc, sizeof(buf));
 485   bufferlist bl;
 486   bl.append(buf, sizeof(buf));
 487
 488   const int max_objs = 16;
 489
 490   for (int i=0; i<max_objs; ++i) {
 491     stringstream ss;
 492     ss << "ns" << i / 4;
 493     ioctx.set_namespace(ss.str());
 494     ASSERT_EQ(0, ioctx.write(stringify(i), bl, bl.length(), 0));
 495   }
 496
 497   ioctx.set_namespace(all_nspaces);
 498
 499   librados::NObjectIterator it = ioctx.nobjects_begin();
 500   std::map<librados::ObjectCursor, string> cursor_to_obj;
 501
 502   int count = 0;
 503
 504   for (; it != ioctx.nobjects_end(); ++it, ++count) {
 505     librados::ObjectCursor cursor = it.get_cursor();
 506     string oid = it->get_oid();
 507     std::cout << oid << " " << it.get_pg_hash_position() << std::endl;
 508     cout << ": oid=" << oid << " cursor=" << it.get_cursor() << std::endl;
 509     cursor_to_obj[cursor] = oid;
 510
 511     it.seek(cursor);
 512     cout << ": seek to " << cursor << std::endl;
 513     ASSERT_EQ(oid, it->get_oid());
 514     ASSERT_LT(count, max_objs); /* avoid infinite loops due to bad seek */
 515   }
 516
 517   ASSERT_EQ(count, max_objs);
 518
 519   auto p = cursor_to_obj.rbegin();
 520   it = ioctx.nobjects_begin();
 521   while (p != cursor_to_obj.rend()) {
 522     cout << ": seek to " << p->first << std::endl;
 523     it.seek(p->first);
 524     ASSERT_EQ(p->first, it.get_cursor());
 525     cout << ": it->get_cursor()=" << it.get_cursor() << " expected=" << p->first << std::endl;
 526     cout << ": it->get_oid()=" << it->get_oid() << " expected=" << p->second << std::endl;
 527     ASSERT_EQ(p->second, it->get_oid());
 528
 529     librados::NObjectIterator it2 = ioctx.nobjects_begin(it.get_cursor());
 530     ASSERT_EQ(it2->get_oid(), it->get_oid());
 531
 532     ++p;
 533   }
 534 }
 535
 536 TEST_F(LibRadosList, ListObjectsCursor) {
 537   char buf[128];
 538   memset(buf, 0xcc, sizeof(buf));
 539
 540   const int max_objs = 16;
 541
 542   for (int i=0; i<max_objs; ++i) {
 543     string n = stringify(i);
 544     ASSERT_EQ(0, rados_write(ioctx, n.c_str(), buf, sizeof(buf), 0));
 545   }
 546
 547   {
 548     rados_list_ctx_t ctx;
 549     const char *entry;
 550     rados_object_list_cursor cursor;
 551     ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 552     ASSERT_EQ(rados_nobjects_list_get_cursor(ctx, &cursor), 0);
 553     rados_object_list_cursor first_cursor = cursor;
 554     cout << "x cursor=" << ObjectCursor(cursor) << std::endl;
 555     while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) == 0) {
 556       string oid = entry;
 557       ASSERT_EQ(rados_nobjects_list_get_cursor(ctx, &cursor), 0);
 558       cout << "> oid=" << oid << " cursor=" << ObjectCursor(cursor) << std::endl;
 559     }
 560     rados_nobjects_list_seek_cursor(ctx, first_cursor);
 561     ASSERT_EQ(rados_nobjects_list_next(ctx, &entry, NULL, NULL), 0);
 562     cout << "FIRST> seek to " << ObjectCursor(first_cursor) << " oid=" << string(entry) << std::endl;
 563   }
 564   rados_list_ctx_t ctx;
 565   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 566
 567   std::map<rados_object_list_cursor, string> cursor_to_obj;
 568   int count = 0;
 569
 570   const char *entry;
 571   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) == 0) {
 572     rados_object_list_cursor cursor;
 573     ASSERT_EQ(rados_nobjects_list_get_cursor(ctx, &cursor), 0);
 574     string oid = entry;
 575     cout << ": oid=" << oid << " cursor=" << ObjectCursor(cursor) << std::endl;
 576     cursor_to_obj[cursor] = oid;
 577
 578     rados_nobjects_list_seek_cursor(ctx, cursor);
 579     cout << ": seek to " << ObjectCursor(cursor) << std::endl;
 580     ASSERT_EQ(rados_nobjects_list_next(ctx, &entry, NULL, NULL), 0);
 581     cout << "> " << ObjectCursor(cursor) << " -> " << entry << std::endl;
 582     ASSERT_EQ(string(entry), oid);
 583     ASSERT_LT(count, max_objs); /* avoid infinite loops due to bad seek */
 584
 585     ++count;
 586   }
 587
 588   ASSERT_EQ(count, max_objs);
 589
 590   auto p = cursor_to_obj.rbegin();
 591   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 592   while (p != cursor_to_obj.rend()) {
 593     cout << ": seek to " << ObjectCursor(p->first) << std::endl;
 594     rados_object_list_cursor cursor;
 595     rados_object_list_cursor oid(p->first);
 596     rados_nobjects_list_seek_cursor(ctx, oid);
 597     ASSERT_EQ(rados_nobjects_list_get_cursor(ctx, &cursor), 0);
 598     cout << ": cursor()=" << ObjectCursor(cursor) << " expected=" << oid << std::endl;
 599     // ASSERT_EQ(ObjectCursor(oid), ObjectCursor(cursor));
 600     ASSERT_EQ(rados_nobjects_list_next(ctx, &entry, NULL, NULL), 0);
 601     cout << "> " << ObjectCursor(cursor) << " -> " << entry << std::endl;
 602     cout << ": entry=" << entry << " expected=" << p->second << std::endl;
 603     ASSERT_EQ(p->second, string(entry));
 604
 605     ++p;
 606
 607     rados_object_list_cursor_free(ctx, cursor);
 608   }
 609 }
 610
 611 TEST_F(LibRadosListEC, ListObjects) {
 612   char buf[128];
 613   memset(buf, 0xcc, sizeof(buf));
 614   ASSERT_EQ(0, rados_write(ioctx, "foo", buf, sizeof(buf), 0));
 615   rados_list_ctx_t ctx;
 616   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 617   const char *entry;
 618   bool foundit = false;
 619   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) != -ENOENT) {
 620     foundit = true;
 621     ASSERT_EQ(std::string(entry), "foo");
 622   }
 623   ASSERT_TRUE(foundit);
 624   rados_nobjects_list_close(ctx);
 625 }
 626
 627 TEST_F(LibRadosListECPP, ListObjectsPP) {
 628   char buf[128];
 629   memset(buf, 0xcc, sizeof(buf));
 630   bufferlist bl1;
 631   bl1.append(buf, sizeof(buf));
 632   ASSERT_EQ(0, ioctx.write("foo", bl1, sizeof(buf), 0));
 633   NObjectIterator iter(ioctx.nobjects_begin());
 634   bool foundit = false;
 635   while (iter != ioctx.nobjects_end()) {
 636     foundit = true;
 637     ASSERT_EQ((*iter).get_oid(), "foo");
 638     ++iter;
 639   }
 640   ASSERT_TRUE(foundit);
 641 }
 642
 643 TEST_F(LibRadosListECPP, ListObjectsTwicePP) {
 644   char buf[128];
 645   memset(buf, 0xcc, sizeof(buf));
 646   bufferlist bl1;
 647   bl1.append(buf, sizeof(buf));
 648   ASSERT_EQ(0, ioctx.write("foo", bl1, sizeof(buf), 0));
 649   NObjectIterator iter(ioctx.nobjects_begin());
 650   bool foundit = false;
 651   while (iter != ioctx.nobjects_end()) {
 652     foundit = true;
 653     ASSERT_EQ((*iter).get_oid(), "foo");
 654     ++iter;
 655   }
 656   ASSERT_TRUE(foundit);
 657   ++iter;
 658   ASSERT_TRUE(iter == ioctx.nobjects_end());
 659   foundit = false;
 660   iter.seek(0);
 661   while (iter != ioctx.nobjects_end()) {
 662     foundit = true;
 663     ASSERT_EQ((*iter).get_oid(), "foo");
 664     ++iter;
 665   }
 666   ASSERT_TRUE(foundit);
 667 }
 668
 669 TEST_F(LibRadosListECPP, ListObjectsCopyIterPP) {
 670   char buf[128];
 671   memset(buf, 0xcc, sizeof(buf));
 672   bufferlist bl1;
 673   bl1.append(buf, sizeof(buf));
 674   ASSERT_EQ(0, ioctx.write("foo", bl1, sizeof(buf), 0));
 675
 676   // make sure this is still valid after the original iterators are gone
 677   NObjectIterator iter3;
 678   {
 679     NObjectIterator iter(ioctx.nobjects_begin());
 680     NObjectIterator iter2(iter);
 681     iter3 = iter2;
 682     ASSERT_EQ((*iter).get_oid(), "foo");
 683     ++iter;
 684     ASSERT_TRUE(iter == ioctx.nobjects_end());
 685     ++iter;
 686     ASSERT_TRUE(iter == ioctx.nobjects_end());
 687
 688     ASSERT_EQ(iter2->get_oid(), "foo");
 689     ASSERT_EQ(iter3->get_oid(), "foo");
 690     ++iter2;
 691     ASSERT_TRUE(iter2 == ioctx.nobjects_end());
 692   }
 693
 694   ASSERT_EQ(iter3->get_oid(), "foo");
 695   iter3 = iter3;
 696   ASSERT_EQ(iter3->get_oid(), "foo");
 697   ++iter3;
 698   ASSERT_TRUE(iter3 == ioctx.nobjects_end());
 699 }
 700
 701 TEST_F(LibRadosListECPP, ListObjectsEndIter) {
 702   char buf[128];
 703   memset(buf, 0xcc, sizeof(buf));
 704   bufferlist bl1;
 705   bl1.append(buf, sizeof(buf));
 706   ASSERT_EQ(0, ioctx.write("foo", bl1, sizeof(buf), 0));
 707
 708   NObjectIterator iter(ioctx.nobjects_begin());
 709   NObjectIterator iter_end(ioctx.nobjects_end());
 710   NObjectIterator iter_end2 = ioctx.nobjects_end();
 711   ASSERT_TRUE(iter_end == iter_end2);
 712   ASSERT_TRUE(iter_end == ioctx.nobjects_end());
 713   ASSERT_TRUE(iter_end2 == ioctx.nobjects_end());
 714
 715   ASSERT_EQ(iter->get_oid(), "foo");
 716   ++iter;
 717   ASSERT_TRUE(iter == ioctx.nobjects_end());
 718   ASSERT_TRUE(iter == iter_end);
 719   ASSERT_TRUE(iter == iter_end2);
 720   NObjectIterator iter2 = iter;
 721   ASSERT_TRUE(iter2 == ioctx.nobjects_end());
 722   ASSERT_TRUE(iter2 == iter_end);
 723   ASSERT_TRUE(iter2 == iter_end2);
 724 }
 725
 726 TEST_F(LibRadosListEC, ListObjectsNS) {
 727   char buf[128];
 728   memset(buf, 0xcc, sizeof(buf));
 729   // Create :foo1, :foo2, :foo3, n1:foo1, ns1:foo4, ns1:foo5, ns2:foo6, n2:foo7
 730   rados_ioctx_set_namespace(ioctx, "");
 731   ASSERT_EQ(0, rados_write(ioctx, "foo1", buf, sizeof(buf), 0));
 732   rados_ioctx_set_namespace(ioctx, "ns1");
 733   ASSERT_EQ(0, rados_write(ioctx, "foo1", buf, sizeof(buf), 0));
 734   rados_ioctx_set_namespace(ioctx, "");
 735   ASSERT_EQ(0, rados_write(ioctx, "foo2", buf, sizeof(buf), 0));
 736   ASSERT_EQ(0, rados_write(ioctx, "foo3", buf, sizeof(buf), 0));
 737   rados_ioctx_set_namespace(ioctx, "ns1");
 738   ASSERT_EQ(0, rados_write(ioctx, "foo4", buf, sizeof(buf), 0));
 739   ASSERT_EQ(0, rados_write(ioctx, "foo5", buf, sizeof(buf), 0));
 740   rados_ioctx_set_namespace(ioctx, "ns2");
 741   ASSERT_EQ(0, rados_write(ioctx, "foo6", buf, sizeof(buf), 0));
 742   ASSERT_EQ(0, rados_write(ioctx, "foo7", buf, sizeof(buf), 0));
 743
 744   std::set<std::string> def, ns1, ns2, all;
 745   def.insert(std::string("foo1"));
 746   def.insert(std::string("foo2"));
 747   def.insert(std::string("foo3"));
 748   ns1.insert(std::string("foo1"));
 749   ns1.insert(std::string("foo4"));
 750   ns1.insert(std::string("foo5"));
 751   ns2.insert(std::string("foo6"));
 752   ns2.insert(std::string("foo7"));
 753   all.insert(std::string(":foo1"));
 754   all.insert(std::string(":foo2"));
 755   all.insert(std::string(":foo3"));
 756   all.insert(std::string("ns1:foo1"));
 757   all.insert(std::string("ns1:foo4"));
 758   all.insert(std::string("ns1:foo5"));
 759   all.insert(std::string("ns2:foo6"));
 760   all.insert(std::string("ns2:foo7"));
 761
 762   rados_list_ctx_t ctx;
 763   // Check default namespace ""
 764   rados_ioctx_set_namespace(ioctx, "");
 765   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 766   check_list(def, ctx, "");
 767   rados_nobjects_list_close(ctx);
 768
 769   // Check default namespace "ns1"
 770   rados_ioctx_set_namespace(ioctx, "ns1");
 771   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 772   check_list(ns1, ctx, "ns1");
 773   rados_nobjects_list_close(ctx);
 774
 775   // Check default namespace "ns2"
 776   rados_ioctx_set_namespace(ioctx, "ns2");
 777   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 778   check_list(ns2, ctx, "ns2");
 779   rados_nobjects_list_close(ctx);
 780
 781   // Check all namespaces
 782   rados_ioctx_set_namespace(ioctx, LIBRADOS_ALL_NSPACES);
 783   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 784   check_list(all, ctx, all_nspaces);
 785   rados_nobjects_list_close(ctx);
 786 }
 787
 788 TEST_F(LibRadosListECPP, ListObjectsPPNS) {
 789   char buf[128];
 790   memset(buf, 0xcc, sizeof(buf));
 791   bufferlist bl1;
 792   bl1.append(buf, sizeof(buf));
 793   // Create :foo1, :foo2, :foo3, n1:foo1, ns1:foo4, ns1:foo5, ns2:foo6, n2:foo7
 794   ioctx.set_namespace("");
 795   ASSERT_EQ(0, ioctx.write("foo1", bl1, sizeof(buf), 0));
 796   ioctx.set_namespace("ns1");
 797   ASSERT_EQ(0, ioctx.write("foo1", bl1, sizeof(buf), 0));
 798   ioctx.set_namespace("");
 799   ASSERT_EQ(0, ioctx.write("foo2", bl1, sizeof(buf), 0));
 800   ASSERT_EQ(0, ioctx.write("foo3", bl1, sizeof(buf), 0));
 801   ioctx.set_namespace("ns1");
 802   ASSERT_EQ(0, ioctx.write("foo4", bl1, sizeof(buf), 0));
 803   ASSERT_EQ(0, ioctx.write("foo5", bl1, sizeof(buf), 0));
 804   ioctx.set_namespace("ns2");
 805   ASSERT_EQ(0, ioctx.write("foo6", bl1, sizeof(buf), 0));
 806   ASSERT_EQ(0, ioctx.write("foo7", bl1, sizeof(buf), 0));
 807
 808   std::set<std::string> def, ns1, ns2;
 809   def.insert(std::string("foo1"));
 810   def.insert(std::string("foo2"));
 811   def.insert(std::string("foo3"));
 812   ns1.insert(std::string("foo1"));
 813   ns1.insert(std::string("foo4"));
 814   ns1.insert(std::string("foo5"));
 815   ns2.insert(std::string("foo6"));
 816   ns2.insert(std::string("foo7"));
 817
 818   ioctx.set_namespace("");
 819   check_listpp(def, ioctx, "");
 820
 821   ioctx.set_namespace("ns1");
 822   check_listpp(ns1, ioctx, "ns1");
 823
 824   ioctx.set_namespace("ns2");
 825   check_listpp(ns2, ioctx, "ns2");
 826 }
 827
 828 TEST_F(LibRadosListECPP, ListObjectsManyPP) {
 829   char buf[128];
 830   memset(buf, 0xcc, sizeof(buf));
 831   bufferlist bl;
 832   bl.append(buf, sizeof(buf));
 833
 834   for (int i=0; i<256; ++i) {
 835     ASSERT_EQ(0, ioctx.write(stringify(i), bl, bl.length(), 0));
 836   }
 837
 838   librados::NObjectIterator it = ioctx.nobjects_begin();
 839   std::set<std::string> saw_obj;
 840   std::set<int> saw_pg;
 841   for (; it != ioctx.nobjects_end(); ++it) {
 842     std::cout << it->get_oid()
 843               << " " << it.get_pg_hash_position() << std::endl;
 844     saw_obj.insert(it->get_oid());
 845     saw_pg.insert(it.get_pg_hash_position());
 846   }
 847   std::cout << "saw " << saw_pg.size() << " pgs " << std::endl;
 848
 849   // make sure they are 0..n
 850   for (unsigned i = 0; i < saw_pg.size(); ++i)
 851     ASSERT_TRUE(saw_pg.count(i));
 852 }
 853
 854 TEST_F(LibRadosListEC, ListObjectsStart) {
 855   char buf[128];
 856   memset(buf, 0xcc, sizeof(buf));
 857
 858   for (int i=0; i<16; ++i) {
 859     string n = stringify(i);
 860     ASSERT_EQ(0, rados_write(ioctx, n.c_str(), buf, sizeof(buf), 0));
 861   }
 862
 863   rados_list_ctx_t ctx;
 864   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 865   std::map<int, std::set<std::string> > pg_to_obj;
 866   const char *entry;
 867   while (rados_nobjects_list_next(ctx, &entry, NULL, NULL) == 0) {
 868     uint32_t pos = rados_nobjects_list_get_pg_hash_position(ctx);
 869     std::cout << entry << " " << pos << std::endl;
 870     pg_to_obj[pos].insert(entry);
 871   }
 872   rados_nobjects_list_close(ctx);
 873
 874   std::map<int, std::set<std::string> >::reverse_iterator p =
 875     pg_to_obj.rbegin();
 876   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 877   while (p != pg_to_obj.rend()) {
 878     ASSERT_EQ((uint32_t)p->first, rados_nobjects_list_seek(ctx, p->first));
 879     ASSERT_EQ(0, rados_nobjects_list_next(ctx, &entry, NULL, NULL));
 880     std::cout << "have " << entry << " expect one of " << p->second << std::endl;
 881     ASSERT_TRUE(p->second.count(entry));
 882     ++p;
 883   }
 884   rados_nobjects_list_close(ctx);
 885 }
 886
 887 TEST_F(LibRadosListECPP, ListObjectsStartPP) {
 888   char buf[128];
 889   memset(buf, 0xcc, sizeof(buf));
 890   bufferlist bl;
 891   bl.append(buf, sizeof(buf));
 892
 893   for (int i=0; i<16; ++i) {
 894     ASSERT_EQ(0, ioctx.write(stringify(i), bl, bl.length(), 0));
 895   }
 896
 897   librados::NObjectIterator it = ioctx.nobjects_begin();
 898   std::map<int, std::set<std::string> > pg_to_obj;
 899   for (; it != ioctx.nobjects_end(); ++it) {
 900     std::cout << it->get_oid() << " " << it.get_pg_hash_position() << std::endl;
 901     pg_to_obj[it.get_pg_hash_position()].insert(it->get_oid());
 902   }
 903
 904   std::map<int, std::set<std::string> >::reverse_iterator p =
 905     pg_to_obj.rbegin();
 906   it = ioctx.nobjects_begin(p->first);
 907   while (p != pg_to_obj.rend()) {
 908     ASSERT_EQ((uint32_t)p->first, it.seek(p->first));
 909     std::cout << "have " << it->get_oid() << " expect one of " << p->second << std::endl;
 910     ASSERT_TRUE(p->second.count(it->get_oid()));
 911     ++p;
 912   }
 913 }
 914
 915 TEST_F(LibRadosListPP, ListObjectsFilterPP) {
 916   char buf[128];
 917   memset(buf, 0xcc, sizeof(buf));
 918   bufferlist obj_content;
 919   obj_content.append(buf, sizeof(buf));
 920
 921   std::string target_str = "content";
 922
 923   // Write xattr bare, no ::encod'ing
 924   bufferlist target_val;
 925   target_val.append(target_str);
 926   bufferlist nontarget_val;
 927   nontarget_val.append("rhubarb");
 928
 929   ASSERT_EQ(0, ioctx.write("has_xattr", obj_content, obj_content.length(), 0));
 930   ASSERT_EQ(0, ioctx.write("has_wrong_xattr", obj_content, obj_content.length(), 0));
 931   ASSERT_EQ(0, ioctx.write("no_xattr", obj_content, obj_content.length(), 0));
 932
 933   ASSERT_EQ(0, ioctx.setxattr("has_xattr", "theattr", target_val));
 934   ASSERT_EQ(0, ioctx.setxattr("has_wrong_xattr", "theattr", nontarget_val));
 935
 936   bufferlist filter_bl;
 937   std::string filter_name = "plain";
 938   ::encode(filter_name, filter_bl);
 939   ::encode("_theattr", filter_bl);
 940   ::encode(target_str, filter_bl);
 941
 942   NObjectIterator iter(ioctx.nobjects_begin(filter_bl));
 943   bool foundit = false;
 944   int k = 0;
 945   while (iter != ioctx.nobjects_end()) {
 946     foundit = true;
 947     // We should only see the object that matches the filter
 948     ASSERT_EQ((*iter).get_oid(), "has_xattr");
 949     // We should only see it once
 950     ASSERT_EQ(k, 0);
 951     ++iter;
 952     ++k;
 953   }
 954   ASSERT_TRUE(foundit);
 955 }
 956
 957 TEST_F(LibRadosListNP, ListObjectsError) {
 958   std::string pool_name;
 959   rados_t cluster;
 960   rados_ioctx_t ioctx;
 961   pool_name = get_temp_pool_name();
 962   ASSERT_EQ("", create_one_pool(pool_name, &cluster));
 963   ASSERT_EQ(0, rados_ioctx_create(cluster, pool_name.c_str(), &ioctx));
 964   char buf[128];
 965   memset(buf, 0xcc, sizeof(buf));
 966   rados_ioctx_set_namespace(ioctx, "");
 967   ASSERT_EQ(0, rados_write(ioctx, "foo", buf, sizeof(buf), 0));
 968
 969   //ASSERT_EQ(0, rados_pool_delete(cluster, pool_name.c_str()));
 970   {
 971     char *buf, *st;
 972     size_t buflen, stlen;
 973     string c = "{\"prefix\":\"osd pool rm\",\"pool\": \"" + pool_name +
 974       "\",\"pool2\":\"" + pool_name +
 975       "\",\"sure\": \"--yes-i-really-really-mean-it-not-faking\"}";
 976     const char *cmd[2] = { c.c_str(), 0 };
 977     ASSERT_EQ(0, rados_mon_command(cluster, (const char **)cmd, 1, "", 0, &buf, &buflen, &st, &stlen));
 978     ASSERT_EQ(0, rados_wait_for_latest_osdmap(cluster));
 979   }
 980
 981   rados_list_ctx_t ctx;
 982   ASSERT_EQ(0, rados_nobjects_list_open(ioctx, &ctx));
 983   const char *entry;
 984   ASSERT_EQ(-ENOENT, rados_nobjects_list_next(ctx, &entry, NULL, NULL));
 985   rados_nobjects_list_close(ctx);
 986   rados_ioctx_destroy(ioctx);
 987   rados_shutdown(cluster);
 988 }
 989
 990
 991
 992 // ---------------------------------------------
 993
 994 TEST_F(LibRadosList, EnumerateObjects) {
 995   char buf[128];
 996   memset(buf, 0xcc, sizeof(buf));
 997
 998   const uint32_t n_objects = 16;
 999   for (unsigned i=0; i<n_objects; ++i) {
1000     ASSERT_EQ(0, rados_write(ioctx, stringify(i).c_str(), buf, sizeof(buf), 0));
1001   }
1002
1003   // Ensure a non-power-of-two PG count to avoid only
1004   // touching the easy path.
1005   std::string err_str = set_pg_num(&s_cluster, pool_name, 11);
1006   ASSERT_TRUE(err_str.empty());
1007
1008   std::set<std::string> saw_obj;
1009   rados_object_list_cursor c = rados_object_list_begin(ioctx);
1010   rados_object_list_cursor end = rados_object_list_end(ioctx);
1011   while(!rados_object_list_is_end(ioctx, c))
1012   {
1013     rados_object_list_item results[12];
1014     memset(results, 0, sizeof(rados_object_list_item) * 12);
1015     rados_object_list_cursor temp_end = rados_object_list_end(ioctx);
1016     int r = rados_object_list(ioctx, c, temp_end,
1017             12, NULL, 0, results, &c);
1018     rados_object_list_cursor_free(ioctx, temp_end);
1019     ASSERT_GE(r, 0);
1020     for (int i = 0; i < r; ++i) {
1021       std::string oid(results[i].oid, results[i].oid_length);
1022       if (saw_obj.count(oid)) {
1023           std::cerr << "duplicate obj " << oid << std::endl;
1024       }
1025       ASSERT_FALSE(saw_obj.count(oid));
1026       saw_obj.insert(oid);
1027     }
1028     rados_object_list_free(12, results);
1029   }
1030   rados_object_list_cursor_free(ioctx, c);
1031   rados_object_list_cursor_free(ioctx, end);
1032
1033   for (unsigned i=0; i<n_objects; ++i) {
1034     if (!saw_obj.count(stringify(i))) {
1035         std::cerr << "missing object " << i << std::endl;
1036     }
1037     ASSERT_TRUE(saw_obj.count(stringify(i)));
1038   }
1039   ASSERT_EQ(n_objects, saw_obj.size());
1040 }
1041
1042 TEST_F(LibRadosList, EnumerateObjectsSplit) {
1043   char buf[128];
1044   memset(buf, 0xcc, sizeof(buf));
1045
1046   const uint32_t n_objects = 16;
1047   for (unsigned i=0; i<n_objects; ++i) {
1048     ASSERT_EQ(0, rados_write(ioctx, stringify(i).c_str(), buf, sizeof(buf), 0));
1049   }
1050
1051   // Ensure a non-power-of-two PG count to avoid only
1052   // touching the easy path.
1053   std::string err_str = set_pg_num(&s_cluster, pool_name, 11);
1054   ASSERT_TRUE(err_str.empty());
1055
1056   rados_object_list_cursor begin = rados_object_list_begin(ioctx);
1057   rados_object_list_cursor end = rados_object_list_end(ioctx);
1058
1059   // Step through an odd number of shards
1060   unsigned m = 5;
1061   std::set<std::string> saw_obj;
1062   for (unsigned n = 0; n < m; ++n) {
1063       rados_object_list_cursor shard_start = rados_object_list_begin(ioctx);;
1064       rados_object_list_cursor shard_end = rados_object_list_end(ioctx);;
1065
1066       rados_object_list_slice(
1067         ioctx,
1068         begin,
1069         end,
1070         n,
1071         m,
1072         &shard_start,
1073         &shard_end);
1074       std::cout << "split " << n << "/" << m << " -> "
1075                 << *(hobject_t*)shard_start << " "
1076                 << *(hobject_t*)shard_end << std::endl;
1077
1078       rados_object_list_cursor c = shard_start;
1079       //while(c < shard_end)
1080       while(rados_object_list_cursor_cmp(ioctx, c, shard_end) == -1)
1081       {
1082         rados_object_list_item results[12];
1083         memset(results, 0, sizeof(rados_object_list_item) * 12);
1084         int r = rados_object_list(ioctx,
1085                 c, shard_end,
1086                 12, NULL, 0, results, &c);
1087         ASSERT_GE(r, 0);
1088         for (int i = 0; i < r; ++i) {
1089           std::string oid(results[i].oid, results[i].oid_length);
1090           if (saw_obj.count(oid)) {
1091               std::cerr << "duplicate obj " << oid << std::endl;
1092           }
1093           ASSERT_FALSE(saw_obj.count(oid));
1094           saw_obj.insert(oid);
1095         }
1096         rados_object_list_free(12, results);
1097       }
1098       rados_object_list_cursor_free(ioctx, shard_start);
1099       rados_object_list_cursor_free(ioctx, shard_end);
1100   }
1101
1102   rados_object_list_cursor_free(ioctx, begin);
1103   rados_object_list_cursor_free(ioctx, end);
1104
1105   for (unsigned i=0; i<n_objects; ++i) {
1106     if (!saw_obj.count(stringify(i))) {
1107         std::cerr << "missing object " << i << std::endl;
1108     }
1109     ASSERT_TRUE(saw_obj.count(stringify(i)));
1110   }
1111   ASSERT_EQ(n_objects, saw_obj.size());
1112 }
1113
1114 TEST_F(LibRadosListPP, EnumerateObjectsPP) {
1115   char buf[128];
1116   memset(buf, 0xcc, sizeof(buf));
1117   bufferlist bl;
1118   bl.append(buf, sizeof(buf));
1119
1120   const uint32_t n_objects = 16;
1121   for (unsigned i=0; i<n_objects; ++i) {
1122     ASSERT_EQ(0, ioctx.write(stringify(i), bl, sizeof(buf), 0));
1123   }
1124
1125   std::set<std::string> saw_obj;
1126   ObjectCursor c = ioctx.object_list_begin();
1127   ObjectCursor end = ioctx.object_list_end();
1128   while(!ioctx.object_list_is_end(c))
1129   {
1130     std::vector<ObjectItem> result;
1131     int r = ioctx.object_list(c, end, 12, {}, &result, &c);
1132     ASSERT_GE(r, 0);
1133     ASSERT_EQ(r, (int)result.size());
1134     for (int i = 0; i < r; ++i) {
1135       auto oid = result[i].oid;
1136       if (saw_obj.count(oid)) {
1137           std::cerr << "duplicate obj " << oid << std::endl;
1138       }
1139       ASSERT_FALSE(saw_obj.count(oid));
1140       saw_obj.insert(oid);
1141     }
1142   }
1143
1144   for (unsigned i=0; i<n_objects; ++i) {
1145     if (!saw_obj.count(stringify(i))) {
1146         std::cerr << "missing object " << i << std::endl;
1147     }
1148     ASSERT_TRUE(saw_obj.count(stringify(i)));
1149   }
1150   ASSERT_EQ(n_objects, saw_obj.size());
1151 }
1152
1153 TEST_F(LibRadosListPP, EnumerateObjectsSplitPP) {
1154   char buf[128];
1155   memset(buf, 0xcc, sizeof(buf));
1156   bufferlist bl;
1157   bl.append(buf, sizeof(buf));
1158
1159   const uint32_t n_objects = 16;
1160   for (unsigned i=0; i<n_objects; ++i) {
1161     ASSERT_EQ(0, ioctx.write(stringify(i), bl, sizeof(buf), 0));
1162   }
1163
1164   ObjectCursor begin = ioctx.object_list_begin();
1165   ObjectCursor end = ioctx.object_list_end();
1166
1167   // Step through an odd number of shards
1168   unsigned m = 5;
1169   std::set<std::string> saw_obj;
1170   for (unsigned n = 0; n < m; ++n) {
1171       ObjectCursor shard_start;
1172       ObjectCursor shard_end;
1173
1174       ioctx.object_list_slice(
1175         begin,
1176         end,
1177         n,
1178         m,
1179         &shard_start,
1180         &shard_end);
1181
1182       ObjectCursor c(shard_start);
1183       while(c < shard_end)
1184       {
1185         std::vector<ObjectItem> result;
1186         int r = ioctx.object_list(c, shard_end, 12, {}, &result, &c);
1187         ASSERT_GE(r, 0);
1188
1189         for (const auto & i : result) {
1190           const auto &oid = i.oid;
1191           if (saw_obj.count(oid)) {
1192               std::cerr << "duplicate obj " << oid << std::endl;
1193           }
1194           ASSERT_FALSE(saw_obj.count(oid));
1195           saw_obj.insert(oid);
1196         }
1197       }
1198   }
1199
1200   for (unsigned i=0; i<n_objects; ++i) {
1201     if (!saw_obj.count(stringify(i))) {
1202         std::cerr << "missing object " << i << std::endl;
1203     }
1204     ASSERT_TRUE(saw_obj.count(stringify(i)));
1205   }
1206   ASSERT_EQ(n_objects, saw_obj.size());
1207 }
1208
1209
1210 TEST_F(LibRadosListPP, EnumerateObjectsFilterPP) {
1211   char buf[128];
1212   memset(buf, 0xcc, sizeof(buf));
1213   bufferlist obj_content;
1214   obj_content.append(buf, sizeof(buf));
1215
1216   std::string target_str = "content";
1217
1218   // Write xattr bare, no ::encod'ing
1219   bufferlist target_val;
1220   target_val.append(target_str);
1221   bufferlist nontarget_val;
1222   nontarget_val.append("rhubarb");
1223
1224   ASSERT_EQ(0, ioctx.write("has_xattr", obj_content, obj_content.length(), 0));
1225   ASSERT_EQ(0, ioctx.write("has_wrong_xattr", obj_content, obj_content.length(), 0));
1226   ASSERT_EQ(0, ioctx.write("no_xattr", obj_content, obj_content.length(), 0));
1227
1228   ASSERT_EQ(0, ioctx.setxattr("has_xattr", "theattr", target_val));
1229   ASSERT_EQ(0, ioctx.setxattr("has_wrong_xattr", "theattr", nontarget_val));
1230
1231   bufferlist filter_bl;
1232   std::string filter_name = "plain";
1233   ::encode(filter_name, filter_bl);
1234   ::encode("_theattr", filter_bl);
1235   ::encode(target_str, filter_bl);
1236
1237   ObjectCursor c = ioctx.object_list_begin();
1238   ObjectCursor end = ioctx.object_list_end();
1239   bool foundit = false;
1240   while(!ioctx.object_list_is_end(c))
1241   {
1242     std::vector<ObjectItem> result;
1243     int r = ioctx.object_list(c, end, 12, filter_bl, &result, &c);
1244     ASSERT_GE(r, 0);
1245     ASSERT_EQ(r, (int)result.size());
1246     for (int i = 0; i < r; ++i) {
1247       auto oid = result[i].oid;
1248       // We should only see the object that matches the filter
1249       ASSERT_EQ(oid, "has_xattr");
1250       // We should only see it once
1251       ASSERT_FALSE(foundit);
1252       foundit = true;
1253     }
1254   }
1255   ASSERT_TRUE(foundit);
1256 }