ceph/src/librados/librados.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2012 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #include <limits.h>
  16
  17 #include "common/config.h"
  18 #include "common/errno.h"
  19 #include "common/ceph_argparse.h"
  20 #include "common/ceph_json.h"
  21 #include "common/common_init.h"
  22 #include "common/TracepointProvider.h"
  23 #include "common/hobject.h"
  24 #include "include/rados/librados.h"
  25 #include "include/rados/librados.hpp"
  26 #include "include/types.h"
  27 #include <include/stringify.h>
  28
  29 #include "librados/AioCompletionImpl.h"
  30 #include "librados/IoCtxImpl.h"
  31 #include "librados/PoolAsyncCompletionImpl.h"
  32 #include "librados/RadosClient.h"
  33 #include "librados/RadosXattrIter.h"
  34 #include "librados/ListObjectImpl.h"
  35 #include <cls/lock/cls_lock_client.h>
  36
  37 #include <string>
  38 #include <map>
  39 #include <set>
  40 #include <vector>
  41 #include <list>
  42 #include <stdexcept>
  43
  44 #ifdef WITH_LTTNG
  45 #define TRACEPOINT_DEFINE
  46 #define TRACEPOINT_PROBE_DYNAMIC_LINKAGE
  47 #include "tracing/librados.h"
  48 #undef TRACEPOINT_PROBE_DYNAMIC_LINKAGE
  49 #undef TRACEPOINT_DEFINE
  50 #else
  51 #define tracepoint(...)
  52 #endif
  53
  54 using std::string;
  55 using std::map;
  56 using std::set;
  57 using std::vector;
  58 using std::list;
  59 using std::runtime_error;
  60
  61 #define dout_subsys ceph_subsys_rados
  62 #undef dout_prefix
  63 #define dout_prefix *_dout << "librados: "
  64
  65 #define RADOS_LIST_MAX_ENTRIES 1024
  66
  67 namespace {
  68
  69 TracepointProvider::Traits tracepoint_traits("librados_tp.so", "rados_tracing");
  70
  71 uint8_t get_checksum_op_type(rados_checksum_type_t type) {
  72   switch (type) {
  73   case LIBRADOS_CHECKSUM_TYPE_XXHASH32:
  74     return CEPH_OSD_CHECKSUM_OP_TYPE_XXHASH32;
  75   case LIBRADOS_CHECKSUM_TYPE_XXHASH64:
  76     return CEPH_OSD_CHECKSUM_OP_TYPE_XXHASH64;
  77   case LIBRADOS_CHECKSUM_TYPE_CRC32C:
  78     return CEPH_OSD_CHECKSUM_OP_TYPE_CRC32C;
  79   default:
  80     return -1;
  81   }
  82 }
  83
  84 } // anonymous namespace
  85
  86 /*
  87  * Structure of this file
  88  *
  89  * RadosClient and the related classes are the internal implementation of librados.
  90  * Above that layer sits the C API, found in include/rados/librados.h, and
  91  * the C++ API, found in include/rados/librados.hpp
  92  *
  93  * The C++ API sometimes implements things in terms of the C API.
  94  * Both the C++ and C API rely on RadosClient.
  95  *
  96  * Visually:
  97  * +--------------------------------------+
  98  * |             C++ API                  |
  99  * +--------------------+                 |
 100  * |       C API        |                 |
 101  * +--------------------+-----------------+
 102  * |          RadosClient                 |
 103  * +--------------------------------------+
 104  */
 105
 106 namespace librados {
 107
 108 struct ObjectOperationImpl {
 109   ::ObjectOperation o;
 110   real_time rt;
 111   real_time *prt;
 112
 113   ObjectOperationImpl() : prt(NULL) {}
 114 };
 115
 116 }
 117
 118 size_t librados::ObjectOperation::size()
 119 {
 120   ::ObjectOperation *o = &impl->o;
 121   return o->size();
 122 }
 123
 124 static void set_op_flags(::ObjectOperation *o, int flags)
 125 {
 126   int rados_flags = 0;
 127   if (flags & LIBRADOS_OP_FLAG_EXCL)
 128     rados_flags |= CEPH_OSD_OP_FLAG_EXCL;
 129   if (flags & LIBRADOS_OP_FLAG_FAILOK)
 130     rados_flags |= CEPH_OSD_OP_FLAG_FAILOK;
 131   if (flags & LIBRADOS_OP_FLAG_FADVISE_RANDOM)
 132     rados_flags |= CEPH_OSD_OP_FLAG_FADVISE_RANDOM;
 133   if (flags & LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL)
 134     rados_flags |= CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL;
 135   if (flags & LIBRADOS_OP_FLAG_FADVISE_WILLNEED)
 136     rados_flags |= CEPH_OSD_OP_FLAG_FADVISE_WILLNEED;
 137   if (flags & LIBRADOS_OP_FLAG_FADVISE_DONTNEED)
 138     rados_flags |= CEPH_OSD_OP_FLAG_FADVISE_DONTNEED;
 139   if (flags & LIBRADOS_OP_FLAG_FADVISE_NOCACHE)
 140     rados_flags |= CEPH_OSD_OP_FLAG_FADVISE_NOCACHE;
 141   o->set_last_op_flags(rados_flags);
 142 }
 143
 144 //deprcated
 145 void librados::ObjectOperation::set_op_flags(ObjectOperationFlags flags)
 146 {
 147   ::set_op_flags(&impl->o, (int)flags);
 148 }
 149
 150 void librados::ObjectOperation::set_op_flags2(int flags)
 151 {
 152   ::ObjectOperation *o = &impl->o;
 153   ::set_op_flags(o, flags);
 154 }
 155
 156 void librados::ObjectOperation::cmpext(uint64_t off,
 157                                        bufferlist &cmp_bl,
 158                                        int *prval)
 159 {
 160   ::ObjectOperation *o = &impl->o;
 161   o->cmpext(off, cmp_bl, prval);
 162 }
 163
 164 void librados::ObjectOperation::cmpxattr(const char *name, uint8_t op, const bufferlist& v)
 165 {
 166   ::ObjectOperation *o = &impl->o;
 167   o->cmpxattr(name, op, CEPH_OSD_CMPXATTR_MODE_STRING, v);
 168 }
 169
 170 void librados::ObjectOperation::cmpxattr(const char *name, uint8_t op, uint64_t v)
 171 {
 172   ::ObjectOperation *o = &impl->o;
 173   bufferlist bl;
 174   ::encode(v, bl);
 175   o->cmpxattr(name, op, CEPH_OSD_CMPXATTR_MODE_U64, bl);
 176 }
 177
 178 void librados::ObjectOperation::assert_version(uint64_t ver)
 179 {
 180   ::ObjectOperation *o = &impl->o;
 181   o->assert_version(ver);
 182 }
 183
 184 void librados::ObjectOperation::assert_exists()
 185 {
 186   ::ObjectOperation *o = &impl->o;
 187   o->stat(NULL, (ceph::real_time*) NULL, NULL);
 188 }
 189
 190 void librados::ObjectOperation::exec(const char *cls, const char *method, bufferlist& inbl)
 191 {
 192   ::ObjectOperation *o = &impl->o;
 193   o->call(cls, method, inbl);
 194 }
 195
 196 void librados::ObjectOperation::exec(const char *cls, const char *method, bufferlist& inbl, bufferlist *outbl, int *prval)
 197 {
 198   ::ObjectOperation *o = &impl->o;
 199   o->call(cls, method, inbl, outbl, NULL, prval);
 200 }
 201
 202 class ObjectOpCompletionCtx : public Context {
 203   librados::ObjectOperationCompletion *completion;
 204   bufferlist bl;
 205 public:
 206   explicit ObjectOpCompletionCtx(librados::ObjectOperationCompletion *c) : completion(c) {}
 207   void finish(int r) override {
 208     completion->handle_completion(r, bl);
 209     delete completion;
 210   }
 211
 212   bufferlist *outbl() {
 213     return &bl;
 214   }
 215 };
 216
 217 void librados::ObjectOperation::exec(const char *cls, const char *method, bufferlist& inbl, librados::ObjectOperationCompletion *completion)
 218 {
 219   ::ObjectOperation *o = &impl->o;
 220
 221   ObjectOpCompletionCtx *ctx = new ObjectOpCompletionCtx(completion);
 222
 223   o->call(cls, method, inbl, ctx->outbl(), ctx, NULL);
 224 }
 225
 226 void librados::ObjectReadOperation::stat(uint64_t *psize, time_t *pmtime, int *prval)
 227 {
 228   ::ObjectOperation *o = &impl->o;
 229   o->stat(psize, pmtime, prval);
 230 }
 231
 232 void librados::ObjectReadOperation::stat2(uint64_t *psize, struct timespec *pts, int *prval)
 233 {
 234   ::ObjectOperation *o = &impl->o;
 235   o->stat(psize, pts, prval);
 236 }
 237
 238 void librados::ObjectReadOperation::read(size_t off, uint64_t len, bufferlist *pbl, int *prval)
 239 {
 240   ::ObjectOperation *o = &impl->o;
 241   o->read(off, len, pbl, prval, NULL);
 242 }
 243
 244 void librados::ObjectReadOperation::sparse_read(uint64_t off, uint64_t len,
 245                                                 std::map<uint64_t,uint64_t> *m,
 246                                                 bufferlist *data_bl, int *prval)
 247 {
 248   ::ObjectOperation *o = &impl->o;
 249   o->sparse_read(off, len, m, data_bl, prval);
 250 }
 251
 252 void librados::ObjectReadOperation::checksum(rados_checksum_type_t type,
 253                                              const bufferlist &init_value_bl,
 254                                              uint64_t off, size_t len,
 255                                              size_t chunk_size, bufferlist *pbl,
 256                                              int *prval)
 257 {
 258   ::ObjectOperation *o = &impl->o;
 259   o->checksum(get_checksum_op_type(type), init_value_bl, off, len, chunk_size,
 260               pbl, prval, nullptr);
 261 }
 262
 263 void librados::ObjectReadOperation::tmap_get(bufferlist *pbl, int *prval)
 264 {
 265   ::ObjectOperation *o = &impl->o;
 266   o->tmap_get(pbl, prval);
 267 }
 268
 269 void librados::ObjectReadOperation::getxattr(const char *name, bufferlist *pbl, int *prval)
 270 {
 271   ::ObjectOperation *o = &impl->o;
 272   o->getxattr(name, pbl, prval);
 273 }
 274
 275 void librados::ObjectReadOperation::omap_get_vals(
 276   const std::string &start_after,
 277   const std::string &filter_prefix,
 278   uint64_t max_return,
 279   std::map<std::string, bufferlist> *out_vals,
 280   int *prval)
 281 {
 282   ::ObjectOperation *o = &impl->o;
 283   o->omap_get_vals(start_after, filter_prefix, max_return, out_vals, nullptr,
 284                    prval);
 285 }
 286
 287 void librados::ObjectReadOperation::omap_get_vals2(
 288   const std::string &start_after,
 289   const std::string &filter_prefix,
 290   uint64_t max_return,
 291   std::map<std::string, bufferlist> *out_vals,
 292   bool *pmore,
 293   int *prval)
 294 {
 295   ::ObjectOperation *o = &impl->o;
 296   o->omap_get_vals(start_after, filter_prefix, max_return, out_vals, pmore,
 297                    prval);
 298 }
 299
 300 void librados::ObjectReadOperation::omap_get_vals(
 301   const std::string &start_after,
 302   uint64_t max_return,
 303   std::map<std::string, bufferlist> *out_vals,
 304   int *prval)
 305 {
 306   ::ObjectOperation *o = &impl->o;
 307   o->omap_get_vals(start_after, "", max_return, out_vals, nullptr, prval);
 308 }
 309
 310 void librados::ObjectReadOperation::omap_get_vals2(
 311   const std::string &start_after,
 312   uint64_t max_return,
 313   std::map<std::string, bufferlist> *out_vals,
 314   bool *pmore,
 315   int *prval)
 316 {
 317   ::ObjectOperation *o = &impl->o;
 318   o->omap_get_vals(start_after, "", max_return, out_vals, pmore, prval);
 319 }
 320
 321 void librados::ObjectReadOperation::omap_get_keys(
 322   const std::string &start_after,
 323   uint64_t max_return,
 324   std::set<std::string> *out_keys,
 325   int *prval)
 326 {
 327   ::ObjectOperation *o = &impl->o;
 328   o->omap_get_keys(start_after, max_return, out_keys, nullptr, prval);
 329 }
 330
 331 void librados::ObjectReadOperation::omap_get_keys2(
 332   const std::string &start_after,
 333   uint64_t max_return,
 334   std::set<std::string> *out_keys,
 335   bool *pmore,
 336   int *prval)
 337 {
 338   ::ObjectOperation *o = &impl->o;
 339   o->omap_get_keys(start_after, max_return, out_keys, pmore, prval);
 340 }
 341
 342 void librados::ObjectReadOperation::omap_get_header(bufferlist *bl, int *prval)
 343 {
 344   ::ObjectOperation *o = &impl->o;
 345   o->omap_get_header(bl, prval);
 346 }
 347
 348 void librados::ObjectReadOperation::omap_get_vals_by_keys(
 349   const std::set<std::string> &keys,
 350   std::map<std::string, bufferlist> *map,
 351   int *prval)
 352 {
 353   ::ObjectOperation *o = &impl->o;
 354   o->omap_get_vals_by_keys(keys, map, prval);
 355 }
 356
 357 void librados::ObjectOperation::omap_cmp(
 358   const std::map<std::string, pair<bufferlist, int> > &assertions,
 359   int *prval)
 360 {
 361   ::ObjectOperation *o = &impl->o;
 362   o->omap_cmp(assertions, prval);
 363 }
 364
 365 void librados::ObjectReadOperation::list_watchers(
 366   list<obj_watch_t> *out_watchers,
 367   int *prval)
 368 {
 369   ::ObjectOperation *o = &impl->o;
 370   o->list_watchers(out_watchers, prval);
 371 }
 372
 373 void librados::ObjectReadOperation::list_snaps(
 374   snap_set_t *out_snaps,
 375   int *prval)
 376 {
 377   ::ObjectOperation *o = &impl->o;
 378   o->list_snaps(out_snaps, prval);
 379 }
 380
 381 void librados::ObjectReadOperation::is_dirty(bool *is_dirty, int *prval)
 382 {
 383   ::ObjectOperation *o = &impl->o;
 384   o->is_dirty(is_dirty, prval);
 385 }
 386
 387 int librados::IoCtx::omap_get_vals(const std::string& oid,
 388                                    const std::string& orig_start_after,
 389                                    const std::string& filter_prefix,
 390                                    uint64_t max_return,
 391                                    std::map<std::string, bufferlist> *out_vals)
 392 {
 393   bool first = true;
 394   string start_after = orig_start_after;
 395   bool more = true;
 396   while (max_return > 0 && more) {
 397     std::map<std::string,bufferlist> out;
 398     ObjectReadOperation op;
 399     op.omap_get_vals2(start_after, filter_prefix, max_return, &out, &more,
 400                       nullptr);
 401     bufferlist bl;
 402     int ret = operate(oid, &op, &bl);
 403     if (ret < 0) {
 404       return ret;
 405     }
 406     if (more) {
 407       if (out.empty()) {
 408         return -EINVAL;  // wth
 409       }
 410       start_after = out.rbegin()->first;
 411     }
 412     if (out.size() <= max_return) {
 413       max_return -= out.size();
 414     } else {
 415       max_return = 0;
 416     }
 417     if (first) {
 418       out_vals->swap(out);
 419       first = false;
 420     } else {
 421       out_vals->insert(out.begin(), out.end());
 422       out.clear();
 423     }
 424   }
 425   return 0;
 426 }
 427
 428 int librados::IoCtx::omap_get_vals2(
 429   const std::string& oid,
 430   const std::string& start_after,
 431   const std::string& filter_prefix,
 432   uint64_t max_return,
 433   std::map<std::string, bufferlist> *out_vals,
 434   bool *pmore)
 435 {
 436   ObjectReadOperation op;
 437   int r;
 438   op.omap_get_vals2(start_after, filter_prefix, max_return, out_vals, pmore, &r);
 439   bufferlist bl;
 440   int ret = operate(oid, &op, &bl);
 441   if (ret < 0)
 442     return ret;
 443   return r;
 444 }
 445
 446 void librados::ObjectReadOperation::getxattrs(map<string, bufferlist> *pattrs, int *prval)
 447 {
 448   ::ObjectOperation *o = &impl->o;
 449   o->getxattrs(pattrs, prval);
 450 }
 451
 452 void librados::ObjectWriteOperation::mtime(time_t *pt)
 453 {
 454   if (pt) {
 455     impl->rt = ceph::real_clock::from_time_t(*pt);
 456     impl->prt = &impl->rt;
 457   }
 458 }
 459
 460 void librados::ObjectWriteOperation::mtime2(struct timespec *pts)
 461 {
 462   if (pts) {
 463     impl->rt = ceph::real_clock::from_timespec(*pts);
 464     impl->prt = &impl->rt;
 465   }
 466 }
 467
 468 void librados::ObjectWriteOperation::create(bool exclusive)
 469 {
 470   ::ObjectOperation *o = &impl->o;
 471   o->create(exclusive);
 472 }
 473
 474 void librados::ObjectWriteOperation::create(bool exclusive,
 475                                             const std::string& category) // unused
 476 {
 477   ::ObjectOperation *o = &impl->o;
 478   o->create(exclusive);
 479 }
 480
 481 void librados::ObjectWriteOperation::write(uint64_t off, const bufferlist& bl)
 482 {
 483   ::ObjectOperation *o = &impl->o;
 484   bufferlist c = bl;
 485   o->write(off, c);
 486 }
 487
 488 void librados::ObjectWriteOperation::write_full(const bufferlist& bl)
 489 {
 490   ::ObjectOperation *o = &impl->o;
 491   bufferlist c = bl;
 492   o->write_full(c);
 493 }
 494
 495 void librados::ObjectWriteOperation::writesame(uint64_t off, uint64_t write_len,
 496                                                const bufferlist& bl)
 497 {
 498   ::ObjectOperation *o = &impl->o;
 499   bufferlist c = bl;
 500   o->writesame(off, write_len, c);
 501 }
 502
 503 void librados::ObjectWriteOperation::append(const bufferlist& bl)
 504 {
 505   ::ObjectOperation *o = &impl->o;
 506   bufferlist c = bl;
 507   o->append(c);
 508 }
 509
 510 void librados::ObjectWriteOperation::remove()
 511 {
 512   ::ObjectOperation *o = &impl->o;
 513   o->remove();
 514 }
 515
 516 void librados::ObjectWriteOperation::truncate(uint64_t off)
 517 {
 518   ::ObjectOperation *o = &impl->o;
 519   o->truncate(off);
 520 }
 521
 522 void librados::ObjectWriteOperation::zero(uint64_t off, uint64_t len)
 523 {
 524   ::ObjectOperation *o = &impl->o;
 525   o->zero(off, len);
 526 }
 527
 528 void librados::ObjectWriteOperation::rmxattr(const char *name)
 529 {
 530   ::ObjectOperation *o = &impl->o;
 531   o->rmxattr(name);
 532 }
 533
 534 void librados::ObjectWriteOperation::setxattr(const char *name, const bufferlist& v)
 535 {
 536   ::ObjectOperation *o = &impl->o;
 537   o->setxattr(name, v);
 538 }
 539
 540 void librados::ObjectWriteOperation::omap_set(
 541   const map<string, bufferlist> &map)
 542 {
 543   ::ObjectOperation *o = &impl->o;
 544   o->omap_set(map);
 545 }
 546
 547 void librados::ObjectWriteOperation::omap_set_header(const bufferlist &bl)
 548 {
 549   bufferlist c = bl;
 550   ::ObjectOperation *o = &impl->o;
 551   o->omap_set_header(c);
 552 }
 553
 554 void librados::ObjectWriteOperation::omap_clear()
 555 {
 556   ::ObjectOperation *o = &impl->o;
 557   o->omap_clear();
 558 }
 559
 560 void librados::ObjectWriteOperation::omap_rm_keys(
 561   const std::set<std::string> &to_rm)
 562 {
 563   ::ObjectOperation *o = &impl->o;
 564   o->omap_rm_keys(to_rm);
 565 }
 566
 567 void librados::ObjectWriteOperation::copy_from(const std::string& src,
 568                                                const IoCtx& src_ioctx,
 569                                                uint64_t src_version)
 570 {
 571   copy_from2(src, src_ioctx, src_version, 0);
 572 }
 573
 574 void librados::ObjectWriteOperation::copy_from2(const std::string& src,
 575                                                 const IoCtx& src_ioctx,
 576                                                 uint64_t src_version,
 577                                                 uint32_t src_fadvise_flags)
 578 {
 579   ::ObjectOperation *o = &impl->o;
 580   o->copy_from(object_t(src), src_ioctx.io_ctx_impl->snap_seq,
 581                src_ioctx.io_ctx_impl->oloc, src_version, 0, src_fadvise_flags);
 582 }
 583
 584 void librados::ObjectWriteOperation::undirty()
 585 {
 586   ::ObjectOperation *o = &impl->o;
 587   o->undirty();
 588 }
 589
 590 void librados::ObjectReadOperation::cache_flush()
 591 {
 592   ::ObjectOperation *o = &impl->o;
 593   o->cache_flush();
 594 }
 595
 596 void librados::ObjectReadOperation::cache_try_flush()
 597 {
 598   ::ObjectOperation *o = &impl->o;
 599   o->cache_try_flush();
 600 }
 601
 602 void librados::ObjectReadOperation::cache_evict()
 603 {
 604   ::ObjectOperation *o = &impl->o;
 605   o->cache_evict();
 606 }
 607
 608 void librados::ObjectWriteOperation::tmap_put(const bufferlist &bl)
 609 {
 610   ::ObjectOperation *o = &impl->o;
 611   bufferlist c = bl;
 612   o->tmap_put(c);
 613 }
 614
 615 void librados::ObjectWriteOperation::tmap_update(const bufferlist& cmdbl)
 616 {
 617   ::ObjectOperation *o = &impl->o;
 618   bufferlist c = cmdbl;
 619   o->tmap_update(c);
 620 }
 621
 622 void librados::ObjectWriteOperation::selfmanaged_snap_rollback(snap_t snapid)
 623 {
 624   ::ObjectOperation *o = &impl->o;
 625   o->rollback(snapid);
 626 }
 627
 628 // You must specify the snapid not the name normally used with pool snapshots
 629 void librados::ObjectWriteOperation::snap_rollback(snap_t snapid)
 630 {
 631   ::ObjectOperation *o = &impl->o;
 632   o->rollback(snapid);
 633 }
 634
 635 void librados::ObjectWriteOperation::set_alloc_hint(
 636                                             uint64_t expected_object_size,
 637                                             uint64_t expected_write_size)
 638 {
 639   ::ObjectOperation *o = &impl->o;
 640   o->set_alloc_hint(expected_object_size, expected_write_size, 0);
 641 }
 642 void librados::ObjectWriteOperation::set_alloc_hint2(
 643                                             uint64_t expected_object_size,
 644                                             uint64_t expected_write_size,
 645                                             uint32_t flags)
 646 {
 647   ::ObjectOperation *o = &impl->o;
 648   o->set_alloc_hint(expected_object_size, expected_write_size, flags);
 649 }
 650
 651 void librados::ObjectWriteOperation::cache_pin()
 652 {
 653   ::ObjectOperation *o = &impl->o;
 654   o->cache_pin();
 655 }
 656
 657 void librados::ObjectWriteOperation::cache_unpin()
 658 {
 659   ::ObjectOperation *o = &impl->o;
 660   o->cache_unpin();
 661 }
 662
 663 librados::WatchCtx::
 664 ~WatchCtx()
 665 {
 666 }
 667
 668 librados::WatchCtx2::
 669 ~WatchCtx2()
 670 {
 671 }
 672
 673
 674 struct librados::ObjListCtx {
 675   librados::IoCtxImpl dupctx;
 676   librados::IoCtxImpl *ctx;
 677   Objecter::NListContext *nlc;
 678   bool legacy_list_api;
 679
 680   ObjListCtx(IoCtxImpl *c, Objecter::NListContext *nl, bool legacy=false)
 681     : nlc(nl),
 682       legacy_list_api(legacy) {
 683     // Get our own private IoCtxImpl so that namespace setting isn't
 684     // changed by caller between uses.
 685     ctx = &dupctx;
 686     dupctx.dup(*c);
 687   }
 688   ~ObjListCtx() {
 689     ctx = NULL;
 690     delete nlc;
 691   }
 692 };
 693
 694 ///////////////////////////// NObjectIteratorImpl /////////////////////////////
 695 librados::NObjectIteratorImpl::NObjectIteratorImpl(ObjListCtx *ctx_)
 696   : ctx(ctx_)
 697 {
 698 }
 699
 700 librados::NObjectIteratorImpl::~NObjectIteratorImpl()
 701 {
 702   ctx.reset();
 703 }
 704
 705 librados::NObjectIteratorImpl::NObjectIteratorImpl(const NObjectIteratorImpl &rhs)
 706 {
 707   *this = rhs;
 708 }
 709
 710 librados::NObjectIteratorImpl& librados::NObjectIteratorImpl::operator=(const librados::NObjectIteratorImpl &rhs)
 711 {
 712   if (&rhs == this)
 713     return *this;
 714   if (rhs.ctx.get() == NULL) {
 715     ctx.reset();
 716     return *this;
 717   }
 718   Objecter::NListContext *list_ctx = new Objecter::NListContext(*rhs.ctx->nlc);
 719   ctx.reset(new ObjListCtx(rhs.ctx->ctx, list_ctx));
 720   cur_obj = rhs.cur_obj;
 721   return *this;
 722 }
 723
 724 bool librados::NObjectIteratorImpl::operator==(const librados::NObjectIteratorImpl& rhs) const {
 725
 726   if (ctx.get() == NULL) {
 727     if (rhs.ctx.get() == NULL)
 728       return true;
 729     return rhs.ctx->nlc->at_end();
 730   }
 731   if (rhs.ctx.get() == NULL) {
 732     // Redundant but same as ObjectIterator version
 733     if (ctx.get() == NULL)
 734       return true;
 735     return ctx->nlc->at_end();
 736   }
 737   return ctx.get() == rhs.ctx.get();
 738 }
 739
 740 bool librados::NObjectIteratorImpl::operator!=(const librados::NObjectIteratorImpl& rhs) const {
 741   return !(*this == rhs);
 742 }
 743
 744 const librados::ListObject& librados::NObjectIteratorImpl::operator*() const {
 745   return cur_obj;
 746 }
 747
 748 const librados::ListObject* librados::NObjectIteratorImpl::operator->() const {
 749   return &cur_obj;
 750 }
 751
 752 librados::NObjectIteratorImpl& librados::NObjectIteratorImpl::operator++()
 753 {
 754   get_next();
 755   return *this;
 756 }
 757
 758 librados::NObjectIteratorImpl librados::NObjectIteratorImpl::operator++(int)
 759 {
 760   librados::NObjectIteratorImpl ret(*this);
 761   get_next();
 762   return ret;
 763 }
 764
 765 uint32_t librados::NObjectIteratorImpl::seek(uint32_t pos)
 766 {
 767   uint32_t r = rados_nobjects_list_seek(ctx.get(), pos);
 768   get_next();
 769   return r;
 770 }
 771
 772 uint32_t librados::NObjectIteratorImpl::seek(const ObjectCursor& cursor)
 773 {
 774   uint32_t r = rados_nobjects_list_seek_cursor(ctx.get(), (rados_object_list_cursor)cursor.c_cursor);
 775   get_next();
 776   return r;
 777 }
 778
 779 librados::ObjectCursor librados::NObjectIteratorImpl::get_cursor()
 780 {
 781   librados::ObjListCtx *lh = (librados::ObjListCtx *)ctx.get();
 782   librados::ObjectCursor oc;
 783   oc.set(lh->ctx->nlist_get_cursor(lh->nlc));
 784   return oc;
 785 }
 786
 787 void librados::NObjectIteratorImpl::set_filter(const bufferlist &bl)
 788 {
 789   assert(ctx);
 790   ctx->nlc->filter = bl;
 791 }
 792
 793 void librados::NObjectIteratorImpl::get_next()
 794 {
 795   const char *entry, *key, *nspace;
 796   if (ctx->nlc->at_end())
 797     return;
 798   int ret = rados_nobjects_list_next(ctx.get(), &entry, &key, &nspace);
 799   if (ret == -ENOENT) {
 800     return;
 801   }
 802   else if (ret) {
 803     ostringstream oss;
 804     oss << "rados returned " << cpp_strerror(ret);
 805     throw std::runtime_error(oss.str());
 806   }
 807
 808   if (cur_obj.impl == NULL)
 809     cur_obj.impl = new ListObjectImpl();
 810   cur_obj.impl->nspace = nspace;
 811   cur_obj.impl->oid = entry;
 812   cur_obj.impl->locator = key ? key : string();
 813 }
 814
 815 uint32_t librados::NObjectIteratorImpl::get_pg_hash_position() const
 816 {
 817   return ctx->nlc->get_pg_hash_position();
 818 }
 819
 820 ///////////////////////////// NObjectIterator /////////////////////////////
 821 librados::NObjectIterator::NObjectIterator(ObjListCtx *ctx_)
 822 {
 823   impl = new NObjectIteratorImpl(ctx_);
 824 }
 825
 826 librados::NObjectIterator::~NObjectIterator()
 827 {
 828   delete impl;
 829 }
 830
 831 librados::NObjectIterator::NObjectIterator(const NObjectIterator &rhs)
 832 {
 833   if (rhs.impl == NULL) {
 834     impl = NULL;
 835     return;
 836   }
 837   impl = new NObjectIteratorImpl();
 838   *impl = *(rhs.impl);
 839 }
 840
 841 librados::NObjectIterator& librados::NObjectIterator::operator=(const librados::NObjectIterator &rhs)
 842 {
 843   if (rhs.impl == NULL) {
 844     delete impl;
 845     impl = NULL;
 846     return *this;
 847   }
 848   if (impl == NULL)
 849     impl = new NObjectIteratorImpl();
 850   *impl = *(rhs.impl);
 851   return *this;
 852 }
 853
 854 bool librados::NObjectIterator::operator==(const librados::NObjectIterator& rhs) const
 855 {
 856   if (impl && rhs.impl) {
 857     return *impl == *(rhs.impl);
 858   } else {
 859     return impl == rhs.impl;
 860   }
 861 }
 862
 863 bool librados::NObjectIterator::operator!=(const librados::NObjectIterator& rhs) const
 864 {
 865   return !(*this == rhs);
 866 }
 867
 868 const librados::ListObject& librados::NObjectIterator::operator*() const {
 869   assert(impl);
 870   return *(impl->get_listobjectp());
 871 }
 872
 873 const librados::ListObject* librados::NObjectIterator::operator->() const {
 874   assert(impl);
 875   return impl->get_listobjectp();
 876 }
 877
 878 librados::NObjectIterator& librados::NObjectIterator::operator++()
 879 {
 880   assert(impl);
 881   impl->get_next();
 882   return *this;
 883 }
 884
 885 librados::NObjectIterator librados::NObjectIterator::operator++(int)
 886 {
 887   librados::NObjectIterator ret(*this);
 888   impl->get_next();
 889   return ret;
 890 }
 891
 892 uint32_t librados::NObjectIterator::seek(uint32_t pos)
 893 {
 894   assert(impl);
 895   return impl->seek(pos);
 896 }
 897
 898 uint32_t librados::NObjectIterator::seek(const ObjectCursor& cursor)
 899 {
 900   assert(impl);
 901   return impl->seek(cursor);
 902 }
 903
 904 librados::ObjectCursor librados::NObjectIterator::get_cursor()
 905 {
 906   assert(impl);
 907   return impl->get_cursor();
 908 }
 909
 910 void librados::NObjectIterator::set_filter(const bufferlist &bl)
 911 {
 912   impl->set_filter(bl);
 913 }
 914
 915 void librados::NObjectIterator::get_next()
 916 {
 917   assert(impl);
 918   impl->get_next();
 919 }
 920
 921 uint32_t librados::NObjectIterator::get_pg_hash_position() const
 922 {
 923   assert(impl);
 924   return impl->get_pg_hash_position();
 925 }
 926
 927 const librados::NObjectIterator librados::NObjectIterator::__EndObjectIterator(NULL);
 928
 929 ///////////////////////////// PoolAsyncCompletion //////////////////////////////
 930 int librados::PoolAsyncCompletion::PoolAsyncCompletion::set_callback(void *cb_arg,
 931                                                                      rados_callback_t cb)
 932 {
 933   PoolAsyncCompletionImpl *c = (PoolAsyncCompletionImpl *)pc;
 934   return c->set_callback(cb_arg, cb);
 935 }
 936
 937 int librados::PoolAsyncCompletion::PoolAsyncCompletion::wait()
 938 {
 939   PoolAsyncCompletionImpl *c = (PoolAsyncCompletionImpl *)pc;
 940   return c->wait();
 941 }
 942
 943 bool librados::PoolAsyncCompletion::PoolAsyncCompletion::is_complete()
 944 {
 945   PoolAsyncCompletionImpl *c = (PoolAsyncCompletionImpl *)pc;
 946   return c->is_complete();
 947 }
 948
 949 int librados::PoolAsyncCompletion::PoolAsyncCompletion::get_return_value()
 950 {
 951   PoolAsyncCompletionImpl *c = (PoolAsyncCompletionImpl *)pc;
 952   return c->get_return_value();
 953 }
 954
 955 void librados::PoolAsyncCompletion::PoolAsyncCompletion::release()
 956 {
 957   PoolAsyncCompletionImpl *c = (PoolAsyncCompletionImpl *)pc;
 958   c->release();
 959   delete this;
 960 }
 961
 962 ///////////////////////////// AioCompletion //////////////////////////////
 963 int librados::AioCompletion::AioCompletion::set_complete_callback(void *cb_arg, rados_callback_t cb)
 964 {
 965   AioCompletionImpl *c = (AioCompletionImpl *)pc;
 966   return c->set_complete_callback(cb_arg, cb);
 967 }
 968
 969 int librados::AioCompletion::AioCompletion::set_safe_callback(void *cb_arg, rados_callback_t cb)
 970 {
 971   AioCompletionImpl *c = (AioCompletionImpl *)pc;
 972   return c->set_safe_callback(cb_arg, cb);
 973 }
 974
 975 int librados::AioCompletion::AioCompletion::wait_for_complete()
 976 {
 977   AioCompletionImpl *c = (AioCompletionImpl *)pc;
 978   return c->wait_for_complete();
 979 }
 980
 981 int librados::AioCompletion::AioCompletion::wait_for_safe()
 982 {
 983   AioCompletionImpl *c = (AioCompletionImpl *)pc;
 984   return c->wait_for_safe();
 985 }
 986
 987 bool librados::AioCompletion::AioCompletion::is_complete()
 988 {
 989   AioCompletionImpl *c = (AioCompletionImpl *)pc;
 990   return c->is_complete();
 991 }
 992
 993 bool librados::AioCompletion::AioCompletion::is_safe()
 994 {
 995   AioCompletionImpl *c = (AioCompletionImpl *)pc;
 996   return c->is_safe();
 997 }
 998
 999 int librados::AioCompletion::AioCompletion::wait_for_complete_and_cb()
1000 {
1001   AioCompletionImpl *c = (AioCompletionImpl *)pc;
1002   return c->wait_for_complete_and_cb();
1003 }
1004
1005 int librados::AioCompletion::AioCompletion::wait_for_safe_and_cb()
1006 {
1007   AioCompletionImpl *c = (AioCompletionImpl *)pc;
1008   return c->wait_for_safe_and_cb();
1009 }
1010
1011 bool librados::AioCompletion::AioCompletion::is_complete_and_cb()
1012 {
1013   AioCompletionImpl *c = (AioCompletionImpl *)pc;
1014   return c->is_complete_and_cb();
1015 }
1016
1017 bool librados::AioCompletion::AioCompletion::is_safe_and_cb()
1018 {
1019   AioCompletionImpl *c = (AioCompletionImpl *)pc;
1020   return c->is_safe_and_cb();
1021 }
1022
1023 int librados::AioCompletion::AioCompletion::get_return_value()
1024 {
1025   AioCompletionImpl *c = (AioCompletionImpl *)pc;
1026   return c->get_return_value();
1027 }
1028
1029 int librados::AioCompletion::AioCompletion::get_version()
1030 {
1031   AioCompletionImpl *c = (AioCompletionImpl *)pc;
1032   return c->get_version();
1033 }
1034
1035 uint64_t librados::AioCompletion::AioCompletion::get_version64()
1036 {
1037   AioCompletionImpl *c = (AioCompletionImpl *)pc;
1038   return c->get_version();
1039 }
1040
1041 void librados::AioCompletion::AioCompletion::release()
1042 {
1043   AioCompletionImpl *c = (AioCompletionImpl *)pc;
1044   c->release();
1045   delete this;
1046 }
1047
1048 ///////////////////////////// IoCtx //////////////////////////////
1049 librados::IoCtx::IoCtx() : io_ctx_impl(NULL)
1050 {
1051 }
1052
1053 void librados::IoCtx::from_rados_ioctx_t(rados_ioctx_t p, IoCtx &io)
1054 {
1055   IoCtxImpl *io_ctx_impl = (IoCtxImpl*)p;
1056
1057   io.io_ctx_impl = io_ctx_impl;
1058   if (io_ctx_impl) {
1059     io_ctx_impl->get();
1060   }
1061 }
1062
1063 librados::IoCtx::IoCtx(const IoCtx& rhs)
1064 {
1065   io_ctx_impl = rhs.io_ctx_impl;
1066   if (io_ctx_impl) {
1067     io_ctx_impl->get();
1068   }
1069 }
1070
1071 librados::IoCtx& librados::IoCtx::operator=(const IoCtx& rhs)
1072 {
1073   if (io_ctx_impl)
1074     io_ctx_impl->put();
1075   io_ctx_impl = rhs.io_ctx_impl;
1076   io_ctx_impl->get();
1077   return *this;
1078 }
1079
1080 librados::IoCtx::~IoCtx()
1081 {
1082   close();
1083 }
1084
1085 void librados::IoCtx::close()
1086 {
1087   if (io_ctx_impl)
1088     io_ctx_impl->put();
1089   io_ctx_impl = 0;
1090 }
1091
1092 void librados::IoCtx::dup(const IoCtx& rhs)
1093 {
1094   if (io_ctx_impl)
1095     io_ctx_impl->put();
1096   io_ctx_impl = new IoCtxImpl();
1097   io_ctx_impl->get();
1098   io_ctx_impl->dup(*rhs.io_ctx_impl);
1099 }
1100
1101 int librados::IoCtx::set_auid(uint64_t auid_)
1102 {
1103   return io_ctx_impl->pool_change_auid(auid_);
1104 }
1105
1106 int librados::IoCtx::set_auid_async(uint64_t auid_, PoolAsyncCompletion *c)
1107 {
1108   return io_ctx_impl->pool_change_auid_async(auid_, c->pc);
1109 }
1110
1111 int librados::IoCtx::get_auid(uint64_t *auid_)
1112 {
1113   return rados_ioctx_pool_get_auid(io_ctx_impl, auid_);
1114 }
1115
1116 bool librados::IoCtx::pool_requires_alignment()
1117 {
1118   return io_ctx_impl->client->pool_requires_alignment(get_id());
1119 }
1120
1121 int librados::IoCtx::pool_requires_alignment2(bool *requires)
1122 {
1123   return io_ctx_impl->client->pool_requires_alignment2(get_id(), requires);
1124 }
1125
1126 uint64_t librados::IoCtx::pool_required_alignment()
1127 {
1128   return io_ctx_impl->client->pool_required_alignment(get_id());
1129 }
1130
1131 int librados::IoCtx::pool_required_alignment2(uint64_t *alignment)
1132 {
1133   return io_ctx_impl->client->pool_required_alignment2(get_id(), alignment);
1134 }
1135
1136 std::string librados::IoCtx::get_pool_name()
1137 {
1138   std::string s;
1139   io_ctx_impl->client->pool_get_name(get_id(), &s);
1140   return s;
1141 }
1142
1143 std::string librados::IoCtx::get_pool_name() const
1144 {
1145   return io_ctx_impl->get_cached_pool_name();
1146 }
1147
1148 uint64_t librados::IoCtx::get_instance_id() const
1149 {
1150   return io_ctx_impl->client->get_instance_id();
1151 }
1152
1153 int librados::IoCtx::create(const std::string& oid, bool exclusive)
1154 {
1155   object_t obj(oid);
1156   return io_ctx_impl->create(obj, exclusive);
1157 }
1158
1159 int librados::IoCtx::create(const std::string& oid, bool exclusive,
1160                             const std::string& category) // unused
1161 {
1162   object_t obj(oid);
1163   return io_ctx_impl->create(obj, exclusive);
1164 }
1165
1166 int librados::IoCtx::write(const std::string& oid, bufferlist& bl, size_t len, uint64_t off)
1167 {
1168   object_t obj(oid);
1169   return io_ctx_impl->write(obj, bl, len, off);
1170 }
1171
1172 int librados::IoCtx::append(const std::string& oid, bufferlist& bl, size_t len)
1173 {
1174   object_t obj(oid);
1175   return io_ctx_impl->append(obj, bl, len);
1176 }
1177
1178 int librados::IoCtx::write_full(const std::string& oid, bufferlist& bl)
1179 {
1180   object_t obj(oid);
1181   return io_ctx_impl->write_full(obj, bl);
1182 }
1183
1184 int librados::IoCtx::writesame(const std::string& oid, bufferlist& bl,
1185                                size_t write_len, uint64_t off)
1186 {
1187   object_t obj(oid);
1188   return io_ctx_impl->writesame(obj, bl, write_len, off);
1189 }
1190
1191
1192 int librados::IoCtx::read(const std::string& oid, bufferlist& bl, size_t len, uint64_t off)
1193 {
1194   object_t obj(oid);
1195   return io_ctx_impl->read(obj, bl, len, off);
1196 }
1197
1198 int librados::IoCtx::checksum(const std::string& oid,
1199                               rados_checksum_type_t type,
1200                               const bufferlist &init_value_bl, size_t len,
1201                               uint64_t off, size_t chunk_size, bufferlist *pbl)
1202 {
1203   object_t obj(oid);
1204   return io_ctx_impl->checksum(obj, get_checksum_op_type(type), init_value_bl,
1205                                len, off, chunk_size, pbl);
1206 }
1207
1208 int librados::IoCtx::remove(const std::string& oid)
1209 {
1210   object_t obj(oid);
1211   return io_ctx_impl->remove(obj);
1212 }
1213
1214 int librados::IoCtx::remove(const std::string& oid, int flags)
1215 {
1216   object_t obj(oid);
1217   return io_ctx_impl->remove(obj, flags);
1218 }
1219
1220 int librados::IoCtx::trunc(const std::string& oid, uint64_t size)
1221 {
1222   object_t obj(oid);
1223   return io_ctx_impl->trunc(obj, size);
1224 }
1225
1226 int librados::IoCtx::mapext(const std::string& oid, uint64_t off, size_t len,
1227                             std::map<uint64_t,uint64_t>& m)
1228 {
1229   object_t obj(oid);
1230   return io_ctx_impl->mapext(obj, off, len, m);
1231 }
1232
1233 int librados::IoCtx::cmpext(const std::string& oid, uint64_t off, bufferlist& cmp_bl)
1234 {
1235   object_t obj(oid);
1236   return io_ctx_impl->cmpext(obj, off, cmp_bl);
1237 }
1238
1239 int librados::IoCtx::sparse_read(const std::string& oid, std::map<uint64_t,uint64_t>& m,
1240                                  bufferlist& bl, size_t len, uint64_t off)
1241 {
1242   object_t obj(oid);
1243   return io_ctx_impl->sparse_read(obj, m, bl, len, off);
1244 }
1245
1246 int librados::IoCtx::getxattr(const std::string& oid, const char *name, bufferlist& bl)
1247 {
1248   object_t obj(oid);
1249   return io_ctx_impl->getxattr(obj, name, bl);
1250 }
1251
1252 int librados::IoCtx::getxattrs(const std::string& oid, map<std::string, bufferlist>& attrset)
1253 {
1254   object_t obj(oid);
1255   return io_ctx_impl->getxattrs(obj, attrset);
1256 }
1257
1258 int librados::IoCtx::setxattr(const std::string& oid, const char *name, bufferlist& bl)
1259 {
1260   object_t obj(oid);
1261   return io_ctx_impl->setxattr(obj, name, bl);
1262 }
1263
1264 int librados::IoCtx::rmxattr(const std::string& oid, const char *name)
1265 {
1266   object_t obj(oid);
1267   return io_ctx_impl->rmxattr(obj, name);
1268 }
1269
1270 int librados::IoCtx::stat(const std::string& oid, uint64_t *psize, time_t *pmtime)
1271 {
1272   object_t obj(oid);
1273   return io_ctx_impl->stat(obj, psize, pmtime);
1274 }
1275
1276 int librados::IoCtx::stat2(const std::string& oid, uint64_t *psize, struct timespec *pts)
1277 {
1278   object_t obj(oid);
1279   return io_ctx_impl->stat2(obj, psize, pts);
1280 }
1281
1282 int librados::IoCtx::exec(const std::string& oid, const char *cls, const char *method,
1283                           bufferlist& inbl, bufferlist& outbl)
1284 {
1285   object_t obj(oid);
1286   return io_ctx_impl->exec(obj, cls, method, inbl, outbl);
1287 }
1288
1289 int librados::IoCtx::tmap_update(const std::string& oid, bufferlist& cmdbl)
1290 {
1291   object_t obj(oid);
1292   return io_ctx_impl->tmap_update(obj, cmdbl);
1293 }
1294
1295 int librados::IoCtx::tmap_put(const std::string& oid, bufferlist& bl)
1296 {
1297   object_t obj(oid);
1298   return io_ctx_impl->tmap_put(obj, bl);
1299 }
1300
1301 int librados::IoCtx::tmap_get(const std::string& oid, bufferlist& bl)
1302 {
1303   object_t obj(oid);
1304   return io_ctx_impl->tmap_get(obj, bl);
1305 }
1306
1307 int librados::IoCtx::tmap_to_omap(const std::string& oid, bool nullok)
1308 {
1309   object_t obj(oid);
1310   return io_ctx_impl->tmap_to_omap(obj, nullok);
1311 }
1312
1313 int librados::IoCtx::omap_get_vals(const std::string& oid,
1314                                    const std::string& start_after,
1315                                    uint64_t max_return,
1316                                    std::map<std::string, bufferlist> *out_vals)
1317 {
1318   return omap_get_vals(oid, start_after, string(), max_return, out_vals);
1319 }
1320
1321 int librados::IoCtx::omap_get_vals2(
1322   const std::string& oid,
1323   const std::string& start_after,
1324   uint64_t max_return,
1325   std::map<std::string, bufferlist> *out_vals,
1326   bool *pmore)
1327 {
1328   ObjectReadOperation op;
1329   int r;
1330   op.omap_get_vals2(start_after, max_return, out_vals, pmore, &r);
1331   bufferlist bl;
1332   int ret = operate(oid, &op, &bl);
1333   if (ret < 0)
1334     return ret;
1335   return r;
1336 }
1337
1338 int librados::IoCtx::omap_get_keys(const std::string& oid,
1339                                    const std::string& orig_start_after,
1340                                    uint64_t max_return,
1341                                    std::set<std::string> *out_keys)
1342 {
1343   bool first = true;
1344   string start_after = orig_start_after;
1345   bool more = true;
1346   while (max_return > 0 && more) {
1347     std::set<std::string> out;
1348     ObjectReadOperation op;
1349     op.omap_get_keys2(start_after, max_return, &out, &more, nullptr);
1350     bufferlist bl;
1351     int ret = operate(oid, &op, &bl);
1352     if (ret < 0) {
1353       return ret;
1354     }
1355     if (more) {
1356       if (out.empty()) {
1357         return -EINVAL;  // wth
1358       }
1359       start_after = *out.rbegin();
1360     }
1361     if (out.size() <= max_return) {
1362       max_return -= out.size();
1363     } else {
1364       max_return = 0;
1365     }
1366     if (first) {
1367       out_keys->swap(out);
1368       first = false;
1369     } else {
1370       out_keys->insert(out.begin(), out.end());
1371       out.clear();
1372     }
1373   }
1374   return 0;
1375 }
1376
1377 int librados::IoCtx::omap_get_keys2(
1378   const std::string& oid,
1379   const std::string& start_after,
1380   uint64_t max_return,
1381   std::set<std::string> *out_keys,
1382   bool *pmore)
1383 {
1384   ObjectReadOperation op;
1385   int r;
1386   op.omap_get_keys2(start_after, max_return, out_keys, pmore, &r);
1387   bufferlist bl;
1388   int ret = operate(oid, &op, &bl);
1389   if (ret < 0)
1390     return ret;
1391   return r;
1392 }
1393
1394 int librados::IoCtx::omap_get_header(const std::string& oid,
1395                                      bufferlist *bl)
1396 {
1397   ObjectReadOperation op;
1398   int r;
1399   op.omap_get_header(bl, &r);
1400   bufferlist b;
1401   int ret = operate(oid, &op, &b);
1402   if (ret < 0)
1403     return ret;
1404
1405   return r;
1406 }
1407
1408 int librados::IoCtx::omap_get_vals_by_keys(const std::string& oid,
1409                                            const std::set<std::string>& keys,
1410                                            std::map<std::string, bufferlist> *vals)
1411 {
1412   ObjectReadOperation op;
1413   int r;
1414   bufferlist bl;
1415   op.omap_get_vals_by_keys(keys, vals, &r);
1416   int ret = operate(oid, &op, &bl);
1417   if (ret < 0)
1418     return ret;
1419
1420   return r;
1421 }
1422
1423 int librados::IoCtx::omap_set(const std::string& oid,
1424                               const map<string, bufferlist>& m)
1425 {
1426   ObjectWriteOperation op;
1427   op.omap_set(m);
1428   return operate(oid, &op);
1429 }
1430
1431 int librados::IoCtx::omap_set_header(const std::string& oid,
1432                                      const bufferlist& bl)
1433 {
1434   ObjectWriteOperation op;
1435   op.omap_set_header(bl);
1436   return operate(oid, &op);
1437 }
1438
1439 int librados::IoCtx::omap_clear(const std::string& oid)
1440 {
1441   ObjectWriteOperation op;
1442   op.omap_clear();
1443   return operate(oid, &op);
1444 }
1445
1446 int librados::IoCtx::omap_rm_keys(const std::string& oid,
1447                                   const std::set<std::string>& keys)
1448 {
1449   ObjectWriteOperation op;
1450   op.omap_rm_keys(keys);
1451   return operate(oid, &op);
1452 }
1453
1454
1455
1456 static int translate_flags(int flags)
1457 {
1458   int op_flags = 0;
1459   if (flags & librados::OPERATION_BALANCE_READS)
1460     op_flags |= CEPH_OSD_FLAG_BALANCE_READS;
1461   if (flags & librados::OPERATION_LOCALIZE_READS)
1462     op_flags |= CEPH_OSD_FLAG_LOCALIZE_READS;
1463   if (flags & librados::OPERATION_ORDER_READS_WRITES)
1464     op_flags |= CEPH_OSD_FLAG_RWORDERED;
1465   if (flags & librados::OPERATION_IGNORE_CACHE)
1466     op_flags |= CEPH_OSD_FLAG_IGNORE_CACHE;
1467   if (flags & librados::OPERATION_SKIPRWLOCKS)
1468     op_flags |= CEPH_OSD_FLAG_SKIPRWLOCKS;
1469   if (flags & librados::OPERATION_IGNORE_OVERLAY)
1470     op_flags |= CEPH_OSD_FLAG_IGNORE_OVERLAY;
1471   if (flags & librados::OPERATION_FULL_TRY)
1472     op_flags |= CEPH_OSD_FLAG_FULL_TRY;
1473   if (flags & librados::OPERATION_FULL_FORCE)
1474     op_flags |= CEPH_OSD_FLAG_FULL_FORCE;
1475
1476   return op_flags;
1477 }
1478
1479 int librados::IoCtx::operate(const std::string& oid, librados::ObjectWriteOperation *o)
1480 {
1481   object_t obj(oid);
1482   return io_ctx_impl->operate(obj, &o->impl->o, (ceph::real_time *)o->impl->prt);
1483 }
1484
1485 int librados::IoCtx::operate(const std::string& oid, librados::ObjectReadOperation *o, bufferlist *pbl)
1486 {
1487   object_t obj(oid);
1488   return io_ctx_impl->operate_read(obj, &o->impl->o, pbl);
1489 }
1490
1491 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c,
1492                                  librados::ObjectWriteOperation *o)
1493 {
1494   object_t obj(oid);
1495   return io_ctx_impl->aio_operate(obj, &o->impl->o, c->pc,
1496                                   io_ctx_impl->snapc, 0);
1497 }
1498 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c,
1499                                  ObjectWriteOperation *o, int flags)
1500 {
1501   object_t obj(oid);
1502   return io_ctx_impl->aio_operate(obj, &o->impl->o, c->pc,
1503                                   io_ctx_impl->snapc,
1504                                   translate_flags(flags));
1505 }
1506
1507 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c,
1508                                  librados::ObjectWriteOperation *o,
1509                                  snap_t snap_seq, std::vector<snap_t>& snaps)
1510 {
1511   object_t obj(oid);
1512   vector<snapid_t> snv;
1513   snv.resize(snaps.size());
1514   for (size_t i = 0; i < snaps.size(); ++i)
1515     snv[i] = snaps[i];
1516   SnapContext snapc(snap_seq, snv);
1517   return io_ctx_impl->aio_operate(obj, &o->impl->o, c->pc,
1518                                   snapc, 0);
1519 }
1520
1521 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c,
1522          librados::ObjectWriteOperation *o,
1523          snap_t snap_seq, std::vector<snap_t>& snaps,
1524          const blkin_trace_info *trace_info)
1525 {
1526   object_t obj(oid);
1527   vector<snapid_t> snv;
1528   snv.resize(snaps.size());
1529   for (size_t i = 0; i < snaps.size(); ++i)
1530     snv[i] = snaps[i];
1531   SnapContext snapc(snap_seq, snv);
1532   return io_ctx_impl->aio_operate(obj, &o->impl->o, c->pc,
1533           snapc, 0, trace_info);
1534 }
1535
1536 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c,
1537                                  librados::ObjectReadOperation *o,
1538                                  bufferlist *pbl)
1539 {
1540   object_t obj(oid);
1541   return io_ctx_impl->aio_operate_read(obj, &o->impl->o, c->pc,
1542                                        0, pbl);
1543 }
1544
1545 // deprecated
1546 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c,
1547                                  librados::ObjectReadOperation *o,
1548                                  snap_t snapid_unused_deprecated,
1549                                  int flags, bufferlist *pbl)
1550 {
1551   object_t obj(oid);
1552   int op_flags = 0;
1553   if (flags & OPERATION_BALANCE_READS)
1554     op_flags |= CEPH_OSD_FLAG_BALANCE_READS;
1555   if (flags & OPERATION_LOCALIZE_READS)
1556     op_flags |= CEPH_OSD_FLAG_LOCALIZE_READS;
1557   if (flags & OPERATION_ORDER_READS_WRITES)
1558     op_flags |= CEPH_OSD_FLAG_RWORDERED;
1559
1560   return io_ctx_impl->aio_operate_read(obj, &o->impl->o, c->pc,
1561                                        op_flags, pbl);
1562 }
1563
1564 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c,
1565                                  librados::ObjectReadOperation *o,
1566                                  int flags, bufferlist *pbl)
1567 {
1568   object_t obj(oid);
1569   return io_ctx_impl->aio_operate_read(obj, &o->impl->o, c->pc,
1570                                        translate_flags(flags), pbl);
1571 }
1572
1573 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c,
1574          librados::ObjectReadOperation *o,
1575          int flags, bufferlist *pbl, const blkin_trace_info *trace_info)
1576 {
1577   object_t obj(oid);
1578   return io_ctx_impl->aio_operate_read(obj, &o->impl->o, c->pc,
1579                translate_flags(flags), pbl, trace_info);
1580 }
1581
1582 void librados::IoCtx::snap_set_read(snap_t seq)
1583 {
1584   io_ctx_impl->set_snap_read(seq);
1585 }
1586
1587 int librados::IoCtx::selfmanaged_snap_set_write_ctx(snap_t seq, vector<snap_t>& snaps)
1588 {
1589   vector<snapid_t> snv;
1590   snv.resize(snaps.size());
1591   for (unsigned i=0; i<snaps.size(); i++)
1592     snv[i] = snaps[i];
1593   return io_ctx_impl->set_snap_write_context(seq, snv);
1594 }
1595
1596 int librados::IoCtx::snap_create(const char *snapname)
1597 {
1598   return io_ctx_impl->snap_create(snapname);
1599 }
1600
1601 int librados::IoCtx::snap_lookup(const char *name, snap_t *snapid)
1602 {
1603   return io_ctx_impl->snap_lookup(name, snapid);
1604 }
1605
1606 int librados::IoCtx::snap_get_stamp(snap_t snapid, time_t *t)
1607 {
1608   return io_ctx_impl->snap_get_stamp(snapid, t);
1609 }
1610
1611 int librados::IoCtx::snap_get_name(snap_t snapid, std::string *s)
1612 {
1613   return io_ctx_impl->snap_get_name(snapid, s);
1614 }
1615
1616 int librados::IoCtx::snap_remove(const char *snapname)
1617 {
1618   return io_ctx_impl->snap_remove(snapname);
1619 }
1620
1621 int librados::IoCtx::snap_list(std::vector<snap_t> *snaps)
1622 {
1623   return io_ctx_impl->snap_list(snaps);
1624 }
1625
1626 int librados::IoCtx::snap_rollback(const std::string& oid, const char *snapname)
1627 {
1628   return io_ctx_impl->rollback(oid, snapname);
1629 }
1630
1631 // Deprecated name kept for backward compatibility
1632 int librados::IoCtx::rollback(const std::string& oid, const char *snapname)
1633 {
1634   return snap_rollback(oid, snapname);
1635 }
1636
1637 int librados::IoCtx::selfmanaged_snap_create(uint64_t *snapid)
1638 {
1639   return io_ctx_impl->selfmanaged_snap_create(snapid);
1640 }
1641
1642 void librados::IoCtx::aio_selfmanaged_snap_create(uint64_t *snapid,
1643                                                   AioCompletion *c)
1644 {
1645   io_ctx_impl->aio_selfmanaged_snap_create(snapid, c->pc);
1646 }
1647
1648 int librados::IoCtx::selfmanaged_snap_remove(uint64_t snapid)
1649 {
1650   return io_ctx_impl->selfmanaged_snap_remove(snapid);
1651 }
1652
1653 void librados::IoCtx::aio_selfmanaged_snap_remove(uint64_t snapid,
1654                                                   AioCompletion *c)
1655 {
1656   io_ctx_impl->aio_selfmanaged_snap_remove(snapid, c->pc);
1657 }
1658
1659 int librados::IoCtx::selfmanaged_snap_rollback(const std::string& oid, uint64_t snapid)
1660 {
1661   return io_ctx_impl->selfmanaged_snap_rollback_object(oid,
1662                                                        io_ctx_impl->snapc,
1663                                                        snapid);
1664 }
1665
1666 int librados::IoCtx::lock_exclusive(const std::string &oid, const std::string &name,
1667                                     const std::string &cookie,
1668                                     const std::string &description,
1669                                     struct timeval * duration, uint8_t flags)
1670 {
1671   utime_t dur = utime_t();
1672   if (duration)
1673     dur.set_from_timeval(duration);
1674
1675   return rados::cls::lock::lock(this, oid, name, LOCK_EXCLUSIVE, cookie, "",
1676                                 description, dur, flags);
1677 }
1678
1679 int librados::IoCtx::lock_shared(const std::string &oid, const std::string &name,
1680                                  const std::string &cookie, const std::string &tag,
1681                                  const std::string &description,
1682                                  struct timeval * duration, uint8_t flags)
1683 {
1684   utime_t dur = utime_t();
1685   if (duration)
1686     dur.set_from_timeval(duration);
1687
1688   return rados::cls::lock::lock(this, oid, name, LOCK_SHARED, cookie, tag,
1689                                 description, dur, flags);
1690 }
1691
1692 int librados::IoCtx::unlock(const std::string &oid, const std::string &name,
1693                             const std::string &cookie)
1694 {
1695   return rados::cls::lock::unlock(this, oid, name, cookie);
1696 }
1697
1698 struct AioUnlockCompletion : public librados::ObjectOperationCompletion {
1699   librados::AioCompletionImpl *completion;
1700   AioUnlockCompletion(librados::AioCompletion *c) : completion(c->pc) {
1701     completion->get();
1702   };
1703   void handle_completion(int r, bufferlist& outbl) override {
1704     rados_callback_t cb = completion->callback_complete;
1705     void *cb_arg = completion->callback_complete_arg;
1706     cb(completion, cb_arg);
1707     completion->lock.Lock();
1708     completion->callback_complete = NULL;
1709     completion->cond.Signal();
1710     completion->put_unlock();
1711   }
1712 };
1713
1714 int librados::IoCtx::aio_unlock(const std::string &oid, const std::string &name,
1715                                 const std::string &cookie, AioCompletion *c)
1716 {
1717   return rados::cls::lock::aio_unlock(this, oid, name, cookie, c);
1718 }
1719
1720 int librados::IoCtx::break_lock(const std::string &oid, const std::string &name,
1721                                 const std::string &client, const std::string &cookie)
1722 {
1723   entity_name_t locker;
1724   if (!locker.parse(client))
1725     return -EINVAL;
1726   return rados::cls::lock::break_lock(this, oid, name, cookie, locker);
1727 }
1728
1729 int librados::IoCtx::list_lockers(const std::string &oid, const std::string &name,
1730                                   int *exclusive,
1731                                   std::string *tag,
1732                                   std::list<librados::locker_t> *lockers)
1733 {
1734   std::list<librados::locker_t> tmp_lockers;
1735   map<rados::cls::lock::locker_id_t, rados::cls::lock::locker_info_t> rados_lockers;
1736   std::string tmp_tag;
1737   ClsLockType tmp_type;
1738   int r = rados::cls::lock::get_lock_info(this, oid, name, &rados_lockers, &tmp_type, &tmp_tag);
1739   if (r < 0)
1740           return r;
1741
1742   map<rados::cls::lock::locker_id_t, rados::cls::lock::locker_info_t>::iterator map_it;
1743   for (map_it = rados_lockers.begin(); map_it != rados_lockers.end(); ++map_it) {
1744     librados::locker_t locker;
1745     locker.client = stringify(map_it->first.locker);
1746     locker.cookie = map_it->first.cookie;
1747     locker.address = stringify(map_it->second.addr);
1748     tmp_lockers.push_back(locker);
1749   }
1750
1751   if (lockers)
1752     *lockers = tmp_lockers;
1753   if (tag)
1754     *tag = tmp_tag;
1755   if (exclusive) {
1756     if (tmp_type == LOCK_EXCLUSIVE)
1757       *exclusive = 1;
1758     else
1759       *exclusive = 0;
1760   }
1761
1762   return tmp_lockers.size();
1763 }
1764
1765 librados::NObjectIterator librados::IoCtx::nobjects_begin()
1766 {
1767   bufferlist bl;
1768   return nobjects_begin(bl);
1769 }
1770
1771 librados::NObjectIterator librados::IoCtx::nobjects_begin(
1772     const bufferlist &filter)
1773 {
1774   rados_list_ctx_t listh;
1775   rados_nobjects_list_open(io_ctx_impl, &listh);
1776   NObjectIterator iter((ObjListCtx*)listh);
1777   if (filter.length() > 0) {
1778     iter.set_filter(filter);
1779   }
1780   iter.get_next();
1781   return iter;
1782 }
1783
1784 librados::NObjectIterator librados::IoCtx::nobjects_begin(uint32_t pos)
1785 {
1786   bufferlist bl;
1787   return nobjects_begin(pos, bl);
1788 }
1789
1790 librados::NObjectIterator librados::IoCtx::nobjects_begin(
1791   uint32_t pos, const bufferlist &filter)
1792 {
1793   rados_list_ctx_t listh;
1794   rados_nobjects_list_open(io_ctx_impl, &listh);
1795   NObjectIterator iter((ObjListCtx*)listh);
1796   if (filter.length() > 0) {
1797     iter.set_filter(filter);
1798   }
1799   iter.seek(pos);
1800   return iter;
1801 }
1802
1803 librados::NObjectIterator librados::IoCtx::nobjects_begin(const ObjectCursor& cursor)
1804 {
1805   bufferlist bl;
1806   return nobjects_begin(cursor, bl);
1807 }
1808
1809 librados::NObjectIterator librados::IoCtx::nobjects_begin(
1810   const ObjectCursor& cursor, const bufferlist &filter)
1811 {
1812   rados_list_ctx_t listh;
1813   rados_nobjects_list_open(io_ctx_impl, &listh);
1814   NObjectIterator iter((ObjListCtx*)listh);
1815   if (filter.length() > 0) {
1816     iter.set_filter(filter);
1817   }
1818   iter.seek(cursor);
1819   return iter;
1820 }
1821
1822 const librados::NObjectIterator& librados::IoCtx::nobjects_end() const
1823 {
1824   return NObjectIterator::__EndObjectIterator;
1825 }
1826
1827 int librados::IoCtx::hit_set_list(uint32_t hash, AioCompletion *c,
1828                                   std::list< std::pair<time_t, time_t> > *pls)
1829 {
1830   return io_ctx_impl->hit_set_list(hash, c->pc, pls);
1831 }
1832
1833 int librados::IoCtx::hit_set_get(uint32_t hash,  AioCompletion *c, time_t stamp,
1834                                  bufferlist *pbl)
1835 {
1836   return io_ctx_impl->hit_set_get(hash, c->pc, stamp, pbl);
1837 }
1838
1839
1840
1841 uint64_t librados::IoCtx::get_last_version()
1842 {
1843   return io_ctx_impl->last_version();
1844 }
1845
1846 int librados::IoCtx::aio_read(const std::string& oid, librados::AioCompletion *c,
1847                               bufferlist *pbl, size_t len, uint64_t off)
1848 {
1849   return io_ctx_impl->aio_read(oid, c->pc, pbl, len, off,
1850                                io_ctx_impl->snap_seq);
1851 }
1852
1853 int librados::IoCtx::aio_read(const std::string& oid, librados::AioCompletion *c,
1854                               bufferlist *pbl, size_t len, uint64_t off,
1855                               uint64_t snapid)
1856 {
1857   return io_ctx_impl->aio_read(oid, c->pc, pbl, len, off, snapid);
1858 }
1859
1860 int librados::IoCtx::aio_exec(const std::string& oid,
1861                               librados::AioCompletion *c, const char *cls,
1862                               const char *method, bufferlist& inbl,
1863                               bufferlist *outbl)
1864 {
1865   object_t obj(oid);
1866   return io_ctx_impl->aio_exec(obj, c->pc, cls, method, inbl, outbl);
1867 }
1868
1869 int librados::IoCtx::aio_cmpext(const std::string& oid,
1870                                 librados::AioCompletion *c,
1871                                 uint64_t off,
1872                                 bufferlist& cmp_bl)
1873 {
1874   return io_ctx_impl->aio_cmpext(oid, c->pc, off, cmp_bl);
1875 }
1876
1877 int librados::IoCtx::aio_sparse_read(const std::string& oid, librados::AioCompletion *c,
1878                                      std::map<uint64_t,uint64_t> *m, bufferlist *data_bl,
1879                                      size_t len, uint64_t off)
1880 {
1881   return io_ctx_impl->aio_sparse_read(oid, c->pc,
1882                                       m, data_bl, len, off,
1883                                       io_ctx_impl->snap_seq);
1884 }
1885
1886 int librados::IoCtx::aio_sparse_read(const std::string& oid, librados::AioCompletion *c,
1887                                      std::map<uint64_t,uint64_t> *m, bufferlist *data_bl,
1888                                      size_t len, uint64_t off, uint64_t snapid)
1889 {
1890   return io_ctx_impl->aio_sparse_read(oid, c->pc,
1891                                       m, data_bl, len, off, snapid);
1892 }
1893
1894 int librados::IoCtx::aio_write(const std::string& oid, librados::AioCompletion *c,
1895                                const bufferlist& bl, size_t len, uint64_t off)
1896 {
1897   return io_ctx_impl->aio_write(oid, c->pc, bl, len, off);
1898 }
1899
1900 int librados::IoCtx::aio_append(const std::string& oid, librados::AioCompletion *c,
1901                                 const bufferlist& bl, size_t len)
1902 {
1903   return io_ctx_impl->aio_append(oid, c->pc, bl, len);
1904 }
1905
1906 int librados::IoCtx::aio_write_full(const std::string& oid, librados::AioCompletion *c,
1907                                     const bufferlist& bl)
1908 {
1909   object_t obj(oid);
1910   return io_ctx_impl->aio_write_full(obj, c->pc, bl);
1911 }
1912
1913 int librados::IoCtx::aio_writesame(const std::string& oid, librados::AioCompletion *c,
1914                                    const bufferlist& bl, size_t write_len,
1915                                    uint64_t off)
1916 {
1917   return io_ctx_impl->aio_writesame(oid, c->pc, bl, write_len, off);
1918 }
1919
1920
1921 int librados::IoCtx::aio_remove(const std::string& oid, librados::AioCompletion *c)
1922 {
1923   return io_ctx_impl->aio_remove(oid, c->pc);
1924 }
1925
1926 int librados::IoCtx::aio_remove(const std::string& oid, librados::AioCompletion *c, int flags)
1927 {
1928   return io_ctx_impl->aio_remove(oid, c->pc, flags);
1929 }
1930
1931 int librados::IoCtx::aio_flush_async(librados::AioCompletion *c)
1932 {
1933   io_ctx_impl->flush_aio_writes_async(c->pc);
1934   return 0;
1935 }
1936
1937 int librados::IoCtx::aio_flush()
1938 {
1939   io_ctx_impl->flush_aio_writes();
1940   return 0;
1941 }
1942
1943 struct AioGetxattrDataPP {
1944   AioGetxattrDataPP(librados::AioCompletionImpl *c, bufferlist *_bl) :
1945     bl(_bl), completion(c) {}
1946   bufferlist *bl;
1947   struct librados::C_AioCompleteAndSafe completion;
1948 };
1949
1950 static void rados_aio_getxattr_completepp(rados_completion_t c, void *arg) {
1951   AioGetxattrDataPP *cdata = reinterpret_cast<AioGetxattrDataPP*>(arg);
1952   int rc = rados_aio_get_return_value(c);
1953   if (rc >= 0) {
1954     rc = cdata->bl->length();
1955   }
1956   cdata->completion.finish(rc);
1957   delete cdata;
1958 }
1959
1960 int librados::IoCtx::aio_getxattr(const std::string& oid, librados::AioCompletion *c,
1961                                   const char *name, bufferlist& bl)
1962 {
1963   // create data object to be passed to async callback
1964   AioGetxattrDataPP *cdata = new AioGetxattrDataPP(c->pc, &bl);
1965   if (!cdata) {
1966     return -ENOMEM;
1967   }
1968   // create completion callback
1969   librados::AioCompletionImpl *comp = new librados::AioCompletionImpl;
1970   comp->set_complete_callback(cdata, rados_aio_getxattr_completepp);
1971   // call actual getxattr from IoCtxImpl
1972   object_t obj(oid);
1973   return io_ctx_impl->aio_getxattr(obj, comp, name, bl);
1974 }
1975
1976 int librados::IoCtx::aio_getxattrs(const std::string& oid, AioCompletion *c,
1977                                    map<std::string, bufferlist>& attrset)
1978 {
1979   object_t obj(oid);
1980   return io_ctx_impl->aio_getxattrs(obj, c->pc, attrset);
1981 }
1982
1983 int librados::IoCtx::aio_setxattr(const std::string& oid, AioCompletion *c,
1984                                   const char *name, bufferlist& bl)
1985 {
1986   object_t obj(oid);
1987   return io_ctx_impl->aio_setxattr(obj, c->pc, name, bl);
1988 }
1989
1990 int librados::IoCtx::aio_rmxattr(const std::string& oid, AioCompletion *c,
1991                                  const char *name)
1992 {
1993   object_t obj(oid);
1994   return io_ctx_impl->aio_rmxattr(obj, c->pc, name);
1995 }
1996
1997 int librados::IoCtx::aio_stat(const std::string& oid, librados::AioCompletion *c,
1998                               uint64_t *psize, time_t *pmtime)
1999 {
2000   object_t obj(oid);
2001   return io_ctx_impl->aio_stat(obj, c->pc, psize, pmtime);
2002 }
2003
2004 int librados::IoCtx::aio_cancel(librados::AioCompletion *c)
2005 {
2006   return io_ctx_impl->aio_cancel(c->pc);
2007 }
2008
2009 int librados::IoCtx::watch(const string& oid, uint64_t ver, uint64_t *cookie,
2010                            librados::WatchCtx *ctx)
2011 {
2012   object_t obj(oid);
2013   return io_ctx_impl->watch(obj, cookie, ctx, NULL);
2014 }
2015
2016 int librados::IoCtx::watch2(const string& oid, uint64_t *cookie,
2017                             librados::WatchCtx2 *ctx2)
2018 {
2019   object_t obj(oid);
2020   return io_ctx_impl->watch(obj, cookie, NULL, ctx2);
2021 }
2022
2023 int librados::IoCtx::watch3(const string& oid, uint64_t *cookie,
2024           librados::WatchCtx2 *ctx2, uint32_t timeout)
2025 {
2026   object_t obj(oid);
2027   return io_ctx_impl->watch(obj, cookie, NULL, ctx2, timeout);
2028 }
2029
2030 int librados::IoCtx::aio_watch(const string& oid, AioCompletion *c,
2031                                uint64_t *cookie,
2032                                librados::WatchCtx2 *ctx2)
2033 {
2034   object_t obj(oid);
2035   return io_ctx_impl->aio_watch(obj, c->pc, cookie, NULL, ctx2);
2036 }
2037
2038 int librados::IoCtx::aio_watch2(const string& oid, AioCompletion *c,
2039                                 uint64_t *cookie,
2040                                 librados::WatchCtx2 *ctx2,
2041                                 uint32_t timeout)
2042 {
2043   object_t obj(oid);
2044   return io_ctx_impl->aio_watch(obj, c->pc, cookie, NULL, ctx2, timeout);
2045 }
2046
2047 int librados::IoCtx::unwatch(const string& oid, uint64_t handle)
2048 {
2049   return io_ctx_impl->unwatch(handle);
2050 }
2051
2052 int librados::IoCtx::unwatch2(uint64_t handle)
2053 {
2054   return io_ctx_impl->unwatch(handle);
2055 }
2056
2057 int librados::IoCtx::aio_unwatch(uint64_t handle, AioCompletion *c)
2058 {
2059   return io_ctx_impl->aio_unwatch(handle, c->pc);
2060 }
2061
2062 int librados::IoCtx::watch_check(uint64_t handle)
2063 {
2064   return io_ctx_impl->watch_check(handle);
2065 }
2066
2067 int librados::IoCtx::notify(const string& oid, uint64_t ver, bufferlist& bl)
2068 {
2069   object_t obj(oid);
2070   return io_ctx_impl->notify(obj, bl, 0, NULL, NULL, NULL);
2071 }
2072
2073 int librados::IoCtx::notify2(const string& oid, bufferlist& bl,
2074                              uint64_t timeout_ms, bufferlist *preplybl)
2075 {
2076   object_t obj(oid);
2077   return io_ctx_impl->notify(obj, bl, timeout_ms, preplybl, NULL, NULL);
2078 }
2079
2080 int librados::IoCtx::aio_notify(const string& oid, AioCompletion *c,
2081                                 bufferlist& bl, uint64_t timeout_ms,
2082                                 bufferlist *preplybl)
2083 {
2084   object_t obj(oid);
2085   return io_ctx_impl->aio_notify(obj, c->pc, bl, timeout_ms, preplybl, NULL,
2086                                  NULL);
2087 }
2088
2089 void librados::IoCtx::notify_ack(const std::string& o,
2090                                  uint64_t notify_id, uint64_t handle,
2091                                  bufferlist& bl)
2092 {
2093   io_ctx_impl->notify_ack(o, notify_id, handle, bl);
2094 }
2095
2096 int librados::IoCtx::list_watchers(const std::string& oid,
2097                                    std::list<obj_watch_t> *out_watchers)
2098 {
2099   ObjectReadOperation op;
2100   int r;
2101   op.list_watchers(out_watchers, &r);
2102   bufferlist bl;
2103   int ret = operate(oid, &op, &bl);
2104   if (ret < 0)
2105     return ret;
2106
2107   return r;
2108 }
2109
2110 int librados::IoCtx::list_snaps(const std::string& oid,
2111                                    snap_set_t *out_snaps)
2112 {
2113   ObjectReadOperation op;
2114   int r;
2115   if (io_ctx_impl->snap_seq != CEPH_SNAPDIR)
2116     return -EINVAL;
2117   op.list_snaps(out_snaps, &r);
2118   bufferlist bl;
2119   int ret = operate(oid, &op, &bl);
2120   if (ret < 0)
2121     return ret;
2122
2123   return r;
2124 }
2125
2126 void librados::IoCtx::set_notify_timeout(uint32_t timeout)
2127 {
2128   io_ctx_impl->set_notify_timeout(timeout);
2129 }
2130
2131 int librados::IoCtx::set_alloc_hint(const std::string& o,
2132                                     uint64_t expected_object_size,
2133                                     uint64_t expected_write_size)
2134 {
2135   object_t oid(o);
2136   return io_ctx_impl->set_alloc_hint(oid, expected_object_size,
2137                                      expected_write_size, 0);
2138 }
2139
2140 int librados::IoCtx::set_alloc_hint2(const std::string& o,
2141                                      uint64_t expected_object_size,
2142                                      uint64_t expected_write_size,
2143                                      uint32_t flags)
2144 {
2145   object_t oid(o);
2146   return io_ctx_impl->set_alloc_hint(oid, expected_object_size,
2147                                      expected_write_size, flags);
2148 }
2149
2150 void librados::IoCtx::set_assert_version(uint64_t ver)
2151 {
2152   io_ctx_impl->set_assert_version(ver);
2153 }
2154
2155 void librados::IoCtx::locator_set_key(const string& key)
2156 {
2157   io_ctx_impl->oloc.key = key;
2158 }
2159
2160 void librados::IoCtx::set_namespace(const string& nspace)
2161 {
2162   io_ctx_impl->oloc.nspace = nspace;
2163 }
2164
2165 int64_t librados::IoCtx::get_id()
2166 {
2167   return io_ctx_impl->get_id();
2168 }
2169
2170 uint32_t librados::IoCtx::get_object_hash_position(const std::string& oid)
2171 {
2172   uint32_t hash;
2173   int r = io_ctx_impl->get_object_hash_position(oid, &hash);
2174   if (r < 0)
2175     hash = 0;
2176   return hash;
2177 }
2178
2179 uint32_t librados::IoCtx::get_object_pg_hash_position(const std::string& oid)
2180 {
2181   uint32_t hash;
2182   int r = io_ctx_impl->get_object_pg_hash_position(oid, &hash);
2183   if (r < 0)
2184     hash = 0;
2185   return hash;
2186 }
2187
2188 int librados::IoCtx::get_object_hash_position2(
2189     const std::string& oid, uint32_t *hash_position)
2190 {
2191   return io_ctx_impl->get_object_hash_position(oid, hash_position);
2192 }
2193
2194 int librados::IoCtx::get_object_pg_hash_position2(
2195     const std::string& oid, uint32_t *pg_hash_position)
2196 {
2197   return io_ctx_impl->get_object_pg_hash_position(oid, pg_hash_position);
2198 }
2199
2200 librados::config_t librados::IoCtx::cct()
2201 {
2202   return (config_t)io_ctx_impl->client->cct;
2203 }
2204
2205 librados::IoCtx::IoCtx(IoCtxImpl *io_ctx_impl_)
2206   : io_ctx_impl(io_ctx_impl_)
2207 {
2208 }
2209
2210 void librados::IoCtx::set_osdmap_full_try()
2211 {
2212   io_ctx_impl->objecter->set_osdmap_full_try();
2213 }
2214
2215 void librados::IoCtx::unset_osdmap_full_try()
2216 {
2217   io_ctx_impl->objecter->unset_osdmap_full_try();
2218 }
2219
2220 ///////////////////////////// Rados //////////////////////////////
2221 void librados::Rados::version(int *major, int *minor, int *extra)
2222 {
2223   rados_version(major, minor, extra);
2224 }
2225
2226 librados::Rados::Rados() : client(NULL)
2227 {
2228 }
2229
2230 librados::Rados::Rados(IoCtx &ioctx)
2231 {
2232   client = ioctx.io_ctx_impl->client;
2233   assert(client != NULL);
2234   client->get();
2235 }
2236
2237 librados::Rados::~Rados()
2238 {
2239   shutdown();
2240 }
2241
2242 int librados::Rados::init(const char * const id)
2243 {
2244   return rados_create((rados_t *)&client, id);
2245 }
2246
2247 int librados::Rados::init2(const char * const name,
2248                            const char * const clustername, uint64_t flags)
2249 {
2250   return rados_create2((rados_t *)&client, clustername, name, flags);
2251 }
2252
2253 int librados::Rados::init_with_context(config_t cct_)
2254 {
2255   return rados_create_with_context((rados_t *)&client, (rados_config_t)cct_);
2256 }
2257
2258 int librados::Rados::connect()
2259 {
2260   return client->connect();
2261 }
2262
2263 librados::config_t librados::Rados::cct()
2264 {
2265   return (config_t)client->cct;
2266 }
2267
2268 int librados::Rados::watch_flush()
2269 {
2270   if (!client)
2271     return -EINVAL;
2272   return client->watch_flush();
2273 }
2274
2275 int librados::Rados::aio_watch_flush(AioCompletion *c)
2276 {
2277   if (!client)
2278     return -EINVAL;
2279   return client->async_watch_flush(c->pc);
2280 }
2281
2282 void librados::Rados::shutdown()
2283 {
2284   if (!client)
2285     return;
2286   if (client->put()) {
2287     client->shutdown();
2288     delete client;
2289     client = NULL;
2290   }
2291 }
2292
2293 uint64_t librados::Rados::get_instance_id()
2294 {
2295   return client->get_instance_id();
2296 }
2297
2298 int librados::Rados::conf_read_file(const char * const path) const
2299 {
2300   return rados_conf_read_file((rados_t)client, path);
2301 }
2302
2303 int librados::Rados::conf_parse_argv(int argc, const char ** argv) const
2304 {
2305   return rados_conf_parse_argv((rados_t)client, argc, argv);
2306 }
2307
2308 int librados::Rados::conf_parse_argv_remainder(int argc, const char ** argv,
2309                                                const char ** remargv) const
2310 {
2311   return rados_conf_parse_argv_remainder((rados_t)client, argc, argv, remargv);
2312 }
2313
2314 int librados::Rados::conf_parse_env(const char *name) const
2315 {
2316   return rados_conf_parse_env((rados_t)client, name);
2317 }
2318
2319 int librados::Rados::conf_set(const char *option, const char *value)
2320 {
2321   return rados_conf_set((rados_t)client, option, value);
2322 }
2323
2324 int librados::Rados::conf_get(const char *option, std::string &val)
2325 {
2326   char *str = NULL;
2327   md_config_t *conf = client->cct->_conf;
2328   int ret = conf->get_val(option, &str, -1);
2329   if (ret) {
2330     free(str);
2331     return ret;
2332   }
2333   val = str;
2334   free(str);
2335   return 0;
2336 }
2337
2338 int librados::Rados::pool_create(const char *name)
2339 {
2340   string str(name);
2341   return client->pool_create(str);
2342 }
2343
2344 int librados::Rados::pool_create(const char *name, uint64_t auid)
2345 {
2346   string str(name);
2347   return client->pool_create(str, auid);
2348 }
2349
2350 int librados::Rados::pool_create(const char *name, uint64_t auid, __u8 crush_rule)
2351 {
2352   string str(name);
2353   return client->pool_create(str, auid, crush_rule);
2354 }
2355
2356 int librados::Rados::pool_create_async(const char *name, PoolAsyncCompletion *c)
2357 {
2358   string str(name);
2359   return client->pool_create_async(str, c->pc);
2360 }
2361
2362 int librados::Rados::pool_create_async(const char *name, uint64_t auid, PoolAsyncCompletion *c)
2363 {
2364   string str(name);
2365   return client->pool_create_async(str, c->pc, auid);
2366 }
2367
2368 int librados::Rados::pool_create_async(const char *name, uint64_t auid, __u8 crush_rule,
2369                                        PoolAsyncCompletion *c)
2370 {
2371   string str(name);
2372   return client->pool_create_async(str, c->pc, auid, crush_rule);
2373 }
2374
2375 int librados::Rados::pool_get_base_tier(int64_t pool_id, int64_t* base_tier)
2376 {
2377   tracepoint(librados, rados_pool_get_base_tier_enter, (rados_t)client, pool_id);
2378   int retval = client->pool_get_base_tier(pool_id, base_tier);
2379   tracepoint(librados, rados_pool_get_base_tier_exit, retval, *base_tier);
2380   return retval;
2381 }
2382
2383 int librados::Rados::pool_delete(const char *name)
2384 {
2385   return client->pool_delete(name);
2386 }
2387
2388 int librados::Rados::pool_delete_async(const char *name, PoolAsyncCompletion *c)
2389 {
2390   return client->pool_delete_async(name, c->pc);
2391 }
2392
2393 int librados::Rados::pool_list(std::list<std::string>& v)
2394 {
2395   std::list<std::pair<int64_t, std::string> > pools;
2396   int r = client->pool_list(pools);
2397   if (r < 0) {
2398     return r;
2399   }
2400
2401   v.clear();
2402   for (std::list<std::pair<int64_t, std::string> >::iterator it = pools.begin();
2403        it != pools.end(); ++it) {
2404     v.push_back(it->second);
2405   }
2406   return 0;
2407 }
2408
2409 int librados::Rados::pool_list2(std::list<std::pair<int64_t, std::string> >& v)
2410 {
2411   return client->pool_list(v);
2412 }
2413
2414 int64_t librados::Rados::pool_lookup(const char *name)
2415 {
2416   return client->lookup_pool(name);
2417 }
2418
2419 int librados::Rados::pool_reverse_lookup(int64_t id, std::string *name)
2420 {
2421   return client->pool_get_name(id, name);
2422 }
2423
2424 int librados::Rados::mon_command(string cmd, const bufferlist& inbl,
2425                                  bufferlist *outbl, string *outs)
2426 {
2427   vector<string> cmdvec;
2428   cmdvec.push_back(cmd);
2429   return client->mon_command(cmdvec, inbl, outbl, outs);
2430 }
2431
2432 int librados::Rados::osd_command(int osdid, std::string cmd, const bufferlist& inbl,
2433                                  bufferlist *outbl, std::string *outs)
2434 {
2435   vector<string> cmdvec;
2436   cmdvec.push_back(cmd);
2437   return client->osd_command(osdid, cmdvec, inbl, outbl, outs);
2438 }
2439
2440 int librados::Rados::mgr_command(std::string cmd, const bufferlist& inbl,
2441                                  bufferlist *outbl, std::string *outs)
2442 {
2443   vector<string> cmdvec;
2444   cmdvec.push_back(cmd);
2445   return client->mgr_command(cmdvec, inbl, outbl, outs);
2446 }
2447
2448
2449
2450 int librados::Rados::pg_command(const char *pgstr, std::string cmd, const bufferlist& inbl,
2451                                 bufferlist *outbl, std::string *outs)
2452 {
2453   vector<string> cmdvec;
2454   cmdvec.push_back(cmd);
2455
2456   pg_t pgid;
2457   if (!pgid.parse(pgstr))
2458     return -EINVAL;
2459
2460   return client->pg_command(pgid, cmdvec, inbl, outbl, outs);
2461 }
2462
2463 int librados::Rados::ioctx_create(const char *name, IoCtx &io)
2464 {
2465   rados_ioctx_t p;
2466   int ret = rados_ioctx_create((rados_t)client, name, &p);
2467   if (ret)
2468     return ret;
2469   io.close();
2470   io.io_ctx_impl = (IoCtxImpl*)p;
2471   return 0;
2472 }
2473
2474 int librados::Rados::ioctx_create2(int64_t pool_id, IoCtx &io)
2475 {
2476   rados_ioctx_t p;
2477   int ret = rados_ioctx_create2((rados_t)client, pool_id, &p);
2478   if (ret)
2479     return ret;
2480   io.close();
2481   io.io_ctx_impl = (IoCtxImpl*)p;
2482   return 0;
2483 }
2484
2485 void librados::Rados::test_blacklist_self(bool set)
2486 {
2487   client->blacklist_self(set);
2488 }
2489
2490 int librados::Rados::get_pool_stats(std::list<string>& v,
2491                                     stats_map& result)
2492 {
2493   map<string,::pool_stat_t> rawresult;
2494   int r = client->get_pool_stats(v, rawresult);
2495   for (map<string,::pool_stat_t>::iterator p = rawresult.begin();
2496        p != rawresult.end();
2497        ++p) {
2498     pool_stat_t& pv = result[p->first];
2499     object_stat_sum_t *sum = &p->second.stats.sum;
2500     pv.num_kb = SHIFT_ROUND_UP(sum->num_bytes, 10);
2501     pv.num_bytes = sum->num_bytes;
2502     pv.num_objects = sum->num_objects;
2503     pv.num_object_clones = sum->num_object_clones;
2504     pv.num_object_copies = sum->num_object_copies;
2505     pv.num_objects_missing_on_primary = sum->num_objects_missing_on_primary;
2506     pv.num_objects_unfound = sum->num_objects_unfound;
2507     pv.num_objects_degraded = sum->num_objects_degraded;
2508     pv.num_rd = sum->num_rd;
2509     pv.num_rd_kb = sum->num_rd_kb;
2510     pv.num_wr = sum->num_wr;
2511     pv.num_wr_kb = sum->num_wr_kb;
2512   }
2513   return r;
2514 }
2515
2516 int librados::Rados::get_pool_stats(std::list<string>& v,
2517                                     std::map<string, stats_map>& result)
2518 {
2519   stats_map m;
2520   int r = get_pool_stats(v, m);
2521   if (r < 0)
2522     return r;
2523   for (map<string,pool_stat_t>::iterator p = m.begin();
2524        p != m.end();
2525        ++p) {
2526     result[p->first][string()] = p->second;
2527   }
2528   return r;
2529 }
2530
2531 int librados::Rados::get_pool_stats(std::list<string>& v,
2532                                     string& category, // unused
2533                                     std::map<string, stats_map>& result)
2534 {
2535   return -EOPNOTSUPP;
2536 }
2537
2538 bool librados::Rados::get_pool_is_selfmanaged_snaps_mode(const std::string& pool)
2539 {
2540   return client->get_pool_is_selfmanaged_snaps_mode(pool);
2541 }
2542
2543 int librados::Rados::cluster_stat(cluster_stat_t& result)
2544 {
2545   ceph_statfs stats;
2546   int r = client->get_fs_stats(stats);
2547   result.kb = stats.kb;
2548   result.kb_used = stats.kb_used;
2549   result.kb_avail = stats.kb_avail;
2550   result.num_objects = stats.num_objects;
2551   return r;
2552 }
2553
2554 int librados::Rados::cluster_fsid(string *fsid)
2555 {
2556   return client->get_fsid(fsid);
2557 }
2558
2559 namespace librados {
2560   struct PlacementGroupImpl {
2561     pg_t pgid;
2562   };
2563
2564   PlacementGroup::PlacementGroup()
2565     : impl{new PlacementGroupImpl}
2566   {}
2567
2568   PlacementGroup::PlacementGroup(const PlacementGroup& pg)
2569     : impl{new PlacementGroupImpl}
2570   {
2571     impl->pgid = pg.impl->pgid;
2572   }
2573
2574   PlacementGroup::~PlacementGroup()
2575   {}
2576
2577   bool PlacementGroup::parse(const char* s)
2578   {
2579     return impl->pgid.parse(s);
2580   }
2581 }
2582
2583 std::ostream& librados::operator<<(std::ostream& out,
2584                                    const librados::PlacementGroup& pg)
2585 {
2586   return out << pg.impl->pgid;
2587 }
2588
2589 namespace {
2590   int decode_json(JSONObj *obj, pg_t& pg)
2591   {
2592     string pg_str;
2593     JSONDecoder::decode_json("pgid", pg_str, obj);
2594     if (pg.parse(pg_str.c_str())) {
2595       return 0;
2596     } else {
2597       return -EINVAL;
2598     }
2599   }
2600
2601   int get_inconsistent_pgs(librados::RadosClient& client,
2602                            int64_t pool_id,
2603                            std::vector<librados::PlacementGroup>* pgs)
2604   {
2605     vector<string> cmd = {
2606       "{\"prefix\": \"pg ls\","
2607       "\"pool\": " + std::to_string(pool_id) + ","
2608       "\"states\": [\"inconsistent\"],"
2609       "\"format\": \"json\"}"
2610     };
2611     bufferlist inbl, outbl;
2612     string outstring;
2613     int ret = client.mon_command(cmd, inbl, &outbl, &outstring);
2614     if (ret) {
2615       return ret;
2616     }
2617     if (!outbl.length()) {
2618       // no pg returned
2619       return ret;
2620     }
2621     JSONParser parser;
2622     if (!parser.parse(outbl.c_str(), outbl.length())) {
2623       return -EINVAL;
2624     }
2625     if (!parser.is_array()) {
2626       return -EINVAL;
2627     }
2628     vector<string> v = parser.get_array_elements();
2629     for (auto i : v) {
2630       JSONParser pg_json;
2631       if (!pg_json.parse(i.c_str(), i.length())) {
2632         return -EINVAL;
2633       }
2634       librados::PlacementGroup pg;
2635       if (decode_json(&pg_json, pg.impl->pgid)) {
2636         return -EINVAL;
2637       }
2638       pgs->emplace_back(pg);
2639     }
2640     return 0;
2641   }
2642 }
2643
2644 int librados::Rados::get_inconsistent_pgs(int64_t pool_id,
2645                                           std::vector<PlacementGroup>* pgs)
2646 {
2647   return ::get_inconsistent_pgs(*client, pool_id, pgs);
2648 }
2649
2650 int librados::Rados::get_inconsistent_objects(const PlacementGroup& pg,
2651                                               const object_id_t &start_after,
2652                                               unsigned max_return,
2653                                               AioCompletion *c,
2654                                               std::vector<inconsistent_obj_t>* objects,
2655                                               uint32_t* interval)
2656 {
2657   IoCtx ioctx;
2658   const pg_t pgid = pg.impl->pgid;
2659   int r = ioctx_create2(pgid.pool(), ioctx);
2660   if (r < 0) {
2661     return r;
2662   }
2663
2664   return ioctx.io_ctx_impl->get_inconsistent_objects(pgid,
2665                                                      start_after,
2666                                                      max_return,
2667                                                      c->pc,
2668                                                      objects,
2669                                                      interval);
2670 }
2671
2672 int librados::Rados::get_inconsistent_snapsets(const PlacementGroup& pg,
2673                                                const object_id_t &start_after,
2674                                                unsigned max_return,
2675                                                AioCompletion *c,
2676                                                std::vector<inconsistent_snapset_t>* snapsets,
2677                                                uint32_t* interval)
2678 {
2679   IoCtx ioctx;
2680   const pg_t pgid = pg.impl->pgid;
2681   int r = ioctx_create2(pgid.pool(), ioctx);
2682   if (r < 0) {
2683     return r;
2684   }
2685
2686   return ioctx.io_ctx_impl->get_inconsistent_snapsets(pgid,
2687                                                       start_after,
2688                                                       max_return,
2689                                                       c->pc,
2690                                                       snapsets,
2691                                                       interval);
2692 }
2693
2694 int librados::Rados::wait_for_latest_osdmap()
2695 {
2696   return client->wait_for_latest_osdmap();
2697 }
2698
2699 int librados::Rados::blacklist_add(const std::string& client_address,
2700                                    uint32_t expire_seconds)
2701 {
2702   return client->blacklist_add(client_address, expire_seconds);
2703 }
2704
2705 librados::PoolAsyncCompletion *librados::Rados::pool_async_create_completion()
2706 {
2707   PoolAsyncCompletionImpl *c = new PoolAsyncCompletionImpl;
2708   return new PoolAsyncCompletion(c);
2709 }
2710
2711 librados::AioCompletion *librados::Rados::aio_create_completion()
2712 {
2713   AioCompletionImpl *c = new AioCompletionImpl;
2714   return new AioCompletion(c);
2715 }
2716
2717 librados::AioCompletion *librados::Rados::aio_create_completion(void *cb_arg,
2718                                                                 callback_t cb_complete,
2719                                                                 callback_t cb_safe)
2720 {
2721   AioCompletionImpl *c;
2722   int r = rados_aio_create_completion(cb_arg, cb_complete, cb_safe, (void**)&c);
2723   assert(r == 0);
2724   return new AioCompletion(c);
2725 }
2726
2727 librados::ObjectOperation::ObjectOperation()
2728 {
2729   impl = new ObjectOperationImpl;
2730 }
2731
2732 librados::ObjectOperation::~ObjectOperation()
2733 {
2734   delete impl;
2735 }
2736
2737 ///////////////////////////// C API //////////////////////////////
2738
2739 static CephContext *rados_create_cct(const char * const clustername,
2740                                      CephInitParameters *iparams)
2741 {
2742   // missing things compared to global_init:
2743   // g_ceph_context, g_conf, g_lockdep, signal handlers
2744   CephContext *cct = common_preinit(*iparams, CODE_ENVIRONMENT_LIBRARY, 0);
2745   if (clustername)
2746     cct->_conf->cluster = clustername;
2747   cct->_conf->parse_env(); // environment variables override
2748   cct->_conf->apply_changes(NULL);
2749
2750   TracepointProvider::initialize<tracepoint_traits>(cct);
2751   return cct;
2752 }
2753
2754 extern "C" int rados_create(rados_t *pcluster, const char * const id)
2755 {
2756   CephInitParameters iparams(CEPH_ENTITY_TYPE_CLIENT);
2757   if (id) {
2758     iparams.name.set(CEPH_ENTITY_TYPE_CLIENT, id);
2759   }
2760   CephContext *cct = rados_create_cct("", &iparams);
2761
2762   tracepoint(librados, rados_create_enter, id);
2763   *pcluster = reinterpret_cast<rados_t>(new librados::RadosClient(cct));
2764   tracepoint(librados, rados_create_exit, 0, *pcluster);
2765
2766   cct->put();
2767   return 0;
2768 }
2769
2770 // as above, but
2771 // 1) don't assume 'client.'; name is a full type.id namestr
2772 // 2) allow setting clustername
2773 // 3) flags is for future expansion (maybe some of the global_init()
2774 //    behavior is appropriate for some consumers of librados, for instance)
2775
2776 extern "C" int rados_create2(rados_t *pcluster, const char *const clustername,
2777                              const char * const name, uint64_t flags)
2778 {
2779   // client is assumed, but from_str will override
2780   int retval = 0;
2781   CephInitParameters iparams(CEPH_ENTITY_TYPE_CLIENT);
2782   if (!name || !iparams.name.from_str(name)) {
2783     retval = -EINVAL;
2784   }
2785
2786   CephContext *cct = rados_create_cct(clustername, &iparams);
2787   tracepoint(librados, rados_create2_enter, clustername, name, flags);
2788   if (retval == 0) {
2789     *pcluster = reinterpret_cast<rados_t>(new librados::RadosClient(cct));
2790   }
2791   tracepoint(librados, rados_create2_exit, retval, *pcluster);
2792
2793   cct->put();
2794   return retval;
2795 }
2796
2797 /* This function is intended for use by Ceph daemons. These daemons have
2798  * already called global_init and want to use that particular configuration for
2799  * their cluster.
2800  */
2801 extern "C" int rados_create_with_context(rados_t *pcluster, rados_config_t cct_)
2802 {
2803   CephContext *cct = (CephContext *)cct_;
2804   TracepointProvider::initialize<tracepoint_traits>(cct);
2805
2806   tracepoint(librados, rados_create_with_context_enter, cct_);
2807   librados::RadosClient *radosp = new librados::RadosClient(cct);
2808   *pcluster = (void *)radosp;
2809   tracepoint(librados, rados_create_with_context_exit, 0, *pcluster);
2810   return 0;
2811 }
2812
2813 extern "C" rados_config_t rados_cct(rados_t cluster)
2814 {
2815   tracepoint(librados, rados_cct_enter, cluster);
2816   librados::RadosClient *client = (librados::RadosClient *)cluster;
2817   rados_config_t retval = (rados_config_t)client->cct;
2818   tracepoint(librados, rados_cct_exit, retval);
2819   return retval;
2820 }
2821
2822 extern "C" int rados_connect(rados_t cluster)
2823 {
2824   tracepoint(librados, rados_connect_enter, cluster);
2825   librados::RadosClient *client = (librados::RadosClient *)cluster;
2826   int retval = client->connect();
2827   tracepoint(librados, rados_connect_exit, retval);
2828   return retval;
2829 }
2830
2831 extern "C" void rados_shutdown(rados_t cluster)
2832 {
2833   tracepoint(librados, rados_shutdown_enter, cluster);
2834   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
2835   radosp->shutdown();
2836   delete radosp;
2837   tracepoint(librados, rados_shutdown_exit);
2838 }
2839
2840 extern "C" uint64_t rados_get_instance_id(rados_t cluster)
2841 {
2842   tracepoint(librados, rados_get_instance_id_enter, cluster);
2843   librados::RadosClient *client = (librados::RadosClient *)cluster;
2844   uint64_t retval = client->get_instance_id();
2845   tracepoint(librados, rados_get_instance_id_exit, retval);
2846   return retval;
2847 }
2848
2849 extern "C" void rados_version(int *major, int *minor, int *extra)
2850 {
2851   tracepoint(librados, rados_version_enter, major, minor, extra);
2852   if (major)
2853     *major = LIBRADOS_VER_MAJOR;
2854   if (minor)
2855     *minor = LIBRADOS_VER_MINOR;
2856   if (extra)
2857     *extra = LIBRADOS_VER_EXTRA;
2858   tracepoint(librados, rados_version_exit, LIBRADOS_VER_MAJOR, LIBRADOS_VER_MINOR, LIBRADOS_VER_EXTRA);
2859 }
2860
2861
2862 // -- config --
2863 extern "C" int rados_conf_read_file(rados_t cluster, const char *path_list)
2864 {
2865   tracepoint(librados, rados_conf_read_file_enter, cluster, path_list);
2866   librados::RadosClient *client = (librados::RadosClient *)cluster;
2867   md_config_t *conf = client->cct->_conf;
2868   ostringstream warnings;
2869   int ret = conf->parse_config_files(path_list, &warnings, 0);
2870   if (ret) {
2871     if (warnings.tellp() > 0)
2872       lderr(client->cct) << warnings.str() << dendl;
2873     client->cct->_conf->complain_about_parse_errors(client->cct);
2874     tracepoint(librados, rados_conf_read_file_exit, ret);
2875     return ret;
2876   }
2877   conf->parse_env(); // environment variables override
2878
2879   conf->apply_changes(NULL);
2880   client->cct->_conf->complain_about_parse_errors(client->cct);
2881   tracepoint(librados, rados_conf_read_file_exit, 0);
2882   return 0;
2883 }
2884
2885 extern "C" int rados_conf_parse_argv(rados_t cluster, int argc, const char **argv)
2886 {
2887   tracepoint(librados, rados_conf_parse_argv_enter, cluster, argc);
2888   int i;
2889   for(i = 0; i < argc; i++) {
2890     tracepoint(librados, rados_conf_parse_argv_arg, argv[i]);
2891   }
2892   librados::RadosClient *client = (librados::RadosClient *)cluster;
2893   md_config_t *conf = client->cct->_conf;
2894   vector<const char*> args;
2895   argv_to_vec(argc, argv, args);
2896   int ret = conf->parse_argv(args);
2897   if (ret) {
2898     tracepoint(librados, rados_conf_parse_argv_exit, ret);
2899     return ret;
2900   }
2901   conf->apply_changes(NULL);
2902   tracepoint(librados, rados_conf_parse_argv_exit, 0);
2903   return 0;
2904 }
2905
2906 // like above, but return the remainder of argv to contain remaining
2907 // unparsed args.  Must be allocated to at least argc by caller.
2908 // remargv will contain n <= argc pointers to original argv[], the end
2909 // of which may be NULL
2910
2911 extern "C" int rados_conf_parse_argv_remainder(rados_t cluster, int argc,
2912                                                const char **argv,
2913                                                const char **remargv)
2914 {
2915   tracepoint(librados, rados_conf_parse_argv_remainder_enter, cluster, argc);
2916   unsigned int i;
2917   for(i = 0; i < (unsigned int) argc; i++) {
2918     tracepoint(librados, rados_conf_parse_argv_remainder_arg, argv[i]);
2919   }
2920   librados::RadosClient *client = (librados::RadosClient *)cluster;
2921   md_config_t *conf = client->cct->_conf;
2922   vector<const char*> args;
2923   for (int i=0; i<argc; i++)
2924     args.push_back(argv[i]);
2925   int ret = conf->parse_argv(args);
2926   if (ret) {
2927     tracepoint(librados, rados_conf_parse_argv_remainder_exit, ret);
2928     return ret;
2929   }
2930   conf->apply_changes(NULL);
2931   assert(args.size() <= (unsigned int)argc);
2932   for (i = 0; i < (unsigned int)argc; ++i) {
2933     if (i < args.size())
2934       remargv[i] = args[i];
2935     else
2936       remargv[i] = (const char *)NULL;
2937     tracepoint(librados, rados_conf_parse_argv_remainder_remarg, remargv[i]);
2938   }
2939   tracepoint(librados, rados_conf_parse_argv_remainder_exit, 0);
2940   return 0;
2941 }
2942
2943 extern "C" int rados_conf_parse_env(rados_t cluster, const char *env)
2944 {
2945   tracepoint(librados, rados_conf_parse_env_enter, cluster, env);
2946   librados::RadosClient *client = (librados::RadosClient *)cluster;
2947   md_config_t *conf = client->cct->_conf;
2948   vector<const char*> args;
2949   env_to_vec(args, env);
2950   int ret = conf->parse_argv(args);
2951   if (ret) {
2952     tracepoint(librados, rados_conf_parse_env_exit, ret);
2953     return ret;
2954   }
2955   conf->apply_changes(NULL);
2956   tracepoint(librados, rados_conf_parse_env_exit, 0);
2957   return 0;
2958 }
2959
2960 extern "C" int rados_conf_set(rados_t cluster, const char *option, const char *value)
2961 {
2962   tracepoint(librados, rados_conf_set_enter, cluster, option, value);
2963   librados::RadosClient *client = (librados::RadosClient *)cluster;
2964   md_config_t *conf = client->cct->_conf;
2965   int ret = conf->set_val(option, value);
2966   if (ret) {
2967     tracepoint(librados, rados_conf_set_exit, ret);
2968     return ret;
2969   }
2970   conf->apply_changes(NULL);
2971   tracepoint(librados, rados_conf_set_exit, 0);
2972   return 0;
2973 }
2974
2975 /* cluster info */
2976 extern "C" int rados_cluster_stat(rados_t cluster, rados_cluster_stat_t *result)
2977 {
2978   tracepoint(librados, rados_cluster_stat_enter, cluster);
2979   librados::RadosClient *client = (librados::RadosClient *)cluster;
2980
2981   ceph_statfs stats;
2982   int r = client->get_fs_stats(stats);
2983   result->kb = stats.kb;
2984   result->kb_used = stats.kb_used;
2985   result->kb_avail = stats.kb_avail;
2986   result->num_objects = stats.num_objects;
2987   tracepoint(librados, rados_cluster_stat_exit, r, result->kb, result->kb_used, result->kb_avail, result->num_objects);
2988   return r;
2989 }
2990
2991 extern "C" int rados_conf_get(rados_t cluster, const char *option, char *buf, size_t len)
2992 {
2993   tracepoint(librados, rados_conf_get_enter, cluster, option, len);
2994   char *tmp = buf;
2995   librados::RadosClient *client = (librados::RadosClient *)cluster;
2996   md_config_t *conf = client->cct->_conf;
2997   int retval = conf->get_val(option, &tmp, len);
2998   tracepoint(librados, rados_conf_get_exit, retval, retval ? "" : option);
2999   return retval;
3000 }
3001
3002 extern "C" int64_t rados_pool_lookup(rados_t cluster, const char *name)
3003 {
3004   tracepoint(librados, rados_pool_lookup_enter, cluster, name);
3005   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3006   int64_t retval = radosp->lookup_pool(name);
3007   tracepoint(librados, rados_pool_lookup_exit, retval);
3008   return retval;
3009 }
3010
3011 extern "C" int rados_pool_reverse_lookup(rados_t cluster, int64_t id,
3012                                          char *buf, size_t maxlen)
3013 {
3014   tracepoint(librados, rados_pool_reverse_lookup_enter, cluster, id, maxlen);
3015   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3016   std::string name;
3017   int r = radosp->pool_get_name(id, &name);
3018   if (r < 0) {
3019     tracepoint(librados, rados_pool_reverse_lookup_exit, r, "");
3020     return r;
3021   }
3022   if (name.length() >= maxlen) {
3023     tracepoint(librados, rados_pool_reverse_lookup_exit, -ERANGE, "");
3024     return -ERANGE;
3025   }
3026   strcpy(buf, name.c_str());
3027   int retval = name.length();
3028   tracepoint(librados, rados_pool_reverse_lookup_exit, retval, buf);
3029   return retval;
3030 }
3031
3032 extern "C" int rados_cluster_fsid(rados_t cluster, char *buf,
3033                                   size_t maxlen)
3034 {
3035   tracepoint(librados, rados_cluster_fsid_enter, cluster, maxlen);
3036   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3037   std::string fsid;
3038   radosp->get_fsid(&fsid);
3039   if (fsid.length() >= maxlen) {
3040     tracepoint(librados, rados_cluster_fsid_exit, -ERANGE, "");
3041     return -ERANGE;
3042   }
3043   strcpy(buf, fsid.c_str());
3044   int retval = fsid.length();
3045   tracepoint(librados, rados_cluster_fsid_exit, retval, buf);
3046   return retval;
3047 }
3048
3049 extern "C" int rados_wait_for_latest_osdmap(rados_t cluster)
3050 {
3051   tracepoint(librados, rados_wait_for_latest_osdmap_enter, cluster);
3052   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3053   int retval = radosp->wait_for_latest_osdmap();
3054   tracepoint(librados, rados_wait_for_latest_osdmap_exit, retval);
3055   return retval;
3056 }
3057
3058 extern "C" int rados_blacklist_add(rados_t cluster, char *client_address,
3059                                    uint32_t expire_seconds)
3060 {
3061   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3062   return radosp->blacklist_add(client_address, expire_seconds);
3063 }
3064
3065 extern "C" int rados_pool_list(rados_t cluster, char *buf, size_t len)
3066 {
3067   tracepoint(librados, rados_pool_list_enter, cluster, len);
3068   librados::RadosClient *client = (librados::RadosClient *)cluster;
3069   std::list<std::pair<int64_t, std::string> > pools;
3070   int r = client->pool_list(pools);
3071   if (r < 0) {
3072     tracepoint(librados, rados_pool_list_exit, r);
3073     return r;
3074   }
3075
3076   if (len > 0 && !buf) {
3077     tracepoint(librados, rados_pool_list_exit, -EINVAL);
3078     return -EINVAL;
3079   }
3080
3081   char *b = buf;
3082   if (b)
3083     memset(b, 0, len);
3084   int needed = 0;
3085   std::list<std::pair<int64_t, std::string> >::const_iterator i = pools.begin();
3086   std::list<std::pair<int64_t, std::string> >::const_iterator p_end =
3087     pools.end();
3088   for (; i != p_end; ++i) {
3089     int rl = i->second.length() + 1;
3090     if (len < (unsigned)rl)
3091       break;
3092     const char* pool = i->second.c_str();
3093     tracepoint(librados, rados_pool_list_pool, pool);
3094     if (b) {
3095       strncat(b, pool, rl);
3096       b += rl;
3097     }
3098     needed += rl;
3099     len -= rl;
3100   }
3101   for (; i != p_end; ++i) {
3102     int rl = i->second.length() + 1;
3103     needed += rl;
3104   }
3105   int retval = needed + 1;
3106   tracepoint(librados, rados_pool_list_exit, retval);
3107   return retval;
3108 }
3109
3110 CEPH_RADOS_API int rados_inconsistent_pg_list(rados_t cluster, int64_t pool_id,
3111                                               char *buf, size_t len)
3112 {
3113   tracepoint(librados, rados_inconsistent_pg_list_enter, cluster, pool_id, len);
3114   librados::RadosClient *client = (librados::RadosClient *)cluster;
3115   std::vector<librados::PlacementGroup> pgs;
3116   int r = ::get_inconsistent_pgs(*client, pool_id, &pgs);
3117   if (r < 0) {
3118     tracepoint(librados, rados_inconsistent_pg_list_exit, r);
3119     return r;
3120   }
3121
3122   if (len > 0 && !buf) {
3123     tracepoint(librados, rados_inconsistent_pg_list_exit, -EINVAL);
3124     return -EINVAL;
3125   }
3126
3127   char *b = buf;
3128   if (b)
3129     memset(b, 0, len);
3130   int needed = 0;
3131   for (const auto pg : pgs) {
3132     std::ostringstream ss;
3133     ss << pg;
3134     auto s = ss.str();
3135     unsigned rl = s.length() + 1;
3136     if (b && len >= rl) {
3137       tracepoint(librados, rados_inconsistent_pg_list_pg, s.c_str());
3138       strncat(b, s.c_str(), rl);
3139       b += rl;
3140       len -= rl;
3141     }
3142     needed += rl;
3143   }
3144   int retval = needed + 1;
3145   tracepoint(librados, rados_inconsistent_pg_list_exit, retval);
3146   return retval;
3147 }
3148
3149 static void do_out_buffer(bufferlist& outbl, char **outbuf, size_t *outbuflen)
3150 {
3151   if (outbuf) {
3152     if (outbl.length() > 0) {
3153       *outbuf = (char *)malloc(outbl.length());
3154       memcpy(*outbuf, outbl.c_str(), outbl.length());
3155     } else {
3156       *outbuf = NULL;
3157     }
3158   }
3159   if (outbuflen)
3160     *outbuflen = outbl.length();
3161 }
3162
3163 static void do_out_buffer(string& outbl, char **outbuf, size_t *outbuflen)
3164 {
3165   if (outbuf) {
3166     if (outbl.length() > 0) {
3167       *outbuf = (char *)malloc(outbl.length());
3168       memcpy(*outbuf, outbl.c_str(), outbl.length());
3169     } else {
3170       *outbuf = NULL;
3171     }
3172   }
3173   if (outbuflen)
3174     *outbuflen = outbl.length();
3175 }
3176
3177 extern "C" int rados_ping_monitor(rados_t cluster, const char *mon_id,
3178                                   char **outstr, size_t *outstrlen)
3179 {
3180   tracepoint(librados, rados_ping_monitor_enter, cluster, mon_id);
3181   librados::RadosClient *client = (librados::RadosClient *)cluster;
3182   string str;
3183
3184   if (!mon_id) {
3185     tracepoint(librados, rados_ping_monitor_exit, -EINVAL, NULL, NULL);
3186     return -EINVAL;
3187   }
3188
3189   int ret = client->ping_monitor(mon_id, &str);
3190   if (ret == 0) {
3191     do_out_buffer(str, outstr, outstrlen);
3192   }
3193   tracepoint(librados, rados_ping_monitor_exit, ret, ret < 0 ? NULL : outstr, ret < 0 ? NULL : outstrlen);
3194   return ret;
3195 }
3196
3197 extern "C" int rados_mon_command(rados_t cluster, const char **cmd,
3198                                  size_t cmdlen,
3199                                  const char *inbuf, size_t inbuflen,
3200                                  char **outbuf, size_t *outbuflen,
3201                                  char **outs, size_t *outslen)
3202 {
3203   tracepoint(librados, rados_mon_command_enter, cluster, cmdlen, inbuf, inbuflen);
3204   librados::RadosClient *client = (librados::RadosClient *)cluster;
3205   bufferlist inbl;
3206   bufferlist outbl;
3207   string outstring;
3208   vector<string> cmdvec;
3209
3210   for (size_t i = 0; i < cmdlen; i++) {
3211     tracepoint(librados, rados_mon_command_cmd, cmd[i]);
3212     cmdvec.push_back(cmd[i]);
3213   }
3214
3215   inbl.append(inbuf, inbuflen);
3216   int ret = client->mon_command(cmdvec, inbl, &outbl, &outstring);
3217
3218   do_out_buffer(outbl, outbuf, outbuflen);
3219   do_out_buffer(outstring, outs, outslen);
3220   tracepoint(librados, rados_mon_command_exit, ret, outbuf, outbuflen, outs, outslen);
3221   return ret;
3222 }
3223
3224 extern "C" int rados_mon_command_target(rados_t cluster, const char *name,
3225                                         const char **cmd,
3226                                         size_t cmdlen,
3227                                         const char *inbuf, size_t inbuflen,
3228                                         char **outbuf, size_t *outbuflen,
3229                                         char **outs, size_t *outslen)
3230 {
3231   tracepoint(librados, rados_mon_command_target_enter, cluster, name, cmdlen, inbuf, inbuflen);
3232   librados::RadosClient *client = (librados::RadosClient *)cluster;
3233   bufferlist inbl;
3234   bufferlist outbl;
3235   string outstring;
3236   vector<string> cmdvec;
3237
3238   // is this a numeric id?
3239   char *endptr;
3240   errno = 0;
3241   long rank = strtol(name, &endptr, 10);
3242   if ((errno == ERANGE && (rank == LONG_MAX || rank == LONG_MIN)) ||
3243       (errno != 0 && rank == 0) ||
3244       endptr == name ||    // no digits
3245       *endptr != '\0') {   // extra characters
3246     rank = -1;
3247   }
3248
3249   for (size_t i = 0; i < cmdlen; i++) {
3250     tracepoint(librados, rados_mon_command_target_cmd, cmd[i]);
3251     cmdvec.push_back(cmd[i]);
3252   }
3253
3254   inbl.append(inbuf, inbuflen);
3255   int ret;
3256   if (rank >= 0)
3257     ret = client->mon_command(rank, cmdvec, inbl, &outbl, &outstring);
3258   else
3259     ret = client->mon_command(name, cmdvec, inbl, &outbl, &outstring);
3260
3261   do_out_buffer(outbl, outbuf, outbuflen);
3262   do_out_buffer(outstring, outs, outslen);
3263   tracepoint(librados, rados_mon_command_target_exit, ret, outbuf, outbuflen, outs, outslen);
3264   return ret;
3265 }
3266
3267 extern "C" int rados_osd_command(rados_t cluster, int osdid, const char **cmd,
3268                                  size_t cmdlen,
3269                                  const char *inbuf, size_t inbuflen,
3270                                  char **outbuf, size_t *outbuflen,
3271                                  char **outs, size_t *outslen)
3272 {
3273   tracepoint(librados, rados_osd_command_enter, cluster, osdid, cmdlen, inbuf, inbuflen);
3274   librados::RadosClient *client = (librados::RadosClient *)cluster;
3275   bufferlist inbl;
3276   bufferlist outbl;
3277   string outstring;
3278   vector<string> cmdvec;
3279
3280   for (size_t i = 0; i < cmdlen; i++) {
3281     tracepoint(librados, rados_osd_command_cmd, cmd[i]);
3282     cmdvec.push_back(cmd[i]);
3283   }
3284
3285   inbl.append(inbuf, inbuflen);
3286   int ret = client->osd_command(osdid, cmdvec, inbl, &outbl, &outstring);
3287
3288   do_out_buffer(outbl, outbuf, outbuflen);
3289   do_out_buffer(outstring, outs, outslen);
3290   tracepoint(librados, rados_osd_command_exit, ret, outbuf, outbuflen, outs, outslen);
3291   return ret;
3292 }
3293
3294 extern "C" int rados_mgr_command(rados_t cluster, const char **cmd,
3295                                  size_t cmdlen,
3296                                  const char *inbuf, size_t inbuflen,
3297                                  char **outbuf, size_t *outbuflen,
3298                                  char **outs, size_t *outslen)
3299 {
3300   tracepoint(librados, rados_mgr_command_enter, cluster, cmdlen, inbuf,
3301       inbuflen);
3302
3303   librados::RadosClient *client = (librados::RadosClient *)cluster;
3304   bufferlist inbl;
3305   bufferlist outbl;
3306   string outstring;
3307   vector<string> cmdvec;
3308
3309   for (size_t i = 0; i < cmdlen; i++) {
3310     tracepoint(librados, rados_mgr_command_cmd, cmd[i]);
3311     cmdvec.push_back(cmd[i]);
3312   }
3313
3314   inbl.append(inbuf, inbuflen);
3315   int ret = client->mgr_command(cmdvec, inbl, &outbl, &outstring);
3316
3317   do_out_buffer(outbl, outbuf, outbuflen);
3318   do_out_buffer(outstring, outs, outslen);
3319   tracepoint(librados, rados_mgr_command_exit, ret, outbuf, outbuflen, outs,
3320       outslen);
3321   return ret;
3322 }
3323
3324 extern "C" int rados_pg_command(rados_t cluster, const char *pgstr,
3325                                 const char **cmd, size_t cmdlen,
3326                                 const char *inbuf, size_t inbuflen,
3327                                 char **outbuf, size_t *outbuflen,
3328                                 char **outs, size_t *outslen)
3329 {
3330   tracepoint(librados, rados_pg_command_enter, cluster, pgstr, cmdlen, inbuf, inbuflen);
3331   librados::RadosClient *client = (librados::RadosClient *)cluster;
3332   bufferlist inbl;
3333   bufferlist outbl;
3334   string outstring;
3335   pg_t pgid;
3336   vector<string> cmdvec;
3337
3338   for (size_t i = 0; i < cmdlen; i++) {
3339     tracepoint(librados, rados_pg_command_cmd, cmd[i]);
3340     cmdvec.push_back(cmd[i]);
3341   }
3342
3343   inbl.append(inbuf, inbuflen);
3344   if (!pgid.parse(pgstr))
3345     return -EINVAL;
3346
3347   int ret = client->pg_command(pgid, cmdvec, inbl, &outbl, &outstring);
3348
3349   do_out_buffer(outbl, outbuf, outbuflen);
3350   do_out_buffer(outstring, outs, outslen);
3351   tracepoint(librados, rados_pg_command_exit, ret, outbuf, outbuflen, outs, outslen);
3352   return ret;
3353 }
3354
3355 extern "C" void rados_buffer_free(char *buf)
3356 {
3357   tracepoint(librados, rados_buffer_free_enter, buf);
3358   if (buf)
3359     free(buf);
3360   tracepoint(librados, rados_buffer_free_exit);
3361 }
3362
3363 extern "C" int rados_monitor_log(rados_t cluster, const char *level, rados_log_callback_t cb, void *arg)
3364 {
3365   tracepoint(librados, rados_monitor_log_enter, cluster, level, cb, arg);
3366   librados::RadosClient *client = (librados::RadosClient *)cluster;
3367   int retval = client->monitor_log(level, cb, arg);
3368   tracepoint(librados, rados_monitor_log_exit, retval);
3369   return retval;
3370 }
3371
3372 extern "C" int rados_ioctx_create(rados_t cluster, const char *name, rados_ioctx_t *io)
3373 {
3374   tracepoint(librados, rados_ioctx_create_enter, cluster, name);
3375   librados::RadosClient *client = (librados::RadosClient *)cluster;
3376   librados::IoCtxImpl *ctx;
3377
3378   int r = client->create_ioctx(name, &ctx);
3379   if (r < 0) {
3380     tracepoint(librados, rados_ioctx_create_exit, r, NULL);
3381     return r;
3382   }
3383
3384   *io = ctx;
3385   ctx->get();
3386   tracepoint(librados, rados_ioctx_create_exit, 0, ctx);
3387   return 0;
3388 }
3389
3390 extern "C" int rados_ioctx_create2(rados_t cluster, int64_t pool_id,
3391                                    rados_ioctx_t *io)
3392 {
3393   tracepoint(librados, rados_ioctx_create2_enter, cluster, pool_id);
3394   librados::RadosClient *client = (librados::RadosClient *)cluster;
3395   librados::IoCtxImpl *ctx;
3396
3397   int r = client->create_ioctx(pool_id, &ctx);
3398   if (r < 0) {
3399     tracepoint(librados, rados_ioctx_create2_exit, r, NULL);
3400     return r;
3401   }
3402
3403   *io = ctx;
3404   ctx->get();
3405   tracepoint(librados, rados_ioctx_create2_exit, 0, ctx);
3406   return 0;
3407 }
3408
3409 extern "C" void rados_ioctx_destroy(rados_ioctx_t io)
3410 {
3411   tracepoint(librados, rados_ioctx_destroy_enter, io);
3412   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3413   ctx->put();
3414   tracepoint(librados, rados_ioctx_destroy_exit);
3415 }
3416
3417 extern "C" int rados_ioctx_pool_stat(rados_ioctx_t io, struct rados_pool_stat_t *stats)
3418 {
3419   tracepoint(librados, rados_ioctx_pool_stat_enter, io);
3420   librados::IoCtxImpl *io_ctx_impl = (librados::IoCtxImpl *)io;
3421   list<string> ls;
3422   std::string pool_name;
3423
3424   int err = io_ctx_impl->client->pool_get_name(io_ctx_impl->get_id(), &pool_name);
3425   if (err) {
3426     tracepoint(librados, rados_ioctx_pool_stat_exit, err, stats);
3427     return err;
3428   }
3429   ls.push_back(pool_name);
3430
3431   map<string, ::pool_stat_t> rawresult;
3432   err = io_ctx_impl->client->get_pool_stats(ls, rawresult);
3433   if (err) {
3434     tracepoint(librados, rados_ioctx_pool_stat_exit, err, stats);
3435     return err;
3436   }
3437
3438   ::pool_stat_t& r = rawresult[pool_name];
3439   stats->num_kb = SHIFT_ROUND_UP(r.stats.sum.num_bytes, 10);
3440   stats->num_bytes = r.stats.sum.num_bytes;
3441   stats->num_objects = r.stats.sum.num_objects;
3442   stats->num_object_clones = r.stats.sum.num_object_clones;
3443   stats->num_object_copies = r.stats.sum.num_object_copies;
3444   stats->num_objects_missing_on_primary = r.stats.sum.num_objects_missing_on_primary;
3445   stats->num_objects_unfound = r.stats.sum.num_objects_unfound;
3446   stats->num_objects_degraded =
3447     r.stats.sum.num_objects_degraded +
3448     r.stats.sum.num_objects_misplaced; // FIXME: this is imprecise
3449   stats->num_rd = r.stats.sum.num_rd;
3450   stats->num_rd_kb = r.stats.sum.num_rd_kb;
3451   stats->num_wr = r.stats.sum.num_wr;
3452   stats->num_wr_kb = r.stats.sum.num_wr_kb;
3453   tracepoint(librados, rados_ioctx_pool_stat_exit, 0, stats);
3454   return 0;
3455 }
3456
3457 extern "C" rados_config_t rados_ioctx_cct(rados_ioctx_t io)
3458 {
3459   tracepoint(librados, rados_ioctx_cct_enter, io);
3460   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3461   rados_config_t retval = (rados_config_t)ctx->client->cct;
3462   tracepoint(librados, rados_ioctx_cct_exit, retval);
3463   return retval;
3464 }
3465
3466 extern "C" void rados_ioctx_snap_set_read(rados_ioctx_t io, rados_snap_t seq)
3467 {
3468   tracepoint(librados, rados_ioctx_snap_set_read_enter, io, seq);
3469   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3470   ctx->set_snap_read((snapid_t)seq);
3471   tracepoint(librados, rados_ioctx_snap_set_read_exit);
3472 }
3473
3474 extern "C" int rados_ioctx_selfmanaged_snap_set_write_ctx(rados_ioctx_t io,
3475             rados_snap_t seq, rados_snap_t *snaps, int num_snaps)
3476 {
3477   tracepoint(librados, rados_ioctx_selfmanaged_snap_set_write_ctx_enter, io, seq, snaps, num_snaps);
3478   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3479   vector<snapid_t> snv;
3480   snv.resize(num_snaps);
3481   for (int i=0; i<num_snaps; i++) {
3482     snv[i] = (snapid_t)snaps[i];
3483   }
3484   int retval = ctx->set_snap_write_context((snapid_t)seq, snv);
3485   tracepoint(librados, rados_ioctx_selfmanaged_snap_set_write_ctx_exit, retval);
3486   return retval;
3487 }
3488
3489 extern "C" int rados_write(rados_ioctx_t io, const char *o, const char *buf, size_t len, uint64_t off)
3490 {
3491   tracepoint(librados, rados_write_enter, io, o, buf, len, off);
3492   if (len > UINT_MAX/2)
3493     return -E2BIG;
3494   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3495   object_t oid(o);
3496   bufferlist bl;
3497   bl.append(buf, len);
3498   int retval = ctx->write(oid, bl, len, off);
3499   tracepoint(librados, rados_write_exit, retval);
3500   return retval;
3501 }
3502
3503 extern "C" int rados_append(rados_ioctx_t io, const char *o, const char *buf, size_t len)
3504 {
3505   tracepoint(librados, rados_append_enter, io, o, buf, len);
3506   if (len > UINT_MAX/2)
3507     return -E2BIG;
3508   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3509   object_t oid(o);
3510   bufferlist bl;
3511   bl.append(buf, len);
3512   int retval = ctx->append(oid, bl, len);
3513   tracepoint(librados, rados_append_exit, retval);
3514   return retval;
3515 }
3516
3517 extern "C" int rados_write_full(rados_ioctx_t io, const char *o, const char *buf, size_t len)
3518 {
3519   tracepoint(librados, rados_write_full_enter, io, o, buf, len);
3520   if (len > UINT_MAX/2)
3521     return -E2BIG;
3522   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3523   object_t oid(o);
3524   bufferlist bl;
3525   bl.append(buf, len);
3526   int retval = ctx->write_full(oid, bl);
3527   tracepoint(librados, rados_write_full_exit, retval);
3528   return retval;
3529 }
3530
3531 extern "C" int rados_writesame(rados_ioctx_t io,
3532                                 const char *o,
3533                                 const char *buf,
3534                                 size_t data_len,
3535                                 size_t write_len,
3536                                 uint64_t off)
3537 {
3538   tracepoint(librados, rados_writesame_enter, io, o, buf, data_len, write_len, off);
3539   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3540   object_t oid(o);
3541   bufferlist bl;
3542   bl.append(buf, data_len);
3543   int retval = ctx->writesame(oid, bl, write_len, off);
3544   tracepoint(librados, rados_writesame_exit, retval);
3545   return retval;
3546 }
3547
3548 extern "C" int rados_trunc(rados_ioctx_t io, const char *o, uint64_t size)
3549 {
3550   tracepoint(librados, rados_trunc_enter, io, o, size);
3551   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3552   object_t oid(o);
3553   int retval = ctx->trunc(oid, size);
3554   tracepoint(librados, rados_trunc_exit, retval);
3555   return retval;
3556 }
3557
3558 extern "C" int rados_remove(rados_ioctx_t io, const char *o)
3559 {
3560   tracepoint(librados, rados_remove_enter, io, o);
3561   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3562   object_t oid(o);
3563   int retval = ctx->remove(oid);
3564   tracepoint(librados, rados_remove_exit, retval);
3565   return retval;
3566 }
3567
3568 extern "C" int rados_read(rados_ioctx_t io, const char *o, char *buf, size_t len, uint64_t off)
3569 {
3570   tracepoint(librados, rados_read_enter, io, o, buf, len, off);
3571   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3572   int ret;
3573   object_t oid(o);
3574
3575   bufferlist bl;
3576   bufferptr bp = buffer::create_static(len, buf);
3577   bl.push_back(bp);
3578
3579   ret = ctx->read(oid, bl, len, off);
3580   if (ret >= 0) {
3581     if (bl.length() > len) {
3582       tracepoint(librados, rados_read_exit, -ERANGE, NULL);
3583       return -ERANGE;
3584     }
3585     if (!bl.is_provided_buffer(buf))
3586       bl.copy(0, bl.length(), buf);
3587     ret = bl.length();    // hrm :/
3588   }
3589
3590   tracepoint(librados, rados_read_exit, ret, buf);
3591   return ret;
3592 }
3593
3594 extern "C" int rados_checksum(rados_ioctx_t io, const char *o,
3595                               rados_checksum_type_t type,
3596                               const char *init_value, size_t init_value_len,
3597                               size_t len, uint64_t off, size_t chunk_size,
3598                               char *pchecksum, size_t checksum_len)
3599 {
3600   tracepoint(librados, rados_checksum_enter, io, o, type, init_value,
3601              init_value_len, len, off, chunk_size);
3602   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3603   object_t oid(o);
3604
3605   bufferlist init_value_bl;
3606   init_value_bl.append(init_value, init_value_len);
3607
3608   bufferlist checksum_bl;
3609
3610   int retval = ctx->checksum(oid, get_checksum_op_type(type), init_value_bl,
3611                              len, off, chunk_size, &checksum_bl);
3612   if (retval >= 0) {
3613     if (checksum_bl.length() > checksum_len) {
3614       tracepoint(librados, rados_checksum_exit, -ERANGE, NULL, 0);
3615       return -ERANGE;
3616     }
3617
3618     checksum_bl.copy(0, checksum_bl.length(), pchecksum);
3619   }
3620   tracepoint(librados, rados_checksum_exit, retval, pchecksum, checksum_len);
3621   return retval;
3622 }
3623
3624 extern "C" uint64_t rados_get_last_version(rados_ioctx_t io)
3625 {
3626   tracepoint(librados, rados_get_last_version_enter, io);
3627   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3628   uint64_t retval = ctx->last_version();
3629   tracepoint(librados, rados_get_last_version_exit, retval);
3630   return retval;
3631 }
3632
3633 extern "C" int rados_pool_create(rados_t cluster, const char *name)
3634 {
3635   tracepoint(librados, rados_pool_create_enter, cluster, name);
3636   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3637   string sname(name);
3638   int retval = radosp->pool_create(sname);
3639   tracepoint(librados, rados_pool_create_exit, retval);
3640   return retval;
3641 }
3642
3643 extern "C" int rados_pool_create_with_auid(rados_t cluster, const char *name,
3644                                            uint64_t auid)
3645 {
3646   tracepoint(librados, rados_pool_create_with_auid_enter, cluster, name, auid);
3647   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3648   string sname(name);
3649   int retval = radosp->pool_create(sname, auid);
3650   tracepoint(librados, rados_pool_create_with_auid_exit, retval);
3651   return retval;
3652 }
3653
3654 extern "C" int rados_pool_create_with_crush_rule(rados_t cluster, const char *name,
3655                                                  __u8 crush_rule_num)
3656 {
3657   tracepoint(librados, rados_pool_create_with_crush_rule_enter, cluster, name, crush_rule_num);
3658   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3659   string sname(name);
3660   int retval = radosp->pool_create(sname, 0, crush_rule_num);
3661   tracepoint(librados, rados_pool_create_with_crush_rule_exit, retval);
3662   return retval;
3663 }
3664
3665 extern "C" int rados_pool_create_with_all(rados_t cluster, const char *name,
3666                                           uint64_t auid, __u8 crush_rule_num)
3667 {
3668   tracepoint(librados, rados_pool_create_with_all_enter, cluster, name, auid, crush_rule_num);
3669   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3670   string sname(name);
3671   int retval = radosp->pool_create(sname, auid, crush_rule_num);
3672   tracepoint(librados, rados_pool_create_with_all_exit, retval);
3673   return retval;
3674 }
3675
3676 extern "C" int rados_pool_get_base_tier(rados_t cluster, int64_t pool_id, int64_t* base_tier)
3677 {
3678   tracepoint(librados, rados_pool_get_base_tier_enter, cluster, pool_id);
3679   librados::RadosClient *client = (librados::RadosClient *)cluster;
3680   int retval = client->pool_get_base_tier(pool_id, base_tier);
3681   tracepoint(librados, rados_pool_get_base_tier_exit, retval, *base_tier);
3682   return retval;
3683 }
3684
3685 extern "C" int rados_pool_delete(rados_t cluster, const char *pool_name)
3686 {
3687   tracepoint(librados, rados_pool_delete_enter, cluster, pool_name);
3688   librados::RadosClient *client = (librados::RadosClient *)cluster;
3689   int retval = client->pool_delete(pool_name);
3690   tracepoint(librados, rados_pool_delete_exit, retval);
3691   return retval;
3692 }
3693
3694 extern "C" int rados_ioctx_pool_set_auid(rados_ioctx_t io, uint64_t auid)
3695 {
3696   tracepoint(librados, rados_ioctx_pool_set_auid_enter, io, auid);
3697   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3698   int retval = ctx->pool_change_auid(auid);
3699   tracepoint(librados, rados_ioctx_pool_set_auid_exit, retval);
3700   return retval;
3701 }
3702
3703 extern "C" int rados_ioctx_pool_get_auid(rados_ioctx_t io, uint64_t *auid)
3704 {
3705   tracepoint(librados, rados_ioctx_pool_get_auid_enter, io);
3706   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3707   int retval = ctx->client->pool_get_auid(ctx->get_id(), (unsigned long long *)auid);
3708   tracepoint(librados, rados_ioctx_pool_get_auid_exit, retval, *auid);
3709   return retval;
3710 }
3711
3712 extern "C" int rados_ioctx_pool_requires_alignment(rados_ioctx_t io)
3713 {
3714   tracepoint(librados, rados_ioctx_pool_requires_alignment_enter, io);
3715   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3716   int retval = ctx->client->pool_requires_alignment(ctx->get_id());
3717   tracepoint(librados, rados_ioctx_pool_requires_alignment_exit, retval);
3718   return retval;
3719 }
3720
3721 extern "C" int rados_ioctx_pool_requires_alignment2(rados_ioctx_t io,
3722         int *requires)
3723 {
3724   tracepoint(librados, rados_ioctx_pool_requires_alignment_enter2, io);
3725   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3726   bool requires_alignment;
3727   int retval = ctx->client->pool_requires_alignment2(ctx->get_id(),
3728         &requires_alignment);
3729   tracepoint(librados, rados_ioctx_pool_requires_alignment_exit2, retval,
3730         requires_alignment);
3731   if (requires)
3732     *requires = requires_alignment;
3733   return retval;
3734 }
3735
3736 extern "C" uint64_t rados_ioctx_pool_required_alignment(rados_ioctx_t io)
3737 {
3738   tracepoint(librados, rados_ioctx_pool_required_alignment_enter, io);
3739   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3740   uint64_t retval = ctx->client->pool_required_alignment(ctx->get_id());
3741   tracepoint(librados, rados_ioctx_pool_required_alignment_exit, retval);
3742   return retval;
3743 }
3744
3745 extern "C" int rados_ioctx_pool_required_alignment2(rados_ioctx_t io,
3746         uint64_t *alignment)
3747 {
3748   tracepoint(librados, rados_ioctx_pool_required_alignment_enter2, io);
3749   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3750   int retval = ctx->client->pool_required_alignment2(ctx->get_id(),
3751         alignment);
3752   tracepoint(librados, rados_ioctx_pool_required_alignment_exit2, retval,
3753         *alignment);
3754   return retval;
3755 }
3756
3757 extern "C" void rados_ioctx_locator_set_key(rados_ioctx_t io, const char *key)
3758 {
3759   tracepoint(librados, rados_ioctx_locator_set_key_enter, io, key);
3760   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3761   if (key)
3762     ctx->oloc.key = key;
3763   else
3764     ctx->oloc.key = "";
3765   tracepoint(librados, rados_ioctx_locator_set_key_exit);
3766 }
3767
3768 extern "C" void rados_ioctx_set_namespace(rados_ioctx_t io, const char *nspace)
3769 {
3770   tracepoint(librados, rados_ioctx_set_namespace_enter, io, nspace);
3771   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3772   if (nspace)
3773     ctx->oloc.nspace = nspace;
3774   else
3775     ctx->oloc.nspace = "";
3776   tracepoint(librados, rados_ioctx_set_namespace_exit);
3777 }
3778
3779 extern "C" rados_t rados_ioctx_get_cluster(rados_ioctx_t io)
3780 {
3781   tracepoint(librados, rados_ioctx_get_cluster_enter, io);
3782   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3783   rados_t retval = (rados_t)ctx->client;
3784   tracepoint(librados, rados_ioctx_get_cluster_exit, retval);
3785   return retval;
3786 }
3787
3788 extern "C" int64_t rados_ioctx_get_id(rados_ioctx_t io)
3789 {
3790   tracepoint(librados, rados_ioctx_get_id_enter, io);
3791   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3792   int64_t retval = ctx->get_id();
3793   tracepoint(librados, rados_ioctx_get_id_exit, retval);
3794   return retval;
3795 }
3796
3797 extern "C" int rados_ioctx_get_pool_name(rados_ioctx_t io, char *s, unsigned maxlen)
3798 {
3799   tracepoint(librados, rados_ioctx_get_pool_name_enter, io, maxlen);
3800   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3801   std::string pool_name;
3802
3803   int err = ctx->client->pool_get_name(ctx->get_id(), &pool_name);
3804   if (err) {
3805     tracepoint(librados, rados_ioctx_get_pool_name_exit, err, "");
3806     return err;
3807   }
3808   if (pool_name.length() >= maxlen) {
3809     tracepoint(librados, rados_ioctx_get_pool_name_exit, -ERANGE, "");
3810     return -ERANGE;
3811   }
3812   strcpy(s, pool_name.c_str());
3813   int retval = pool_name.length();
3814   tracepoint(librados, rados_ioctx_get_pool_name_exit, retval, s);
3815   return retval;
3816 }
3817
3818 // snaps
3819
3820 extern "C" int rados_ioctx_snap_create(rados_ioctx_t io, const char *snapname)
3821 {
3822   tracepoint(librados, rados_ioctx_snap_create_enter, io, snapname);
3823   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3824   int retval = ctx->snap_create(snapname);
3825   tracepoint(librados, rados_ioctx_snap_create_exit, retval);
3826   return retval;
3827 }
3828
3829 extern "C" int rados_ioctx_snap_remove(rados_ioctx_t io, const char *snapname)
3830 {
3831   tracepoint(librados, rados_ioctx_snap_remove_enter, io, snapname);
3832   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3833   int retval = ctx->snap_remove(snapname);
3834   tracepoint(librados, rados_ioctx_snap_remove_exit, retval);
3835   return retval;
3836 }
3837
3838 extern "C" int rados_ioctx_snap_rollback(rados_ioctx_t io, const char *oid,
3839                               const char *snapname)
3840 {
3841   tracepoint(librados, rados_ioctx_snap_rollback_enter, io, oid, snapname);
3842   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3843   int retval = ctx->rollback(oid, snapname);
3844   tracepoint(librados, rados_ioctx_snap_rollback_exit, retval);
3845   return retval;
3846 }
3847
3848 // Deprecated name kept for backward compatibility
3849 extern "C" int rados_rollback(rados_ioctx_t io, const char *oid,
3850                               const char *snapname)
3851 {
3852   return rados_ioctx_snap_rollback(io, oid, snapname);
3853 }
3854
3855 extern "C" int rados_ioctx_selfmanaged_snap_create(rados_ioctx_t io,
3856                                              uint64_t *snapid)
3857 {
3858   tracepoint(librados, rados_ioctx_selfmanaged_snap_create_enter, io);
3859   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3860   int retval = ctx->selfmanaged_snap_create(snapid);
3861   tracepoint(librados, rados_ioctx_selfmanaged_snap_create_exit, retval, *snapid);
3862   return retval;
3863 }
3864
3865 extern "C" void
3866 rados_aio_ioctx_selfmanaged_snap_create(rados_ioctx_t io,
3867                                         rados_snap_t *snapid,
3868                                         rados_completion_t completion)
3869 {
3870   tracepoint(librados, rados_ioctx_selfmanaged_snap_create_enter, io);
3871   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3872   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
3873   ctx->aio_selfmanaged_snap_create(snapid, c);
3874   tracepoint(librados, rados_ioctx_selfmanaged_snap_create_exit, 0, 0);
3875 }
3876
3877 extern "C" int rados_ioctx_selfmanaged_snap_remove(rados_ioctx_t io,
3878                                              uint64_t snapid)
3879 {
3880   tracepoint(librados, rados_ioctx_selfmanaged_snap_remove_enter, io, snapid);
3881   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3882   int retval = ctx->selfmanaged_snap_remove(snapid);
3883   tracepoint(librados, rados_ioctx_selfmanaged_snap_remove_exit, retval);
3884   return retval;
3885 }
3886
3887 extern "C" void
3888 rados_aio_ioctx_selfmanaged_snap_remove(rados_ioctx_t io,
3889                                         rados_snap_t snapid,
3890                                         rados_completion_t completion)
3891 {
3892   tracepoint(librados, rados_ioctx_selfmanaged_snap_remove_enter, io, snapid);
3893   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3894   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
3895   ctx->aio_selfmanaged_snap_remove(snapid, c);
3896   tracepoint(librados, rados_ioctx_selfmanaged_snap_remove_exit, 0);
3897 }
3898
3899 extern "C" int rados_ioctx_selfmanaged_snap_rollback(rados_ioctx_t io,
3900                                                      const char *oid,
3901                                                      uint64_t snapid)
3902 {
3903   tracepoint(librados, rados_ioctx_selfmanaged_snap_rollback_enter, io, oid, snapid);
3904   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3905   int retval = ctx->selfmanaged_snap_rollback_object(oid, ctx->snapc, snapid);
3906   tracepoint(librados, rados_ioctx_selfmanaged_snap_rollback_exit, retval);
3907   return retval;
3908 }
3909
3910 extern "C" int rados_ioctx_snap_list(rados_ioctx_t io, rados_snap_t *snaps,
3911                                     int maxlen)
3912 {
3913   tracepoint(librados, rados_ioctx_snap_list_enter, io, maxlen);
3914   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3915   vector<uint64_t> snapvec;
3916   int r = ctx->snap_list(&snapvec);
3917   if (r < 0) {
3918     tracepoint(librados, rados_ioctx_snap_list_exit, r, snaps, 0);
3919     return r;
3920   }
3921   if ((int)snapvec.size() <= maxlen) {
3922     for (unsigned i=0; i<snapvec.size(); i++) {
3923       snaps[i] = snapvec[i];
3924     }
3925     int retval = snapvec.size();
3926     tracepoint(librados, rados_ioctx_snap_list_exit, retval, snaps, retval);
3927     return retval;
3928   }
3929   int retval = -ERANGE;
3930   tracepoint(librados, rados_ioctx_snap_list_exit, retval, snaps, 0);
3931   return retval;
3932 }
3933
3934 extern "C" int rados_ioctx_snap_lookup(rados_ioctx_t io, const char *name,
3935                                       rados_snap_t *id)
3936 {
3937   tracepoint(librados, rados_ioctx_snap_lookup_enter, io, name);
3938   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3939   int retval = ctx->snap_lookup(name, (uint64_t *)id);
3940   tracepoint(librados, rados_ioctx_snap_lookup_exit, retval, *id);
3941   return retval;
3942 }
3943
3944 extern "C" int rados_ioctx_snap_get_name(rados_ioctx_t io, rados_snap_t id,
3945                                         char *name, int maxlen)
3946 {
3947   tracepoint(librados, rados_ioctx_snap_get_name_enter, io, id, maxlen);
3948   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3949   std::string sname;
3950   int r = ctx->snap_get_name(id, &sname);
3951   if (r < 0) {
3952     tracepoint(librados, rados_ioctx_snap_get_name_exit, r, "");
3953     return r;
3954   }
3955   if ((int)sname.length() >= maxlen) {
3956     int retval = -ERANGE;
3957     tracepoint(librados, rados_ioctx_snap_get_name_exit, retval, "");
3958     return retval;
3959   }
3960   strncpy(name, sname.c_str(), maxlen);
3961   tracepoint(librados, rados_ioctx_snap_get_name_exit, 0, name);
3962   return 0;
3963 }
3964
3965 extern "C" int rados_ioctx_snap_get_stamp(rados_ioctx_t io, rados_snap_t id, time_t *t)
3966 {
3967   tracepoint(librados, rados_ioctx_snap_get_stamp_enter, io, id);
3968   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3969   int retval = ctx->snap_get_stamp(id, t);
3970   tracepoint(librados, rados_ioctx_snap_get_stamp_exit, retval, *t);
3971   return retval;
3972 }
3973
3974 extern "C" int rados_cmpext(rados_ioctx_t io, const char *o,
3975                             const char *cmp_buf, size_t cmp_len, uint64_t off)
3976 {
3977   tracepoint(librados, rados_cmpext_enter, io, o, cmp_buf, cmp_len, off);
3978   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3979   int ret;
3980   object_t oid(o);
3981
3982   bufferlist cmp_bl;
3983   cmp_bl.append(cmp_buf, cmp_len);
3984
3985   ret = ctx->cmpext(oid, off, cmp_bl);
3986   tracepoint(librados, rados_cmpext_exit, ret);
3987
3988   return ret;
3989 }
3990
3991 extern "C" int rados_getxattr(rados_ioctx_t io, const char *o, const char *name,
3992                               char *buf, size_t len)
3993 {
3994   tracepoint(librados, rados_getxattr_enter, io, o, name, len);
3995   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3996   int ret;
3997   object_t oid(o);
3998   bufferlist bl;
3999   bl.push_back(buffer::create_static(len, buf));
4000   ret = ctx->getxattr(oid, name, bl);
4001   if (ret >= 0) {
4002     if (bl.length() > len) {
4003       tracepoint(librados, rados_getxattr_exit, -ERANGE, buf, 0);
4004       return -ERANGE;
4005     }
4006     if (!bl.is_provided_buffer(buf))
4007       bl.copy(0, bl.length(), buf);
4008     ret = bl.length();
4009   }
4010
4011   tracepoint(librados, rados_getxattr_exit, ret, buf, ret);
4012   return ret;
4013 }
4014
4015 extern "C" int rados_getxattrs(rados_ioctx_t io, const char *oid,
4016                                rados_xattrs_iter_t *iter)
4017 {
4018   tracepoint(librados, rados_getxattrs_enter, io, oid);
4019   librados::RadosXattrsIter *it = new librados::RadosXattrsIter();
4020   if (!it) {
4021     tracepoint(librados, rados_getxattrs_exit, -ENOMEM, NULL);
4022     return -ENOMEM;
4023   }
4024   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4025   object_t obj(oid);
4026   int ret = ctx->getxattrs(obj, it->attrset);
4027   if (ret) {
4028     delete it;
4029     tracepoint(librados, rados_getxattrs_exit, ret, NULL);
4030     return ret;
4031   }
4032   it->i = it->attrset.begin();
4033
4034   librados::RadosXattrsIter **iret = (librados::RadosXattrsIter**)iter;
4035   *iret = it;
4036   *iter = it;
4037   tracepoint(librados, rados_getxattrs_exit, 0, *iter);
4038   return 0;
4039 }
4040
4041 extern "C" int rados_getxattrs_next(rados_xattrs_iter_t iter,
4042                                     const char **name, const char **val, size_t *len)
4043 {
4044   tracepoint(librados, rados_getxattrs_next_enter, iter);
4045   librados::RadosXattrsIter *it = static_cast<librados::RadosXattrsIter*>(iter);
4046   if (it->i == it->attrset.end()) {
4047     *name = NULL;
4048     *val = NULL;
4049     *len = 0;
4050     tracepoint(librados, rados_getxattrs_next_exit, 0, NULL, NULL, 0);
4051     return 0;
4052   }
4053   free(it->val);
4054   const std::string &s(it->i->first);
4055   *name = s.c_str();
4056   bufferlist &bl(it->i->second);
4057   size_t bl_len = bl.length();
4058   if (!bl_len) {
4059     // malloc(0) is not guaranteed to return a valid pointer
4060     *val = (char *)NULL;
4061   } else {
4062     it->val = (char*)malloc(bl_len);
4063     if (!it->val) {
4064       tracepoint(librados, rados_getxattrs_next_exit, -ENOMEM, *name, NULL, 0);
4065       return -ENOMEM;
4066     }
4067     memcpy(it->val, bl.c_str(), bl_len);
4068     *val = it->val;
4069   }
4070   *len = bl_len;
4071   ++it->i;
4072   tracepoint(librados, rados_getxattrs_next_exit, 0, *name, *val, *len);
4073   return 0;
4074 }
4075
4076 extern "C" void rados_getxattrs_end(rados_xattrs_iter_t iter)
4077 {
4078   tracepoint(librados, rados_getxattrs_end_enter, iter);
4079   librados::RadosXattrsIter *it = static_cast<librados::RadosXattrsIter*>(iter);
4080   delete it;
4081   tracepoint(librados, rados_getxattrs_end_exit);
4082 }
4083
4084 extern "C" int rados_setxattr(rados_ioctx_t io, const char *o, const char *name, const char *buf, size_t len)
4085 {
4086   tracepoint(librados, rados_setxattr_enter, io, o, name, buf, len);
4087   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4088   object_t oid(o);
4089   bufferlist bl;
4090   bl.append(buf, len);
4091   int retval = ctx->setxattr(oid, name, bl);
4092   tracepoint(librados, rados_setxattr_exit, retval);
4093   return retval;
4094 }
4095
4096 extern "C" int rados_rmxattr(rados_ioctx_t io, const char *o, const char *name)
4097 {
4098   tracepoint(librados, rados_rmxattr_enter, io, o, name);
4099   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4100   object_t oid(o);
4101   int retval = ctx->rmxattr(oid, name);
4102   tracepoint(librados, rados_rmxattr_exit, retval);
4103   return retval;
4104 }
4105
4106 extern "C" int rados_stat(rados_ioctx_t io, const char *o, uint64_t *psize, time_t *pmtime)
4107 {
4108   tracepoint(librados, rados_stat_enter, io, o);
4109   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4110   object_t oid(o);
4111   int retval = ctx->stat(oid, psize, pmtime);
4112   tracepoint(librados, rados_stat_exit, retval, psize, pmtime);
4113   return retval;
4114 }
4115
4116 extern "C" int rados_tmap_update(rados_ioctx_t io, const char *o, const char *cmdbuf, size_t cmdbuflen)
4117 {
4118   tracepoint(librados, rados_tmap_update_enter, io, o, cmdbuf, cmdbuflen);
4119   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4120   object_t oid(o);
4121   bufferlist cmdbl;
4122   cmdbl.append(cmdbuf, cmdbuflen);
4123   int retval = ctx->tmap_update(oid, cmdbl);
4124   tracepoint(librados, rados_tmap_update_exit, retval);
4125   return retval;
4126 }
4127
4128 extern "C" int rados_tmap_put(rados_ioctx_t io, const char *o, const char *buf, size_t buflen)
4129 {
4130   tracepoint(librados, rados_tmap_put_enter, io, o, buf, buflen);
4131   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4132   object_t oid(o);
4133   bufferlist bl;
4134   bl.append(buf, buflen);
4135   int retval = ctx->tmap_put(oid, bl);
4136   tracepoint(librados, rados_tmap_put_exit, retval);
4137   return retval;
4138 }
4139
4140 extern "C" int rados_tmap_get(rados_ioctx_t io, const char *o, char *buf, size_t buflen)
4141 {
4142   tracepoint(librados, rados_tmap_get_enter, io, o, buflen);
4143   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4144   object_t oid(o);
4145   bufferlist bl;
4146   int r = ctx->tmap_get(oid, bl);
4147   if (r < 0) {
4148     tracepoint(librados, rados_tmap_get_exit, r, buf, 0);
4149     return r;
4150   }
4151   if (bl.length() > buflen) {
4152     tracepoint(librados, rados_tmap_get_exit, -ERANGE, buf, 0);
4153     return -ERANGE;
4154   }
4155   bl.copy(0, bl.length(), buf);
4156   int retval = bl.length();
4157   tracepoint(librados, rados_tmap_get_exit, retval, buf, retval);
4158   return retval;
4159 }
4160
4161 extern "C" int rados_tmap_to_omap(rados_ioctx_t io, const char *o, bool nullok)
4162 {
4163   tracepoint(librados, rados_tmap_to_omap_enter, io, o, nullok);
4164   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4165   object_t oid(o);
4166   int retval = ctx->tmap_to_omap(oid, nullok);
4167   tracepoint(librados, rados_tmap_to_omap_exit, retval);
4168   return retval;
4169 }
4170
4171 extern "C" int rados_exec(rados_ioctx_t io, const char *o, const char *cls, const char *method,
4172                          const char *inbuf, size_t in_len, char *buf, size_t out_len)
4173 {
4174   tracepoint(librados, rados_exec_enter, io, o, cls, method, inbuf, in_len, out_len);
4175   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4176   object_t oid(o);
4177   bufferlist inbl, outbl;
4178   int ret;
4179   inbl.append(inbuf, in_len);
4180   ret = ctx->exec(oid, cls, method, inbl, outbl);
4181   if (ret >= 0) {
4182     if (outbl.length()) {
4183       if (outbl.length() > out_len) {
4184         tracepoint(librados, rados_exec_exit, -ERANGE, buf, 0);
4185         return -ERANGE;
4186       }
4187       outbl.copy(0, outbl.length(), buf);
4188       ret = outbl.length();   // hrm :/
4189     }
4190   }
4191   tracepoint(librados, rados_exec_exit, ret, buf, ret);
4192   return ret;
4193 }
4194
4195 extern "C" rados_object_list_cursor rados_object_list_begin(rados_ioctx_t io)
4196 {
4197   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4198
4199   hobject_t *result = new hobject_t(ctx->objecter->enumerate_objects_begin());
4200   return (rados_object_list_cursor)result;
4201 }
4202
4203 extern "C" rados_object_list_cursor rados_object_list_end(rados_ioctx_t io)
4204 {
4205   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4206
4207   hobject_t *result = new hobject_t(ctx->objecter->enumerate_objects_end());
4208   return (rados_object_list_cursor)result;
4209 }
4210
4211 extern "C" int rados_object_list_is_end(
4212     rados_ioctx_t io, rados_object_list_cursor cur)
4213 {
4214   hobject_t *hobj = (hobject_t*)cur;
4215   return hobj->is_max();
4216 }
4217
4218 extern "C" void rados_object_list_cursor_free(
4219     rados_ioctx_t io, rados_object_list_cursor cur)
4220 {
4221   hobject_t *hobj = (hobject_t*)cur;
4222   delete hobj;
4223 }
4224
4225 extern "C" int rados_object_list_cursor_cmp(
4226     rados_ioctx_t io,
4227     rados_object_list_cursor lhs_cur,
4228     rados_object_list_cursor rhs_cur)
4229 {
4230   hobject_t *lhs = (hobject_t*)lhs_cur;
4231   hobject_t *rhs = (hobject_t*)rhs_cur;
4232   return cmp(*lhs, *rhs);
4233 }
4234
4235 extern "C" int rados_object_list(rados_ioctx_t io,
4236     const rados_object_list_cursor start,
4237     const rados_object_list_cursor finish,
4238     const size_t result_item_count,
4239     const char *filter_buf,
4240     const size_t filter_buf_len,
4241     rados_object_list_item *result_items,
4242     rados_object_list_cursor *next)
4243 {
4244   assert(next);
4245
4246   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4247
4248   // Zero out items so that they will be safe to free later
4249   memset(result_items, 0, sizeof(rados_object_list_item) * result_item_count);
4250
4251   std::list<librados::ListObjectImpl> result;
4252   hobject_t next_hash;
4253
4254   bufferlist filter_bl;
4255   if (filter_buf != nullptr) {
4256     filter_bl.append(filter_buf, filter_buf_len);
4257   }
4258
4259   C_SaferCond cond;
4260   ctx->objecter->enumerate_objects(
4261       ctx->poolid,
4262       ctx->oloc.nspace,
4263       *((hobject_t*)start),
4264       *((hobject_t*)finish),
4265       result_item_count,
4266       filter_bl,
4267       &result,
4268       &next_hash,
4269       &cond);
4270
4271   hobject_t *next_hobj = (hobject_t*)(*next);
4272   assert(next_hobj);
4273
4274   int r = cond.wait();
4275   if (r < 0) {
4276     *next_hobj = hobject_t::get_max();
4277     return r;
4278   }
4279
4280   assert(result.size() <= result_item_count);  // Don't overflow!
4281
4282   int k = 0;
4283   for (std::list<librados::ListObjectImpl>::iterator i = result.begin();
4284        i != result.end(); ++i) {
4285     rados_object_list_item &item = result_items[k++];
4286     do_out_buffer(i->oid, &item.oid, &item.oid_length);
4287     do_out_buffer(i->nspace, &item.nspace, &item.nspace_length);
4288     do_out_buffer(i->locator, &item.locator, &item.locator_length);
4289   }
4290
4291   *next_hobj = next_hash;
4292
4293   return result.size();
4294 }
4295
4296 extern "C" void rados_object_list_free(
4297     const size_t result_size,
4298     rados_object_list_item *results)
4299 {
4300   assert(results);
4301
4302   for (unsigned int i = 0; i < result_size; ++i) {
4303     rados_buffer_free(results[i].oid);
4304     rados_buffer_free(results[i].locator);
4305     rados_buffer_free(results[i].nspace);
4306   }
4307 }
4308
4309 /* list objects */
4310
4311 extern "C" int rados_nobjects_list_open(rados_ioctx_t io, rados_list_ctx_t *listh)
4312 {
4313   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4314
4315   tracepoint(librados, rados_nobjects_list_open_enter, io);
4316
4317   Objecter::NListContext *h = new Objecter::NListContext;
4318   h->pool_id = ctx->poolid;
4319   h->pool_snap_seq = ctx->snap_seq;
4320   h->nspace = ctx->oloc.nspace; // After dropping compatibility need nspace
4321   *listh = (void *)new librados::ObjListCtx(ctx, h);
4322   tracepoint(librados, rados_nobjects_list_open_exit, 0, *listh);
4323   return 0;
4324 }
4325
4326 extern "C" void rados_nobjects_list_close(rados_list_ctx_t h)
4327 {
4328   tracepoint(librados, rados_nobjects_list_close_enter, h);
4329   librados::ObjListCtx *lh = (librados::ObjListCtx *)h;
4330   delete lh;
4331   tracepoint(librados, rados_nobjects_list_close_exit);
4332 }
4333
4334 extern "C" uint32_t rados_nobjects_list_seek(rados_list_ctx_t listctx,
4335                                             uint32_t pos)
4336 {
4337   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4338   tracepoint(librados, rados_nobjects_list_seek_enter, listctx, pos);
4339   uint32_t r = lh->ctx->nlist_seek(lh->nlc, pos);
4340   tracepoint(librados, rados_nobjects_list_seek_exit, r);
4341   return r;
4342 }
4343
4344 extern "C" uint32_t rados_nobjects_list_seek_cursor(rados_list_ctx_t listctx,
4345                                                     rados_object_list_cursor cursor)
4346 {
4347   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4348
4349   tracepoint(librados, rados_nobjects_list_seek_cursor_enter, listctx);
4350   uint32_t r = lh->ctx->nlist_seek(lh->nlc, cursor);
4351   tracepoint(librados, rados_nobjects_list_seek_cursor_exit, r);
4352   return r;
4353 }
4354
4355 extern "C" int rados_nobjects_list_get_cursor(rados_list_ctx_t listctx,
4356                                               rados_object_list_cursor *cursor)
4357 {
4358   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4359
4360   tracepoint(librados, rados_nobjects_list_get_cursor_enter, listctx);
4361   *cursor = lh->ctx->nlist_get_cursor(lh->nlc);
4362   tracepoint(librados, rados_nobjects_list_get_cursor_exit, 0);
4363   return 0;
4364 }
4365
4366 extern "C" uint32_t rados_nobjects_list_get_pg_hash_position(
4367   rados_list_ctx_t listctx)
4368 {
4369   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4370   tracepoint(librados, rados_nobjects_list_get_pg_hash_position_enter, listctx);
4371   uint32_t retval = lh->nlc->get_pg_hash_position();
4372   tracepoint(librados, rados_nobjects_list_get_pg_hash_position_exit, retval);
4373   return retval;
4374 }
4375
4376 extern "C" int rados_nobjects_list_next(rados_list_ctx_t listctx, const char **entry, const char **key, const char **nspace)
4377 {
4378   tracepoint(librados, rados_nobjects_list_next_enter, listctx);
4379   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4380   Objecter::NListContext *h = lh->nlc;
4381
4382   // if the list is non-empty, this method has been called before
4383   if (!h->list.empty())
4384     // so let's kill the previously-returned object
4385     h->list.pop_front();
4386
4387   if (h->list.empty()) {
4388     int ret = lh->ctx->nlist(lh->nlc, RADOS_LIST_MAX_ENTRIES);
4389     if (ret < 0) {
4390       tracepoint(librados, rados_nobjects_list_next_exit, ret, NULL, NULL, NULL);
4391       return ret;
4392     }
4393     if (h->list.empty()) {
4394       tracepoint(librados, rados_nobjects_list_next_exit, -ENOENT, NULL, NULL, NULL);
4395       return -ENOENT;
4396     }
4397   }
4398
4399   *entry = h->list.front().oid.c_str();
4400
4401   if (key) {
4402     if (h->list.front().locator.size())
4403       *key = h->list.front().locator.c_str();
4404     else
4405       *key = NULL;
4406   }
4407   if (nspace)
4408     *nspace = h->list.front().nspace.c_str();
4409   tracepoint(librados, rados_nobjects_list_next_exit, 0, *entry, key, nspace);
4410   return 0;
4411 }
4412
4413
4414 /*
4415  * removed legacy v2 list objects stubs
4416  *
4417  * thse return -ENOTSUP where possible.
4418  */
4419 extern "C" int rados_objects_list_open(
4420   rados_ioctx_t io,
4421   rados_list_ctx_t *ctx)
4422 {
4423   return -ENOTSUP;
4424 }
4425
4426 extern "C" uint32_t rados_objects_list_get_pg_hash_position(
4427   rados_list_ctx_t ctx)
4428 {
4429   return 0;
4430 }
4431
4432 extern "C" uint32_t rados_objects_list_seek(
4433   rados_list_ctx_t ctx,
4434   uint32_t pos)
4435 {
4436   return 0;
4437 }
4438
4439 extern "C" int rados_objects_list_next(
4440   rados_list_ctx_t ctx,
4441   const char **entry,
4442   const char **key)
4443 {
4444   return -ENOTSUP;
4445 }
4446
4447 extern "C" void rados_objects_list_close(
4448   rados_list_ctx_t ctx)
4449 {
4450 }
4451
4452
4453 // -------------------------
4454 // aio
4455
4456 extern "C" int rados_aio_create_completion(void *cb_arg,
4457                                            rados_callback_t cb_complete,
4458                                            rados_callback_t cb_safe,
4459                                            rados_completion_t *pc)
4460 {
4461   tracepoint(librados, rados_aio_create_completion_enter, cb_arg, cb_complete, cb_safe);
4462   librados::AioCompletionImpl *c = new librados::AioCompletionImpl;
4463   if (cb_complete)
4464     c->set_complete_callback(cb_arg, cb_complete);
4465   if (cb_safe)
4466     c->set_safe_callback(cb_arg, cb_safe);
4467   *pc = c;
4468   tracepoint(librados, rados_aio_create_completion_exit, 0, *pc);
4469   return 0;
4470 }
4471
4472 extern "C" int rados_aio_wait_for_complete(rados_completion_t c)
4473 {
4474   tracepoint(librados, rados_aio_wait_for_complete_enter, c);
4475   int retval = ((librados::AioCompletionImpl*)c)->wait_for_complete();
4476   tracepoint(librados, rados_aio_wait_for_complete_exit, retval);
4477   return retval;
4478 }
4479
4480 extern "C" int rados_aio_wait_for_safe(rados_completion_t c)
4481 {
4482   tracepoint(librados, rados_aio_wait_for_safe_enter, c);
4483   int retval = ((librados::AioCompletionImpl*)c)->wait_for_safe();
4484   tracepoint(librados, rados_aio_wait_for_safe_exit, retval);
4485   return retval;
4486 }
4487
4488 extern "C" int rados_aio_is_complete(rados_completion_t c)
4489 {
4490   tracepoint(librados, rados_aio_is_complete_enter, c);
4491   int retval = ((librados::AioCompletionImpl*)c)->is_complete();
4492   tracepoint(librados, rados_aio_is_complete_exit, retval);
4493   return retval;
4494 }
4495
4496 extern "C" int rados_aio_is_safe(rados_completion_t c)
4497 {
4498   tracepoint(librados, rados_aio_is_safe_enter, c);
4499   int retval = ((librados::AioCompletionImpl*)c)->is_safe();
4500   tracepoint(librados, rados_aio_is_safe_exit, retval);
4501   return retval;
4502 }
4503
4504 extern "C" int rados_aio_wait_for_complete_and_cb(rados_completion_t c)
4505 {
4506   tracepoint(librados, rados_aio_wait_for_complete_and_cb_enter, c);
4507   int retval = ((librados::AioCompletionImpl*)c)->wait_for_complete_and_cb();
4508   tracepoint(librados, rados_aio_wait_for_complete_and_cb_exit, retval);
4509   return retval;
4510 }
4511
4512 extern "C" int rados_aio_wait_for_safe_and_cb(rados_completion_t c)
4513 {
4514   tracepoint(librados, rados_aio_wait_for_safe_and_cb_enter, c);
4515   int retval = ((librados::AioCompletionImpl*)c)->wait_for_safe_and_cb();
4516   tracepoint(librados, rados_aio_wait_for_safe_and_cb_exit, retval);
4517   return retval;
4518 }
4519
4520 extern "C" int rados_aio_is_complete_and_cb(rados_completion_t c)
4521 {
4522   tracepoint(librados, rados_aio_is_complete_and_cb_enter, c);
4523   int retval = ((librados::AioCompletionImpl*)c)->is_complete_and_cb();
4524   tracepoint(librados, rados_aio_is_complete_and_cb_exit, retval);
4525   return retval;
4526 }
4527
4528 extern "C" int rados_aio_is_safe_and_cb(rados_completion_t c)
4529 {
4530   tracepoint(librados, rados_aio_is_safe_and_cb_enter, c);
4531   int retval = ((librados::AioCompletionImpl*)c)->is_safe_and_cb();
4532   tracepoint(librados, rados_aio_is_safe_and_cb_exit, retval);
4533   return retval;
4534 }
4535
4536 extern "C" int rados_aio_get_return_value(rados_completion_t c)
4537 {
4538   tracepoint(librados, rados_aio_get_return_value_enter, c);
4539   int retval = ((librados::AioCompletionImpl*)c)->get_return_value();
4540   tracepoint(librados, rados_aio_get_return_value_exit, retval);
4541   return retval;
4542 }
4543
4544 extern "C" uint64_t rados_aio_get_version(rados_completion_t c)
4545 {
4546   tracepoint(librados, rados_aio_get_version_enter, c);
4547   uint64_t retval = ((librados::AioCompletionImpl*)c)->get_version();
4548   tracepoint(librados, rados_aio_get_version_exit, retval);
4549   return retval;
4550 }
4551
4552 extern "C" void rados_aio_release(rados_completion_t c)
4553 {
4554   tracepoint(librados, rados_aio_release_enter, c);
4555   ((librados::AioCompletionImpl*)c)->put();
4556   tracepoint(librados, rados_aio_release_exit);
4557 }
4558
4559 extern "C" int rados_aio_read(rados_ioctx_t io, const char *o,
4560                                rados_completion_t completion,
4561                                char *buf, size_t len, uint64_t off)
4562 {
4563   tracepoint(librados, rados_aio_read_enter, io, o, completion, len, off);
4564   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4565   object_t oid(o);
4566   int retval = ctx->aio_read(oid, (librados::AioCompletionImpl*)completion,
4567                        buf, len, off, ctx->snap_seq);
4568   tracepoint(librados, rados_aio_read_exit, retval);
4569   return retval;
4570 }
4571
4572 #ifdef WITH_BLKIN
4573 extern "C" int rados_aio_read_traced(rados_ioctx_t io, const char *o,
4574                                      rados_completion_t completion,
4575                                      char *buf, size_t len, uint64_t off,
4576                                      struct blkin_trace_info *info)
4577 {
4578   tracepoint(librados, rados_aio_read_enter, io, o, completion, len, off);
4579   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4580   object_t oid(o);
4581   int retval = ctx->aio_read(oid, (librados::AioCompletionImpl*)completion,
4582                              buf, len, off, ctx->snap_seq, info);
4583   tracepoint(librados, rados_aio_read_exit, retval);
4584   return retval;
4585 }
4586 #endif
4587
4588 extern "C" int rados_aio_write(rados_ioctx_t io, const char *o,
4589                                 rados_completion_t completion,
4590                                 const char *buf, size_t len, uint64_t off)
4591 {
4592   tracepoint(librados, rados_aio_write_enter, io, o, completion, buf, len, off);
4593   if (len > UINT_MAX/2)
4594     return -E2BIG;
4595   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4596   object_t oid(o);
4597   bufferlist bl;
4598   bl.append(buf, len);
4599   int retval = ctx->aio_write(oid, (librados::AioCompletionImpl*)completion,
4600                         bl, len, off);
4601   tracepoint(librados, rados_aio_write_exit, retval);
4602   return retval;
4603 }
4604
4605 #ifdef WITH_BLKIN
4606 extern "C" int rados_aio_write_traced(rados_ioctx_t io, const char *o,
4607                                       rados_completion_t completion,
4608                                       const char *buf, size_t len, uint64_t off,
4609                                       struct blkin_trace_info *info)
4610 {
4611   tracepoint(librados, rados_aio_write_enter, io, o, completion, buf, len, off);
4612   if (len > UINT_MAX/2)
4613     return -E2BIG;
4614   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4615   object_t oid(o);
4616   bufferlist bl;
4617   bl.append(buf, len);
4618   int retval = ctx->aio_write(oid, (librados::AioCompletionImpl*)completion,
4619                               bl, len, off, info);
4620   tracepoint(librados, rados_aio_write_exit, retval);
4621   return retval;
4622 }
4623 #endif
4624
4625 extern "C" int rados_aio_append(rados_ioctx_t io, const char *o,
4626                                 rados_completion_t completion,
4627                                 const char *buf, size_t len)
4628 {
4629   tracepoint(librados, rados_aio_append_enter, io, o, completion, buf, len);
4630   if (len > UINT_MAX/2)
4631     return -E2BIG;
4632   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4633   object_t oid(o);
4634   bufferlist bl;
4635   bl.append(buf, len);
4636   int retval = ctx->aio_append(oid, (librados::AioCompletionImpl*)completion,
4637                          bl, len);
4638   tracepoint(librados, rados_aio_append_exit, retval);
4639   return retval;
4640 }
4641
4642 extern "C" int rados_aio_write_full(rados_ioctx_t io, const char *o,
4643                                     rados_completion_t completion,
4644                                     const char *buf, size_t len)
4645 {
4646   tracepoint(librados, rados_aio_write_full_enter, io, o, completion, buf, len);
4647   if (len > UINT_MAX/2)
4648     return -E2BIG;
4649   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4650   object_t oid(o);
4651   bufferlist bl;
4652   bl.append(buf, len);
4653   int retval = ctx->aio_write_full(oid, (librados::AioCompletionImpl*)completion, bl);
4654   tracepoint(librados, rados_aio_write_full_exit, retval);
4655   return retval;
4656 }
4657
4658 extern "C" int rados_aio_writesame(rados_ioctx_t io, const char *o,
4659                                    rados_completion_t completion,
4660                                    const char *buf, size_t data_len,
4661                                    size_t write_len, uint64_t off)
4662 {
4663   tracepoint(librados, rados_aio_writesame_enter, io, o, completion, buf,
4664                                                 data_len, write_len, off);
4665   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4666   object_t oid(o);
4667   bufferlist bl;
4668   bl.append(buf, data_len);
4669   int retval = ctx->aio_writesame(o, (librados::AioCompletionImpl*)completion,
4670                                   bl, write_len, off);
4671   tracepoint(librados, rados_aio_writesame_exit, retval);
4672   return retval;
4673 }
4674
4675 extern "C" int rados_aio_remove(rados_ioctx_t io, const char *o,
4676                                 rados_completion_t completion)
4677 {
4678   tracepoint(librados, rados_aio_remove_enter, io, o, completion);
4679   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4680   object_t oid(o);
4681   int retval = ctx->aio_remove(oid, (librados::AioCompletionImpl*)completion);
4682   tracepoint(librados, rados_aio_remove_exit, retval);
4683   return retval;
4684 }
4685
4686 extern "C" int rados_aio_flush_async(rados_ioctx_t io,
4687                                      rados_completion_t completion)
4688 {
4689   tracepoint(librados, rados_aio_flush_async_enter, io, completion);
4690   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4691   ctx->flush_aio_writes_async((librados::AioCompletionImpl*)completion);
4692   tracepoint(librados, rados_aio_flush_async_exit, 0);
4693   return 0;
4694 }
4695
4696 extern "C" int rados_aio_flush(rados_ioctx_t io)
4697 {
4698   tracepoint(librados, rados_aio_flush_enter, io);
4699   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4700   ctx->flush_aio_writes();
4701   tracepoint(librados, rados_aio_flush_exit, 0);
4702   return 0;
4703 }
4704
4705 struct AioGetxattrData {
4706   AioGetxattrData(char* buf, rados_completion_t c, size_t l) :
4707     user_buf(buf), len(l), user_completion((librados::AioCompletionImpl*)c) {}
4708   bufferlist bl;
4709   char* user_buf;
4710   size_t len;
4711   struct librados::C_AioCompleteAndSafe user_completion;
4712 };
4713
4714 static void rados_aio_getxattr_complete(rados_completion_t c, void *arg) {
4715   AioGetxattrData *cdata = reinterpret_cast<AioGetxattrData*>(arg);
4716   int rc = rados_aio_get_return_value(c);
4717   if (rc >= 0) {
4718     if (cdata->bl.length() > cdata->len) {
4719       rc = -ERANGE;
4720     } else {
4721       if (!cdata->bl.is_provided_buffer(cdata->user_buf))
4722         cdata->bl.copy(0, cdata->bl.length(), cdata->user_buf);
4723       rc = cdata->bl.length();
4724     }
4725   }
4726   cdata->user_completion.finish(rc);
4727   delete cdata;
4728 }
4729
4730 extern "C" int rados_aio_getxattr(rados_ioctx_t io, const char *o,
4731                                   rados_completion_t completion,
4732                                   const char *name, char *buf, size_t len)
4733 {
4734   tracepoint(librados, rados_aio_getxattr_enter, io, o, completion, name, len);
4735   // create data object to be passed to async callback
4736   AioGetxattrData *cdata = new AioGetxattrData(buf, completion, len);
4737   if (!cdata) {
4738     tracepoint(librados, rados_aio_getxattr_exit, -ENOMEM, NULL, 0);
4739     return -ENOMEM;
4740   }
4741   cdata->bl.push_back(buffer::create_static(len, buf));
4742   // create completion callback
4743   librados::AioCompletionImpl *c = new librados::AioCompletionImpl;
4744   c->set_complete_callback(cdata, rados_aio_getxattr_complete);
4745   // call async getxattr of IoCtx
4746   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4747   object_t oid(o);
4748   int ret = ctx->aio_getxattr(oid, c, name, cdata->bl);
4749   tracepoint(librados, rados_aio_getxattr_exit, ret, buf, ret);
4750   return ret;
4751 }
4752
4753 namespace {
4754 struct AioGetxattrsData {
4755   AioGetxattrsData(rados_completion_t c, rados_xattrs_iter_t *_iter) :
4756     iter(_iter), user_completion((librados::AioCompletionImpl*)c) {
4757     it = new librados::RadosXattrsIter();
4758   }
4759   ~AioGetxattrsData() {
4760     if (it) delete it;
4761   }
4762   librados::RadosXattrsIter *it;
4763   rados_xattrs_iter_t *iter;
4764   struct librados::C_AioCompleteAndSafe user_completion;
4765 };
4766 }
4767
4768 static void rados_aio_getxattrs_complete(rados_completion_t c, void *arg) {
4769   AioGetxattrsData *cdata = reinterpret_cast<AioGetxattrsData*>(arg);
4770   int rc = rados_aio_get_return_value(c);
4771   if (rc) {
4772     cdata->user_completion.finish(rc);
4773   } else {
4774     cdata->it->i = cdata->it->attrset.begin();
4775     *cdata->iter = cdata->it;
4776     cdata->it = 0;
4777     cdata->user_completion.finish(0);
4778   }
4779   delete cdata;
4780 }
4781
4782 extern "C" int rados_aio_getxattrs(rados_ioctx_t io, const char *oid,
4783                                    rados_completion_t completion,
4784                                    rados_xattrs_iter_t *iter)
4785 {
4786   tracepoint(librados, rados_aio_getxattrs_enter, io, oid, completion);
4787   // create data object to be passed to async callback
4788   AioGetxattrsData *cdata = new AioGetxattrsData(completion, iter);
4789   if (!cdata) {
4790     tracepoint(librados, rados_getxattrs_exit, -ENOMEM, NULL);
4791     return -ENOMEM;
4792   }
4793   // create completion callback
4794   librados::AioCompletionImpl *c = new librados::AioCompletionImpl;
4795   c->set_complete_callback(cdata, rados_aio_getxattrs_complete);
4796   // call async getxattrs of IoCtx
4797   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4798   object_t obj(oid);
4799   int ret = ctx->aio_getxattrs(obj, c, cdata->it->attrset);
4800   tracepoint(librados, rados_aio_getxattrs_exit, ret, cdata->it);
4801   return ret;
4802 }
4803
4804 extern "C" int rados_aio_setxattr(rados_ioctx_t io, const char *o,
4805                                   rados_completion_t completion,
4806                                   const char *name, const char *buf, size_t len)
4807 {
4808   tracepoint(librados, rados_aio_setxattr_enter, io, o, completion, name, buf, len);
4809   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4810   object_t oid(o);
4811   bufferlist bl;
4812   bl.append(buf, len);
4813   int retval = ctx->aio_setxattr(oid, (librados::AioCompletionImpl*)completion, name, bl);
4814   tracepoint(librados, rados_aio_setxattr_exit, retval);
4815   return retval;
4816 }
4817
4818 extern "C" int rados_aio_rmxattr(rados_ioctx_t io, const char *o,
4819                                  rados_completion_t completion,
4820                                  const char *name)
4821 {
4822   tracepoint(librados, rados_aio_rmxattr_enter, io, o, completion, name);
4823   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4824   object_t oid(o);
4825   int retval = ctx->aio_rmxattr(oid, (librados::AioCompletionImpl*)completion, name);
4826   tracepoint(librados, rados_aio_rmxattr_exit, retval);
4827   return retval;
4828 }
4829
4830 extern "C" int rados_aio_stat(rados_ioctx_t io, const char *o,
4831                               rados_completion_t completion,
4832                               uint64_t *psize, time_t *pmtime)
4833 {
4834   tracepoint(librados, rados_aio_stat_enter, io, o, completion);
4835   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4836   object_t oid(o);
4837   int retval = ctx->aio_stat(oid, (librados::AioCompletionImpl*)completion,
4838                        psize, pmtime);
4839   tracepoint(librados, rados_aio_stat_exit, retval);
4840   return retval;
4841 }
4842
4843 extern "C" int rados_aio_cmpext(rados_ioctx_t io, const char *o,
4844                                 rados_completion_t completion, const char *cmp_buf,
4845                                 size_t cmp_len, uint64_t off)
4846 {
4847   tracepoint(librados, rados_aio_cmpext_enter, io, o, completion, cmp_buf,
4848              cmp_len, off);
4849   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4850   object_t oid(o);
4851   int retval = ctx->aio_cmpext(oid, (librados::AioCompletionImpl*)completion,
4852                                cmp_buf, cmp_len, off);
4853   tracepoint(librados, rados_aio_cmpext_exit, retval);
4854   return retval;
4855 }
4856
4857 extern "C" int rados_aio_cancel(rados_ioctx_t io, rados_completion_t completion)
4858 {
4859   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4860   return ctx->aio_cancel((librados::AioCompletionImpl*)completion);
4861 }
4862
4863 extern "C" int rados_aio_exec(rados_ioctx_t io, const char *o,
4864                               rados_completion_t completion,
4865                               const char *cls, const char *method,
4866                               const char *inbuf, size_t in_len,
4867                               char *buf, size_t out_len)
4868 {
4869   tracepoint(librados, rados_aio_exec_enter, io, o, completion);
4870   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4871   object_t oid(o);
4872   bufferlist inbl;
4873   inbl.append(inbuf, in_len);
4874   int retval = ctx->aio_exec(oid, (librados::AioCompletionImpl*)completion,
4875                        cls, method, inbl, buf, out_len);
4876   tracepoint(librados, rados_aio_exec_exit, retval);
4877   return retval;
4878 }
4879
4880 struct C_WatchCB : public librados::WatchCtx {
4881   rados_watchcb_t wcb;
4882   void *arg;
4883   C_WatchCB(rados_watchcb_t _wcb, void *_arg) : wcb(_wcb), arg(_arg) {}
4884   void notify(uint8_t opcode, uint64_t ver, bufferlist& bl) override {
4885     wcb(opcode, ver, arg);
4886   }
4887 };
4888
4889 extern "C" int rados_watch(rados_ioctx_t io, const char *o, uint64_t ver,
4890                            uint64_t *handle,
4891                            rados_watchcb_t watchcb, void *arg)
4892 {
4893   tracepoint(librados, rados_watch_enter, io, o, ver, watchcb, arg);
4894   uint64_t *cookie = handle;
4895   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4896   object_t oid(o);
4897   C_WatchCB *wc = new C_WatchCB(watchcb, arg);
4898   int retval = ctx->watch(oid, cookie, wc, NULL, true);
4899   tracepoint(librados, rados_watch_exit, retval, *handle);
4900   return retval;
4901 }
4902
4903 struct C_WatchCB2 : public librados::WatchCtx2 {
4904   rados_watchcb2_t wcb;
4905   rados_watcherrcb_t errcb;
4906   void *arg;
4907   C_WatchCB2(rados_watchcb2_t _wcb,
4908              rados_watcherrcb_t _errcb,
4909              void *_arg) : wcb(_wcb), errcb(_errcb), arg(_arg) {}
4910   void handle_notify(uint64_t notify_id,
4911                      uint64_t cookie,
4912                      uint64_t notifier_gid,
4913                      bufferlist& bl) override {
4914     wcb(arg, notify_id, cookie, notifier_gid, bl.c_str(), bl.length());
4915   }
4916   void handle_error(uint64_t cookie, int err) override {
4917     if (errcb)
4918       errcb(arg, cookie, err);
4919   }
4920 };
4921
4922 extern "C" int rados_watch2(rados_ioctx_t io, const char *o, uint64_t *handle,
4923                             rados_watchcb2_t watchcb,
4924                             rados_watcherrcb_t watcherrcb,
4925                             void *arg) {
4926   return rados_watch3(io, o, handle, watchcb, watcherrcb, 0, arg);
4927 }
4928
4929 extern "C" int rados_watch3(rados_ioctx_t io, const char *o, uint64_t *handle,
4930                             rados_watchcb2_t watchcb,
4931                             rados_watcherrcb_t watcherrcb,
4932                             uint32_t timeout,
4933                             void *arg)
4934 {
4935   tracepoint(librados, rados_watch3_enter, io, o, handle, watchcb, timeout, arg);
4936   int ret;
4937   if (!watchcb || !o || !handle) {
4938     ret = -EINVAL;
4939   } else {
4940     uint64_t *cookie = handle;
4941     librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4942     object_t oid(o);
4943     C_WatchCB2 *wc = new C_WatchCB2(watchcb, watcherrcb, arg);
4944     ret = ctx->watch(oid, cookie, NULL, wc, timeout, true);
4945   }
4946   tracepoint(librados, rados_watch3_exit, ret, handle ? *handle : 0);
4947   return ret;
4948 }
4949
4950 extern "C" int rados_aio_watch(rados_ioctx_t io, const char *o,
4951                                rados_completion_t completion,
4952                                uint64_t *handle,
4953                                rados_watchcb2_t watchcb,
4954                                rados_watcherrcb_t watcherrcb, void *arg) {
4955   return rados_aio_watch2(io, o, completion, handle, watchcb, watcherrcb, 0, arg);
4956 }
4957
4958 extern "C" int rados_aio_watch2(rados_ioctx_t io, const char *o,
4959                                 rados_completion_t completion,
4960                                 uint64_t *handle,
4961                                 rados_watchcb2_t watchcb,
4962                                 rados_watcherrcb_t watcherrcb,
4963                                 uint32_t timeout, void *arg)
4964 {
4965   tracepoint(librados, rados_aio_watch2_enter, io, o, completion, handle, watchcb, timeout, arg);
4966   int ret;
4967   if (!completion || !watchcb || !o || !handle) {
4968     ret = -EINVAL;
4969   } else {
4970     uint64_t *cookie = handle;
4971     librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4972     object_t oid(o);
4973     librados::AioCompletionImpl *c =
4974       reinterpret_cast<librados::AioCompletionImpl*>(completion);
4975     C_WatchCB2 *wc = new C_WatchCB2(watchcb, watcherrcb, arg);
4976     ret = ctx->aio_watch(oid, c, cookie, NULL, wc, timeout, true);
4977   }
4978   tracepoint(librados, rados_aio_watch2_exit, ret, handle ? *handle : 0);
4979   return ret;
4980 }
4981
4982
4983 extern "C" int rados_unwatch(rados_ioctx_t io, const char *o, uint64_t handle)
4984 {
4985   tracepoint(librados, rados_unwatch_enter, io, o, handle);
4986   uint64_t cookie = handle;
4987   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4988   int retval = ctx->unwatch(cookie);
4989   tracepoint(librados, rados_unwatch_exit, retval);
4990   return retval;
4991 }
4992
4993 extern "C" int rados_unwatch2(rados_ioctx_t io, uint64_t handle)
4994 {
4995   tracepoint(librados, rados_unwatch2_enter, io, handle);
4996   uint64_t cookie = handle;
4997   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4998   int retval = ctx->unwatch(cookie);
4999   tracepoint(librados, rados_unwatch2_exit, retval);
5000   return retval;
5001 }
5002
5003 extern "C" int rados_aio_unwatch(rados_ioctx_t io, uint64_t handle,
5004                                  rados_completion_t completion)
5005 {
5006   tracepoint(librados, rados_aio_unwatch_enter, io, handle, completion);
5007   uint64_t cookie = handle;
5008   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5009   librados::AioCompletionImpl *c =
5010     reinterpret_cast<librados::AioCompletionImpl*>(completion);
5011   int retval = ctx->aio_unwatch(cookie, c);
5012   tracepoint(librados, rados_aio_unwatch_exit, retval);
5013   return retval;
5014 }
5015
5016 extern "C" int rados_watch_check(rados_ioctx_t io, uint64_t handle)
5017 {
5018   tracepoint(librados, rados_watch_check_enter, io, handle);
5019   uint64_t cookie = handle;
5020   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5021   int retval = ctx->watch_check(cookie);
5022   tracepoint(librados, rados_watch_check_exit, retval);
5023   return retval;
5024 }
5025
5026 extern "C" int rados_notify(rados_ioctx_t io, const char *o,
5027                             uint64_t ver, const char *buf, int buf_len)
5028 {
5029   tracepoint(librados, rados_notify_enter, io, o, ver, buf, buf_len);
5030   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5031   object_t oid(o);
5032   bufferlist bl;
5033   if (buf) {
5034     bufferptr p = buffer::create(buf_len);
5035     memcpy(p.c_str(), buf, buf_len);
5036     bl.push_back(p);
5037   }
5038   int retval = ctx->notify(oid, bl, 0, NULL, NULL, NULL);
5039   tracepoint(librados, rados_notify_exit, retval);
5040   return retval;
5041 }
5042
5043 extern "C" int rados_notify2(rados_ioctx_t io, const char *o,
5044                              const char *buf, int buf_len,
5045                              uint64_t timeout_ms,
5046                              char **reply_buffer,
5047                              size_t *reply_buffer_len)
5048 {
5049   tracepoint(librados, rados_notify2_enter, io, o, buf, buf_len, timeout_ms);
5050   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5051   object_t oid(o);
5052   bufferlist bl;
5053   if (buf) {
5054     bufferptr p = buffer::create(buf_len);
5055     memcpy(p.c_str(), buf, buf_len);
5056     bl.push_back(p);
5057   }
5058   int ret = ctx->notify(oid, bl, timeout_ms, NULL, reply_buffer, reply_buffer_len);
5059   tracepoint(librados, rados_notify2_exit, ret);
5060   return ret;
5061 }
5062
5063 extern "C" int rados_aio_notify(rados_ioctx_t io, const char *o,
5064                                 rados_completion_t completion,
5065                                 const char *buf, int buf_len,
5066                                 uint64_t timeout_ms, char **reply_buffer,
5067                                 size_t *reply_buffer_len)
5068 {
5069   tracepoint(librados, rados_aio_notify_enter, io, o, completion, buf, buf_len,
5070              timeout_ms);
5071   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5072   object_t oid(o);
5073   bufferlist bl;
5074   if (buf) {
5075     bl.push_back(buffer::copy(buf, buf_len));
5076   }
5077   librados::AioCompletionImpl *c =
5078     reinterpret_cast<librados::AioCompletionImpl*>(completion);
5079   int ret = ctx->aio_notify(oid, c, bl, timeout_ms, NULL, reply_buffer,
5080                             reply_buffer_len);
5081   tracepoint(librados, rados_aio_notify_exit, ret);
5082   return ret;
5083 }
5084
5085 extern "C" int rados_notify_ack(rados_ioctx_t io, const char *o,
5086                                 uint64_t notify_id, uint64_t handle,
5087                                 const char *buf, int buf_len)
5088 {
5089   tracepoint(librados, rados_notify_ack_enter, io, o, notify_id, handle, buf, buf_len);
5090   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5091   object_t oid(o);
5092   bufferlist bl;
5093   if (buf) {
5094     bufferptr p = buffer::create(buf_len);
5095     memcpy(p.c_str(), buf, buf_len);
5096     bl.push_back(p);
5097   }
5098   ctx->notify_ack(oid, notify_id, handle, bl);
5099   tracepoint(librados, rados_notify_ack_exit, 0);
5100   return 0;
5101 }
5102
5103 extern "C" int rados_watch_flush(rados_t cluster)
5104 {
5105   tracepoint(librados, rados_watch_flush_enter, cluster);
5106   librados::RadosClient *client = (librados::RadosClient *)cluster;
5107   int retval = client->watch_flush();
5108   tracepoint(librados, rados_watch_flush_exit, retval);
5109   return retval;
5110 }
5111
5112 extern "C" int rados_aio_watch_flush(rados_t cluster, rados_completion_t completion)
5113 {
5114   tracepoint(librados, rados_aio_watch_flush_enter, cluster, completion);
5115   librados::RadosClient *client = (librados::RadosClient *)cluster;
5116   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
5117   int retval = client->async_watch_flush(c);
5118   tracepoint(librados, rados_aio_watch_flush_exit, retval);
5119   return retval;
5120 }
5121
5122 extern "C" int rados_set_alloc_hint(rados_ioctx_t io, const char *o,
5123                                     uint64_t expected_object_size,
5124                                     uint64_t expected_write_size)
5125 {
5126   tracepoint(librados, rados_set_alloc_hint_enter, io, o, expected_object_size, expected_write_size);
5127   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5128   object_t oid(o);
5129   int retval = ctx->set_alloc_hint(oid, expected_object_size,
5130                                    expected_write_size, 0);
5131   tracepoint(librados, rados_set_alloc_hint_exit, retval);
5132   return retval;
5133 }
5134
5135 extern "C" int rados_set_alloc_hint2(rados_ioctx_t io, const char *o,
5136                                      uint64_t expected_object_size,
5137                                      uint64_t expected_write_size,
5138                                      uint32_t flags)
5139 {
5140   tracepoint(librados, rados_set_alloc_hint2_enter, io, o, expected_object_size, expected_write_size, flags);
5141   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5142   object_t oid(o);
5143   int retval = ctx->set_alloc_hint(oid, expected_object_size,
5144                                    expected_write_size, flags);
5145   tracepoint(librados, rados_set_alloc_hint2_exit, retval);
5146   return retval;
5147 }
5148
5149 extern "C" int rados_lock_exclusive(rados_ioctx_t io, const char * o,
5150                           const char * name, const char * cookie,
5151                           const char * desc, struct timeval * duration,
5152                           uint8_t flags)
5153 {
5154   tracepoint(librados, rados_lock_exclusive_enter, io, o, name, cookie, desc, duration, flags);
5155   librados::IoCtx ctx;
5156   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5157
5158   int retval = ctx.lock_exclusive(o, name, cookie, desc, duration, flags);
5159   tracepoint(librados, rados_lock_exclusive_exit, retval);
5160   return retval;
5161 }
5162
5163 extern "C" int rados_lock_shared(rados_ioctx_t io, const char * o,
5164                           const char * name, const char * cookie,
5165                           const char * tag, const char * desc,
5166                           struct timeval * duration, uint8_t flags)
5167 {
5168   tracepoint(librados, rados_lock_shared_enter, io, o, name, cookie, tag, desc, duration, flags);
5169   librados::IoCtx ctx;
5170   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5171
5172   int retval = ctx.lock_shared(o, name, cookie, tag, desc, duration, flags);
5173   tracepoint(librados, rados_lock_shared_exit, retval);
5174   return retval;
5175 }
5176 extern "C" int rados_unlock(rados_ioctx_t io, const char *o, const char *name,
5177                             const char *cookie)
5178 {
5179   tracepoint(librados, rados_unlock_enter, io, o, name, cookie);
5180   librados::IoCtx ctx;
5181   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5182
5183   int retval = ctx.unlock(o, name, cookie);
5184   tracepoint(librados, rados_unlock_exit, retval);
5185   return retval;
5186 }
5187
5188 extern "C" int rados_aio_unlock(rados_ioctx_t io, const char *o, const char *name,
5189                                 const char *cookie, rados_completion_t completion)
5190 {
5191   tracepoint(librados, rados_aio_unlock_enter, io, o, name, cookie, completion);
5192   librados::IoCtx ctx;
5193   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5194   librados::AioCompletionImpl *comp = (librados::AioCompletionImpl*)completion;
5195   librados::AioCompletion c(comp);
5196   int retval = ctx.aio_unlock(o, name, cookie, &c);
5197   tracepoint(librados, rados_aio_unlock_exit, retval);
5198   return retval;
5199 }
5200
5201 extern "C" ssize_t rados_list_lockers(rados_ioctx_t io, const char *o,
5202                                       const char *name, int *exclusive,
5203                                       char *tag, size_t *tag_len,
5204                                       char *clients, size_t *clients_len,
5205                                       char *cookies, size_t *cookies_len,
5206                                       char *addrs, size_t *addrs_len)
5207 {
5208   tracepoint(librados, rados_list_lockers_enter, io, o, name, *tag_len, *clients_len, *cookies_len, *addrs_len);
5209   librados::IoCtx ctx;
5210   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5211   std::string name_str = name;
5212   std::string oid = o;
5213   std::string tag_str;
5214   int tmp_exclusive;
5215   std::list<librados::locker_t> lockers;
5216   int r = ctx.list_lockers(oid, name_str, &tmp_exclusive, &tag_str, &lockers);
5217   if (r < 0) {
5218     tracepoint(librados, rados_list_lockers_exit, r, *exclusive, "", *tag_len, *clients_len, *cookies_len, *addrs_len);
5219           return r;
5220   }
5221
5222   size_t clients_total = 0;
5223   size_t cookies_total = 0;
5224   size_t addrs_total = 0;
5225   list<librados::locker_t>::const_iterator it;
5226   for (it = lockers.begin(); it != lockers.end(); ++it) {
5227     clients_total += it->client.length() + 1;
5228     cookies_total += it->cookie.length() + 1;
5229     addrs_total += it->address.length() + 1;
5230   }
5231
5232   bool too_short = ((clients_total > *clients_len) ||
5233                     (cookies_total > *cookies_len) ||
5234                     (addrs_total > *addrs_len) ||
5235                     (tag_str.length() + 1 > *tag_len));
5236   *clients_len = clients_total;
5237   *cookies_len = cookies_total;
5238   *addrs_len = addrs_total;
5239   *tag_len = tag_str.length() + 1;
5240   if (too_short) {
5241     tracepoint(librados, rados_list_lockers_exit, -ERANGE, *exclusive, "", *tag_len, *clients_len, *cookies_len, *addrs_len);
5242     return -ERANGE;
5243   }
5244
5245   strcpy(tag, tag_str.c_str());
5246   char *clients_p = clients;
5247   char *cookies_p = cookies;
5248   char *addrs_p = addrs;
5249   for (it = lockers.begin(); it != lockers.end(); ++it) {
5250     strcpy(clients_p, it->client.c_str());
5251     strcpy(cookies_p, it->cookie.c_str());
5252     strcpy(addrs_p, it->address.c_str());
5253     tracepoint(librados, rados_list_lockers_locker, clients_p, cookies_p, addrs_p);
5254     clients_p += it->client.length() + 1;
5255     cookies_p += it->cookie.length() + 1;
5256     addrs_p += it->address.length() + 1;
5257   }
5258   if (tmp_exclusive)
5259     *exclusive = 1;
5260   else
5261     *exclusive = 0;
5262
5263   int retval = lockers.size();
5264   tracepoint(librados, rados_list_lockers_exit, retval, *exclusive, tag, *tag_len, *clients_len, *cookies_len, *addrs_len);
5265   return retval;
5266 }
5267
5268 extern "C" int rados_break_lock(rados_ioctx_t io, const char *o,
5269                                 const char *name, const char *client,
5270                                 const char *cookie)
5271 {
5272   tracepoint(librados, rados_break_lock_enter, io, o, name, client, cookie);
5273   librados::IoCtx ctx;
5274   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5275
5276   int retval = ctx.break_lock(o, name, client, cookie);
5277   tracepoint(librados, rados_break_lock_exit, retval);
5278   return retval;
5279 }
5280
5281 extern "C" rados_write_op_t rados_create_write_op()
5282 {
5283   tracepoint(librados, rados_create_write_op_enter);
5284   rados_write_op_t retval = new (std::nothrow)::ObjectOperation;
5285   tracepoint(librados, rados_create_write_op_exit, retval);
5286   return retval;
5287 }
5288
5289 extern "C" void rados_release_write_op(rados_write_op_t write_op)
5290 {
5291   tracepoint(librados, rados_release_write_op_enter, write_op);
5292   delete (::ObjectOperation*)write_op;
5293   tracepoint(librados, rados_release_write_op_exit);
5294 }
5295
5296 extern "C" void rados_write_op_set_flags(rados_write_op_t write_op, int flags)
5297 {
5298   tracepoint(librados, rados_write_op_set_flags_enter, write_op, flags);
5299   set_op_flags((::ObjectOperation *)write_op, flags);
5300   tracepoint(librados, rados_write_op_set_flags_exit);
5301 }
5302
5303 extern "C" void rados_write_op_assert_version(rados_write_op_t write_op, uint64_t ver)
5304 {
5305   tracepoint(librados, rados_write_op_assert_version_enter, write_op, ver);
5306   ((::ObjectOperation *)write_op)->assert_version(ver);
5307   tracepoint(librados, rados_write_op_assert_version_exit);
5308 }
5309
5310 extern "C" void rados_write_op_assert_exists(rados_write_op_t write_op)
5311 {
5312   tracepoint(librados, rados_write_op_assert_exists_enter, write_op);
5313   ((::ObjectOperation *)write_op)->stat(NULL, (ceph::real_time *)NULL, NULL);
5314   tracepoint(librados, rados_write_op_assert_exists_exit);
5315 }
5316
5317 extern "C" void rados_write_op_cmpext(rados_write_op_t write_op,
5318                                       const char *cmp_buf,
5319                                       size_t cmp_len,
5320                                       uint64_t off,
5321                                       int *prval)
5322 {
5323   tracepoint(librados, rados_write_op_cmpext_enter, write_op, cmp_buf,
5324              cmp_len, off, prval);
5325   ((::ObjectOperation *)write_op)->cmpext(off, cmp_len, cmp_buf, prval);
5326   tracepoint(librados, rados_write_op_cmpext_exit);
5327 }
5328
5329 extern "C" void rados_write_op_cmpxattr(rados_write_op_t write_op,
5330                                        const char *name,
5331                                        uint8_t comparison_operator,
5332                                        const char *value,
5333                                        size_t value_len)
5334 {
5335   tracepoint(librados, rados_write_op_cmpxattr_enter, write_op, name, comparison_operator, value, value_len);
5336   bufferlist bl;
5337   bl.append(value, value_len);
5338   ((::ObjectOperation *)write_op)->cmpxattr(name,
5339                                             comparison_operator,
5340                                             CEPH_OSD_CMPXATTR_MODE_STRING,
5341                                             bl);
5342   tracepoint(librados, rados_write_op_cmpxattr_exit);
5343 }
5344
5345 static void rados_c_omap_cmp(ObjectOperation *op,
5346                              const char *key,
5347                              uint8_t comparison_operator,
5348                              const char *val,
5349                              size_t val_len,
5350                              int *prval)
5351 {
5352   bufferlist bl;
5353   bl.append(val, val_len);
5354   std::map<std::string, pair<bufferlist, int> > assertions;
5355   assertions[key] = std::make_pair(bl, comparison_operator);
5356   op->omap_cmp(assertions, prval);
5357 }
5358
5359 extern "C" void rados_write_op_omap_cmp(rados_write_op_t write_op,
5360                                         const char *key,
5361                                         uint8_t comparison_operator,
5362                                         const char *val,
5363                                         size_t val_len,
5364                                         int *prval)
5365 {
5366   tracepoint(librados, rados_write_op_omap_cmp_enter, write_op, key, comparison_operator, val, val_len, prval);
5367   rados_c_omap_cmp((::ObjectOperation *)write_op, key, comparison_operator,
5368                    val, val_len, prval);
5369   tracepoint(librados, rados_write_op_omap_cmp_exit);
5370 }
5371
5372 extern "C" void rados_write_op_setxattr(rados_write_op_t write_op,
5373                                        const char *name,
5374                                        const char *value,
5375                                        size_t value_len)
5376 {
5377   tracepoint(librados, rados_write_op_setxattr_enter, write_op, name, value, value_len);
5378   bufferlist bl;
5379   bl.append(value, value_len);
5380   ((::ObjectOperation *)write_op)->setxattr(name, bl);
5381   tracepoint(librados, rados_write_op_setxattr_exit);
5382 }
5383
5384 extern "C" void rados_write_op_rmxattr(rados_write_op_t write_op,
5385                                        const char *name)
5386 {
5387   tracepoint(librados, rados_write_op_rmxattr_enter, write_op, name);
5388   ((::ObjectOperation *)write_op)->rmxattr(name);
5389   tracepoint(librados, rados_write_op_rmxattr_exit);
5390 }
5391
5392 extern "C" void rados_write_op_create(rados_write_op_t write_op,
5393                                       int exclusive,
5394                                       const char* category) // unused
5395 {
5396   tracepoint(librados, rados_write_op_create_enter, write_op, exclusive);
5397   ::ObjectOperation *oo = (::ObjectOperation *) write_op;
5398   oo->create(!!exclusive);
5399   tracepoint(librados, rados_write_op_create_exit);
5400 }
5401
5402 extern "C" void rados_write_op_write(rados_write_op_t write_op,
5403                                      const char *buffer,
5404                                      size_t len,
5405                                      uint64_t offset)
5406 {
5407   tracepoint(librados, rados_write_op_write_enter, write_op, buffer, len, offset);
5408   bufferlist bl;
5409   bl.append(buffer,len);
5410   ((::ObjectOperation *)write_op)->write(offset, bl);
5411   tracepoint(librados, rados_write_op_write_exit);
5412 }
5413
5414 extern "C" void rados_write_op_write_full(rados_write_op_t write_op,
5415                                           const char *buffer,
5416                                           size_t len)
5417 {
5418   tracepoint(librados, rados_write_op_write_full_enter, write_op, buffer, len);
5419   bufferlist bl;
5420   bl.append(buffer,len);
5421   ((::ObjectOperation *)write_op)->write_full(bl);
5422   tracepoint(librados, rados_write_op_write_full_exit);
5423 }
5424
5425 extern "C" void rados_write_op_writesame(rados_write_op_t write_op,
5426                                          const char *buffer,
5427                                          size_t data_len,
5428                                          size_t write_len,
5429                                          uint64_t offset)
5430 {
5431   tracepoint(librados, rados_write_op_writesame_enter, write_op, buffer, data_len, write_len, offset);
5432   bufferlist bl;
5433   bl.append(buffer, data_len);
5434   ((::ObjectOperation *)write_op)->writesame(offset, write_len, bl);
5435   tracepoint(librados, rados_write_op_writesame_exit);
5436 }
5437
5438 extern "C" void rados_write_op_append(rados_write_op_t write_op,
5439                                       const char *buffer,
5440                                       size_t len)
5441 {
5442   tracepoint(librados, rados_write_op_append_enter, write_op, buffer, len);
5443   bufferlist bl;
5444   bl.append(buffer,len);
5445   ((::ObjectOperation *)write_op)->append(bl);
5446   tracepoint(librados, rados_write_op_append_exit);
5447 }
5448
5449 extern "C" void rados_write_op_remove(rados_write_op_t write_op)
5450 {
5451   tracepoint(librados, rados_write_op_remove_enter, write_op);
5452   ((::ObjectOperation *)write_op)->remove();
5453   tracepoint(librados, rados_write_op_remove_exit);
5454 }
5455
5456 extern "C" void rados_write_op_truncate(rados_write_op_t write_op,
5457                                         uint64_t offset)
5458 {
5459   tracepoint(librados, rados_write_op_truncate_enter, write_op, offset);
5460   ((::ObjectOperation *)write_op)->truncate(offset);
5461   tracepoint(librados, rados_write_op_truncate_exit);
5462 }
5463
5464 extern "C" void rados_write_op_zero(rados_write_op_t write_op,
5465                                     uint64_t offset,
5466                                     uint64_t len)
5467 {
5468   tracepoint(librados, rados_write_op_zero_enter, write_op, offset, len);
5469   ((::ObjectOperation *)write_op)->zero(offset, len);
5470   tracepoint(librados, rados_write_op_zero_exit);
5471 }
5472
5473 extern "C" void rados_write_op_exec(rados_write_op_t write_op,
5474                                     const char *cls,
5475                                     const char *method,
5476                                     const char *in_buf,
5477                                     size_t in_len,
5478                                     int *prval)
5479 {
5480   tracepoint(librados, rados_write_op_exec_enter, write_op, cls, method, in_buf, in_len, prval);
5481   bufferlist inbl;
5482   inbl.append(in_buf, in_len);
5483   ((::ObjectOperation *)write_op)->call(cls, method, inbl, NULL, NULL, prval);
5484   tracepoint(librados, rados_write_op_exec_exit);
5485 }
5486
5487 extern "C" void rados_write_op_omap_set(rados_write_op_t write_op,
5488                                         char const* const* keys,
5489                                         char const* const* vals,
5490                                         const size_t *lens,
5491                                         size_t num)
5492 {
5493   tracepoint(librados, rados_write_op_omap_set_enter, write_op, num);
5494   std::map<std::string, bufferlist> entries;
5495   for (size_t i = 0; i < num; ++i) {
5496     tracepoint(librados, rados_write_op_omap_set_entry, keys[i], vals[i], lens[i]);
5497     bufferlist bl(lens[i]);
5498     bl.append(vals[i], lens[i]);
5499     entries[keys[i]] = bl;
5500   }
5501   ((::ObjectOperation *)write_op)->omap_set(entries);
5502   tracepoint(librados, rados_write_op_omap_set_exit);
5503 }
5504
5505 extern "C" void rados_write_op_omap_rm_keys(rados_write_op_t write_op,
5506                                             char const* const* keys,
5507                                             size_t keys_len)
5508 {
5509   tracepoint(librados, rados_write_op_omap_rm_keys_enter, write_op, keys_len);
5510   for(size_t i = 0; i < keys_len; i++) {
5511     tracepoint(librados, rados_write_op_omap_rm_keys_entry, keys[i]);
5512   }
5513   std::set<std::string> to_remove(keys, keys + keys_len);
5514   ((::ObjectOperation *)write_op)->omap_rm_keys(to_remove);
5515   tracepoint(librados, rados_write_op_omap_rm_keys_exit);
5516 }
5517
5518 extern "C" void rados_write_op_omap_clear(rados_write_op_t write_op)
5519 {
5520   tracepoint(librados, rados_write_op_omap_clear_enter, write_op);
5521   ((::ObjectOperation *)write_op)->omap_clear();
5522   tracepoint(librados, rados_write_op_omap_clear_exit);
5523 }
5524
5525 extern "C" void rados_write_op_set_alloc_hint(rados_write_op_t write_op,
5526                                             uint64_t expected_object_size,
5527                                             uint64_t expected_write_size)
5528 {
5529   tracepoint(librados, rados_write_op_set_alloc_hint_enter, write_op, expected_object_size, expected_write_size);
5530   ((::ObjectOperation *)write_op)->set_alloc_hint(expected_object_size,
5531                                                   expected_write_size, 0);
5532   tracepoint(librados, rados_write_op_set_alloc_hint_exit);
5533 }
5534
5535 extern "C" void rados_write_op_set_alloc_hint2(rados_write_op_t write_op,
5536                                                uint64_t expected_object_size,
5537                                                uint64_t expected_write_size,
5538                                                uint32_t flags)
5539 {
5540   tracepoint(librados, rados_write_op_set_alloc_hint2_enter, write_op, expected_object_size, expected_write_size, flags);
5541   ((::ObjectOperation *)write_op)->set_alloc_hint(expected_object_size,
5542                                                   expected_write_size,
5543                                                   flags);
5544   tracepoint(librados, rados_write_op_set_alloc_hint2_exit);
5545 }
5546
5547 extern "C" int rados_write_op_operate(rados_write_op_t write_op,
5548                                       rados_ioctx_t io,
5549                                       const char *oid,
5550                                       time_t *mtime,
5551                                       int flags)
5552 {
5553   tracepoint(librados, rados_write_op_operate_enter, write_op, io, oid, mtime, flags);
5554   object_t obj(oid);
5555   ::ObjectOperation *oo = (::ObjectOperation *) write_op;
5556   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5557
5558   ceph::real_time *prt = NULL;
5559   ceph::real_time rt;
5560
5561   if (mtime) {
5562     rt = ceph::real_clock::from_time_t(*mtime);
5563     prt = &rt;
5564   }
5565
5566   int retval = ctx->operate(obj, oo, prt, translate_flags(flags));
5567   tracepoint(librados, rados_write_op_operate_exit, retval);
5568   return retval;
5569 }
5570
5571 extern "C" int rados_write_op_operate2(rados_write_op_t write_op,
5572                                        rados_ioctx_t io,
5573                                        const char *oid,
5574                                        struct timespec *ts,
5575                                        int flags)
5576 {
5577   tracepoint(librados, rados_write_op_operate2_enter, write_op, io, oid, ts, flags);
5578   object_t obj(oid);
5579   ::ObjectOperation *oo = (::ObjectOperation *) write_op;
5580   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5581
5582   ceph::real_time *prt = NULL;
5583   ceph::real_time rt;
5584
5585   if (ts) {
5586     rt = ceph::real_clock::from_timespec(*ts);
5587     prt = &rt;
5588   }
5589
5590   int retval = ctx->operate(obj, oo, prt, translate_flags(flags));
5591   tracepoint(librados, rados_write_op_operate_exit, retval);
5592   return retval;
5593 }
5594
5595 extern "C" int rados_aio_write_op_operate(rados_write_op_t write_op,
5596                                           rados_ioctx_t io,
5597                                           rados_completion_t completion,
5598                                           const char *oid,
5599                                           time_t *mtime,
5600                                           int flags)
5601 {
5602   tracepoint(librados, rados_aio_write_op_operate_enter, write_op, io, completion, oid, mtime, flags);
5603   object_t obj(oid);
5604   ::ObjectOperation *oo = (::ObjectOperation *) write_op;
5605   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5606   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
5607   int retval = ctx->aio_operate(obj, oo, c, ctx->snapc, translate_flags(flags));
5608   tracepoint(librados, rados_aio_write_op_operate_exit, retval);
5609   return retval;
5610 }
5611
5612 extern "C" rados_read_op_t rados_create_read_op()
5613 {
5614   tracepoint(librados, rados_create_read_op_enter);
5615   rados_read_op_t retval = new (std::nothrow)::ObjectOperation;
5616   tracepoint(librados, rados_create_read_op_exit, retval);
5617   return retval;
5618 }
5619
5620 extern "C" void rados_release_read_op(rados_read_op_t read_op)
5621 {
5622   tracepoint(librados, rados_release_read_op_enter, read_op);
5623   delete (::ObjectOperation *)read_op;
5624   tracepoint(librados, rados_release_read_op_exit);
5625 }
5626
5627 extern "C" void rados_read_op_set_flags(rados_read_op_t read_op, int flags)
5628 {
5629   tracepoint(librados, rados_read_op_set_flags_enter, read_op, flags);
5630   set_op_flags((::ObjectOperation *)read_op, flags);
5631   tracepoint(librados, rados_read_op_set_flags_exit);
5632 }
5633
5634 extern "C" void rados_read_op_assert_version(rados_read_op_t read_op, uint64_t ver)
5635 {
5636   tracepoint(librados, rados_read_op_assert_version_enter, read_op, ver);
5637   ((::ObjectOperation *)read_op)->assert_version(ver);
5638   tracepoint(librados, rados_read_op_assert_version_exit);
5639 }
5640
5641 extern "C" void rados_read_op_assert_exists(rados_read_op_t read_op)
5642 {
5643   tracepoint(librados, rados_read_op_assert_exists_enter, read_op);
5644   ((::ObjectOperation *)read_op)->stat(NULL, (ceph::real_time *)NULL, NULL);
5645   tracepoint(librados, rados_read_op_assert_exists_exit);
5646 }
5647
5648 extern "C" void rados_read_op_cmpext(rados_read_op_t read_op,
5649                                      const char *cmp_buf,
5650                                      size_t cmp_len,
5651                                      uint64_t off,
5652                                      int *prval)
5653 {
5654   tracepoint(librados, rados_read_op_cmpext_enter, read_op, cmp_buf,
5655              cmp_len, off, prval);
5656   ((::ObjectOperation *)read_op)->cmpext(off, cmp_len, cmp_buf, prval);
5657   tracepoint(librados, rados_read_op_cmpext_exit);
5658 }
5659
5660 extern "C" void rados_read_op_cmpxattr(rados_read_op_t read_op,
5661                                        const char *name,
5662                                        uint8_t comparison_operator,
5663                                        const char *value,
5664                                        size_t value_len)
5665 {
5666   tracepoint(librados, rados_read_op_cmpxattr_enter, read_op, name, comparison_operator, value, value_len);
5667   bufferlist bl;
5668   bl.append(value, value_len);
5669   ((::ObjectOperation *)read_op)->cmpxattr(name,
5670                                            comparison_operator,
5671                                            CEPH_OSD_CMPXATTR_MODE_STRING,
5672                                            bl);
5673   tracepoint(librados, rados_read_op_cmpxattr_exit);
5674 }
5675
5676 extern "C" void rados_read_op_omap_cmp(rados_read_op_t read_op,
5677                                        const char *key,
5678                                        uint8_t comparison_operator,
5679                                        const char *val,
5680                                        size_t val_len,
5681                                        int *prval)
5682 {
5683   tracepoint(librados, rados_read_op_omap_cmp_enter, read_op, key, comparison_operator, val, val_len, prval);
5684   rados_c_omap_cmp((::ObjectOperation *)read_op, key, comparison_operator,
5685                    val, val_len, prval);
5686   tracepoint(librados, rados_read_op_omap_cmp_exit);
5687 }
5688
5689 extern "C" void rados_read_op_stat(rados_read_op_t read_op,
5690                                    uint64_t *psize,
5691                                    time_t *pmtime,
5692                                    int *prval)
5693 {
5694   tracepoint(librados, rados_read_op_stat_enter, read_op, psize, pmtime, prval);
5695   ((::ObjectOperation *)read_op)->stat(psize, pmtime, prval);
5696   tracepoint(librados, rados_read_op_stat_exit);
5697 }
5698
5699 class C_bl_to_buf : public Context {
5700   char *out_buf;
5701   size_t out_len;
5702   size_t *bytes_read;
5703   int *prval;
5704 public:
5705   bufferlist out_bl;
5706   C_bl_to_buf(char *out_buf,
5707               size_t out_len,
5708               size_t *bytes_read,
5709               int *prval) : out_buf(out_buf), out_len(out_len),
5710                             bytes_read(bytes_read), prval(prval) {}
5711   void finish(int r) override {
5712     if (out_bl.length() > out_len) {
5713       if (prval)
5714         *prval = -ERANGE;
5715       if (bytes_read)
5716         *bytes_read = 0;
5717       return;
5718     }
5719     if (bytes_read)
5720       *bytes_read = out_bl.length();
5721     if (out_buf && !out_bl.is_provided_buffer(out_buf))
5722       out_bl.copy(0, out_bl.length(), out_buf);
5723   }
5724 };
5725
5726 extern "C" void rados_read_op_read(rados_read_op_t read_op,
5727                                    uint64_t offset,
5728                                    size_t len,
5729                                    char *buf,
5730                                    size_t *bytes_read,
5731                                    int *prval)
5732 {
5733   tracepoint(librados, rados_read_op_read_enter, read_op, offset, len, buf, bytes_read, prval);
5734   C_bl_to_buf *ctx = new C_bl_to_buf(buf, len, bytes_read, prval);
5735   ctx->out_bl.push_back(buffer::create_static(len, buf));
5736   ((::ObjectOperation *)read_op)->read(offset, len, &ctx->out_bl, prval, ctx);
5737   tracepoint(librados, rados_read_op_read_exit);
5738 }
5739
5740 extern "C" void rados_read_op_checksum(rados_read_op_t read_op,
5741                                        rados_checksum_type_t type,
5742                                        const char *init_value,
5743                                        size_t init_value_len,
5744                                        uint64_t offset, size_t len,
5745                                        size_t chunk_size, char *pchecksum,
5746                                        size_t checksum_len, int *prval)
5747 {
5748   tracepoint(librados, rados_read_op_checksum_enter, read_op, type, init_value,
5749              init_value_len, offset, len, chunk_size);
5750   bufferlist init_value_bl;
5751   init_value_bl.append(init_value, init_value_len);
5752
5753   C_bl_to_buf *ctx = nullptr;
5754   if (pchecksum != nullptr) {
5755     ctx = new C_bl_to_buf(pchecksum, checksum_len, nullptr, prval);
5756   }
5757   ((::ObjectOperation *)read_op)->checksum(get_checksum_op_type(type),
5758                                            init_value_bl, offset, len,
5759                                            chunk_size,
5760                                            (ctx ? &ctx->out_bl : nullptr),
5761                                            prval, ctx);
5762   tracepoint(librados, rados_read_op_checksum_exit);
5763 }
5764
5765 class C_out_buffer : public Context {
5766   char **out_buf;
5767   size_t *out_len;
5768 public:
5769   bufferlist out_bl;
5770   C_out_buffer(char **out_buf, size_t *out_len) : out_buf(out_buf),
5771                                                   out_len(out_len) {}
5772   void finish(int r) override {
5773     // ignore r since we don't know the meaning of return values
5774     // from custom class methods
5775     do_out_buffer(out_bl, out_buf, out_len);
5776   }
5777 };
5778
5779 extern "C" void rados_read_op_exec(rados_read_op_t read_op,
5780                                    const char *cls,
5781                                    const char *method,
5782                                    const char *in_buf,
5783                                    size_t in_len,
5784                                    char **out_buf,
5785                                    size_t *out_len,
5786                                    int *prval)
5787 {
5788   tracepoint(librados, rados_read_op_exec_enter, read_op, cls, method, in_buf, in_len, out_buf, out_len, prval);
5789   bufferlist inbl;
5790   inbl.append(in_buf, in_len);
5791   C_out_buffer *ctx = new C_out_buffer(out_buf, out_len);
5792   ((::ObjectOperation *)read_op)->call(cls, method, inbl, &ctx->out_bl, ctx,
5793                                        prval);
5794   tracepoint(librados, rados_read_op_exec_exit);
5795 }
5796
5797 extern "C" void rados_read_op_exec_user_buf(rados_read_op_t read_op,
5798                                             const char *cls,
5799                                             const char *method,
5800                                             const char *in_buf,
5801                                             size_t in_len,
5802                                             char *out_buf,
5803                                             size_t out_len,
5804                                             size_t *used_len,
5805                                             int *prval)
5806 {
5807   tracepoint(librados, rados_read_op_exec_user_buf_enter, read_op, cls, method, in_buf, in_len, out_buf, out_len, used_len, prval);
5808   C_bl_to_buf *ctx = new C_bl_to_buf(out_buf, out_len, used_len, prval);
5809   bufferlist inbl;
5810   inbl.append(in_buf, in_len);
5811   ((::ObjectOperation *)read_op)->call(cls, method, inbl, &ctx->out_bl, ctx,
5812                                        prval);
5813   tracepoint(librados, rados_read_op_exec_user_buf_exit);
5814 }
5815
5816 struct RadosOmapIter {
5817   std::map<std::string, bufferlist> values;
5818   std::map<std::string, bufferlist>::iterator i;
5819 };
5820
5821 class C_OmapIter : public Context {
5822   RadosOmapIter *iter;
5823 public:
5824   explicit C_OmapIter(RadosOmapIter *iter) : iter(iter) {}
5825   void finish(int r) override {
5826     iter->i = iter->values.begin();
5827   }
5828 };
5829
5830 class C_XattrsIter : public Context {
5831   librados::RadosXattrsIter *iter;
5832 public:
5833   explicit C_XattrsIter(librados::RadosXattrsIter *iter) : iter(iter) {}
5834   void finish(int r) override {
5835     iter->i = iter->attrset.begin();
5836   }
5837 };
5838
5839 extern "C" void rados_read_op_getxattrs(rados_read_op_t read_op,
5840                                         rados_xattrs_iter_t *iter,
5841                                         int *prval)
5842 {
5843   tracepoint(librados, rados_read_op_getxattrs_enter, read_op, prval);
5844   librados::RadosXattrsIter *xattrs_iter = new librados::RadosXattrsIter;
5845   ((::ObjectOperation *)read_op)->getxattrs(&xattrs_iter->attrset, prval);
5846   ((::ObjectOperation *)read_op)->add_handler(new C_XattrsIter(xattrs_iter));
5847   *iter = xattrs_iter;
5848   tracepoint(librados, rados_read_op_getxattrs_exit, *iter);
5849 }
5850
5851 extern "C" void rados_read_op_omap_get_vals(rados_read_op_t read_op,
5852                                             const char *start_after,
5853                                             const char *filter_prefix,
5854                                             uint64_t max_return,
5855                                             rados_omap_iter_t *iter,
5856                                             int *prval)
5857 {
5858   tracepoint(librados, rados_read_op_omap_get_vals_enter, read_op, start_after, filter_prefix, max_return, prval);
5859   RadosOmapIter *omap_iter = new RadosOmapIter;
5860   const char *start = start_after ? start_after : "";
5861   const char *filter = filter_prefix ? filter_prefix : "";
5862   ((::ObjectOperation *)read_op)->omap_get_vals(
5863     start,
5864     filter,
5865     max_return,
5866     &omap_iter->values,
5867     nullptr,
5868     prval);
5869   ((::ObjectOperation *)read_op)->add_handler(new C_OmapIter(omap_iter));
5870   *iter = omap_iter;
5871   tracepoint(librados, rados_read_op_omap_get_vals_exit, *iter);
5872 }
5873
5874 extern "C" void rados_read_op_omap_get_vals2(rados_read_op_t read_op,
5875                                              const char *start_after,
5876                                              const char *filter_prefix,
5877                                              uint64_t max_return,
5878                                              rados_omap_iter_t *iter,
5879                                              unsigned char *pmore,
5880                                              int *prval)
5881 {
5882   tracepoint(librados, rados_read_op_omap_get_vals_enter, read_op, start_after, filter_prefix, max_return, prval);
5883   RadosOmapIter *omap_iter = new RadosOmapIter;
5884   const char *start = start_after ? start_after : "";
5885   const char *filter = filter_prefix ? filter_prefix : "";
5886   ((::ObjectOperation *)read_op)->omap_get_vals(
5887     start,
5888     filter,
5889     max_return,
5890     &omap_iter->values,
5891     (bool*)pmore,
5892     prval);
5893   ((::ObjectOperation *)read_op)->add_handler(new C_OmapIter(omap_iter));
5894   *iter = omap_iter;
5895   tracepoint(librados, rados_read_op_omap_get_vals_exit, *iter);
5896 }
5897
5898 struct C_OmapKeysIter : public Context {
5899   RadosOmapIter *iter;
5900   std::set<std::string> keys;
5901   explicit C_OmapKeysIter(RadosOmapIter *iter) : iter(iter) {}
5902   void finish(int r) override {
5903     // map each key to an empty bl
5904     for (std::set<std::string>::const_iterator i = keys.begin();
5905          i != keys.end(); ++i) {
5906       iter->values[*i];
5907     }
5908     iter->i = iter->values.begin();
5909   }
5910 };
5911
5912 extern "C" void rados_read_op_omap_get_keys(rados_read_op_t read_op,
5913                                             const char *start_after,
5914                                             uint64_t max_return,
5915                                             rados_omap_iter_t *iter,
5916                                             int *prval)
5917 {
5918   tracepoint(librados, rados_read_op_omap_get_keys_enter, read_op, start_after, max_return, prval);
5919   RadosOmapIter *omap_iter = new RadosOmapIter;
5920   C_OmapKeysIter *ctx = new C_OmapKeysIter(omap_iter);
5921   ((::ObjectOperation *)read_op)->omap_get_keys(
5922     start_after ? start_after : "",
5923     max_return, &ctx->keys, nullptr, prval);
5924   ((::ObjectOperation *)read_op)->add_handler(ctx);
5925   *iter = omap_iter;
5926   tracepoint(librados, rados_read_op_omap_get_keys_exit, *iter);
5927 }
5928
5929 extern "C" void rados_read_op_omap_get_keys2(rados_read_op_t read_op,
5930                                              const char *start_after,
5931                                              uint64_t max_return,
5932                                              rados_omap_iter_t *iter,
5933                                              unsigned char *pmore,
5934                                              int *prval)
5935 {
5936   tracepoint(librados, rados_read_op_omap_get_keys_enter, read_op, start_after, max_return, prval);
5937   RadosOmapIter *omap_iter = new RadosOmapIter;
5938   C_OmapKeysIter *ctx = new C_OmapKeysIter(omap_iter);
5939   ((::ObjectOperation *)read_op)->omap_get_keys(
5940     start_after ? start_after : "",
5941     max_return, &ctx->keys,
5942     (bool*)pmore, prval);
5943   ((::ObjectOperation *)read_op)->add_handler(ctx);
5944   *iter = omap_iter;
5945   tracepoint(librados, rados_read_op_omap_get_keys_exit, *iter);
5946 }
5947
5948 extern "C" void rados_read_op_omap_get_vals_by_keys(rados_read_op_t read_op,
5949                                                     char const* const* keys,
5950                                                     size_t keys_len,
5951                                                     rados_omap_iter_t *iter,
5952                                                     int *prval)
5953 {
5954   tracepoint(librados, rados_read_op_omap_get_vals_by_keys_enter, read_op, keys, keys_len, iter, prval);
5955   std::set<std::string> to_get(keys, keys + keys_len);
5956
5957   RadosOmapIter *omap_iter = new RadosOmapIter;
5958   ((::ObjectOperation *)read_op)->omap_get_vals_by_keys(to_get,
5959                                                         &omap_iter->values,
5960                                                         prval);
5961   ((::ObjectOperation *)read_op)->add_handler(new C_OmapIter(omap_iter));
5962   *iter = omap_iter;
5963   tracepoint(librados, rados_read_op_omap_get_vals_by_keys_exit, *iter);
5964 }
5965
5966 extern "C" int rados_omap_get_next(rados_omap_iter_t iter,
5967                                    char **key,
5968                                    char **val,
5969                                    size_t *len)
5970 {
5971   tracepoint(librados, rados_omap_get_next_enter, iter);
5972   RadosOmapIter *it = static_cast<RadosOmapIter *>(iter);
5973   if (it->i == it->values.end()) {
5974     *key = NULL;
5975     *val = NULL;
5976     *len = 0;
5977     tracepoint(librados, rados_omap_get_next_exit, 0, key, val, len);
5978     return 0;
5979   }
5980   if (key)
5981     *key = (char*)it->i->first.c_str();
5982   if (val)
5983     *val = it->i->second.c_str();
5984   if (len)
5985     *len = it->i->second.length();
5986   ++it->i;
5987   tracepoint(librados, rados_omap_get_next_exit, 0, key, val, len);
5988   return 0;
5989 }
5990
5991 extern "C" void rados_omap_get_end(rados_omap_iter_t iter)
5992 {
5993   tracepoint(librados, rados_omap_get_end_enter, iter);
5994   RadosOmapIter *it = static_cast<RadosOmapIter *>(iter);
5995   delete it;
5996   tracepoint(librados, rados_omap_get_end_exit);
5997 }
5998
5999 extern "C" int rados_read_op_operate(rados_read_op_t read_op,
6000                                      rados_ioctx_t io,
6001                                      const char *oid,
6002                                      int flags)
6003 {
6004   tracepoint(librados, rados_read_op_operate_enter, read_op, io, oid, flags);
6005   object_t obj(oid);
6006   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6007   int retval = ctx->operate_read(obj, (::ObjectOperation *)read_op, NULL,
6008                                  translate_flags(flags));
6009   tracepoint(librados, rados_read_op_operate_exit, retval);
6010   return retval;
6011 }
6012
6013 extern "C" int rados_aio_read_op_operate(rados_read_op_t read_op,
6014                                          rados_ioctx_t io,
6015                                          rados_completion_t completion,
6016                                          const char *oid,
6017                                          int flags)
6018 {
6019   tracepoint(librados, rados_aio_read_op_operate_enter, read_op, io, completion, oid, flags);
6020   object_t obj(oid);
6021   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6022   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
6023   int retval = ctx->aio_operate_read(obj, (::ObjectOperation *)read_op,
6024                                      c, translate_flags(flags), NULL);
6025   tracepoint(librados, rados_aio_read_op_operate_exit, retval);
6026   return retval;
6027 }
6028
6029 extern "C" int rados_cache_pin(rados_ioctx_t io, const char *o)
6030 {
6031   tracepoint(librados, rados_cache_pin_enter, io, o);
6032   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6033   object_t oid(o);
6034   int retval = ctx->cache_pin(oid);
6035   tracepoint(librados, rados_cache_pin_exit, retval);
6036   return retval;
6037 }
6038
6039 extern "C" int rados_cache_unpin(rados_ioctx_t io, const char *o)
6040 {
6041   tracepoint(librados, rados_cache_unpin_enter, io, o);
6042   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6043   object_t oid(o);
6044   int retval = ctx->cache_unpin(oid);
6045   tracepoint(librados, rados_cache_unpin_exit, retval);
6046   return retval;
6047 }
6048
6049
6050 ///////////////////////////// ListObject //////////////////////////////
6051 librados::ListObject::ListObject() : impl(NULL)
6052 {
6053 }
6054
6055 librados::ListObject::ListObject(librados::ListObjectImpl *i): impl(i)
6056 {
6057 }
6058
6059 librados::ListObject::ListObject(const ListObject& rhs)
6060 {
6061   if (rhs.impl == NULL) {
6062     impl = NULL;
6063     return;
6064   }
6065   impl = new ListObjectImpl();
6066   *impl = *(rhs.impl);
6067 }
6068
6069 librados::ListObject& librados::ListObject::operator=(const ListObject& rhs)
6070 {
6071   if (rhs.impl == NULL) {
6072     delete impl;
6073     impl = NULL;
6074     return *this;
6075   }
6076   if (impl == NULL)
6077     impl = new ListObjectImpl();
6078   *impl = *(rhs.impl);
6079   return *this;
6080 }
6081
6082 librados::ListObject::~ListObject()
6083 {
6084   if (impl)
6085     delete impl;
6086   impl = NULL;
6087 }
6088
6089 const std::string& librados::ListObject::get_nspace() const
6090 {
6091   return impl->get_nspace();
6092 }
6093
6094 const std::string& librados::ListObject::get_oid() const
6095 {
6096   return impl->get_oid();
6097 }
6098
6099 const std::string& librados::ListObject::get_locator() const
6100 {
6101   return impl->get_locator();
6102 }
6103
6104 std::ostream& librados::operator<<(std::ostream& out, const librados::ListObject& lop)
6105 {
6106   out << *(lop.impl);
6107   return out;
6108 }
6109
6110 CEPH_RADOS_API void rados_object_list_slice(
6111     rados_ioctx_t io,
6112     const rados_object_list_cursor start,
6113     const rados_object_list_cursor finish,
6114     const size_t n,
6115     const size_t m,
6116     rados_object_list_cursor *split_start,
6117     rados_object_list_cursor *split_finish)
6118 {
6119   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6120
6121   assert(split_start);
6122   assert(split_finish);
6123   hobject_t *split_start_hobj = (hobject_t*)(*split_start);
6124   hobject_t *split_finish_hobj = (hobject_t*)(*split_finish);
6125   assert(split_start_hobj);
6126   assert(split_finish_hobj);
6127   hobject_t *start_hobj = (hobject_t*)(start);
6128   hobject_t *finish_hobj = (hobject_t*)(finish);
6129
6130   ctx->object_list_slice(
6131       *start_hobj,
6132       *finish_hobj,
6133       n,
6134       m,
6135       split_start_hobj,
6136       split_finish_hobj);
6137 }
6138
6139 librados::ObjectCursor::ObjectCursor()
6140 {
6141   c_cursor = (rados_object_list_cursor)new hobject_t();
6142 }
6143
6144 librados::ObjectCursor::~ObjectCursor()
6145 {
6146   hobject_t *h = (hobject_t *)c_cursor;
6147   delete h;
6148 }
6149
6150 librados::ObjectCursor::ObjectCursor(rados_object_list_cursor c)
6151 {
6152   if (!c) {
6153     c_cursor = nullptr;
6154   } else {
6155     c_cursor = (rados_object_list_cursor)new hobject_t(*(hobject_t *)c);
6156   }
6157 }
6158
6159 librados::ObjectCursor& librados::ObjectCursor::operator=(const librados::ObjectCursor& rhs)
6160 {
6161   if (rhs.c_cursor != nullptr) {
6162     hobject_t *h = (hobject_t*)rhs.c_cursor;
6163     c_cursor = (rados_object_list_cursor)(new hobject_t(*h));
6164   } else {
6165     c_cursor = nullptr;
6166   }
6167   return *this;
6168 }
6169
6170 bool librados::ObjectCursor::operator<(const librados::ObjectCursor &rhs) const
6171 {
6172   const hobject_t lhs_hobj = (c_cursor == nullptr) ? hobject_t() : *((hobject_t*)c_cursor);
6173   const hobject_t rhs_hobj = (rhs.c_cursor == nullptr) ? hobject_t() : *((hobject_t*)(rhs.c_cursor));
6174   return lhs_hobj < rhs_hobj;
6175 }
6176
6177 bool librados::ObjectCursor::operator==(const librados::ObjectCursor &rhs) const
6178 {
6179   const hobject_t lhs_hobj = (c_cursor == nullptr) ? hobject_t() : *((hobject_t*)c_cursor);
6180   const hobject_t rhs_hobj = (rhs.c_cursor == nullptr) ? hobject_t() : *((hobject_t*)(rhs.c_cursor));
6181   return cmp(lhs_hobj, rhs_hobj) == 0;
6182 }
6183 librados::ObjectCursor::ObjectCursor(const librados::ObjectCursor &rhs)
6184 {
6185   *this = rhs;
6186 }
6187
6188 librados::ObjectCursor librados::IoCtx::object_list_begin()
6189 {
6190   hobject_t *h = new hobject_t(io_ctx_impl->objecter->enumerate_objects_begin());
6191   ObjectCursor oc;
6192   oc.set((rados_object_list_cursor)h);
6193   return oc;
6194 }
6195
6196
6197 librados::ObjectCursor librados::IoCtx::object_list_end()
6198 {
6199   hobject_t *h = new hobject_t(io_ctx_impl->objecter->enumerate_objects_end());
6200   librados::ObjectCursor oc;
6201   oc.set((rados_object_list_cursor)h);
6202   return oc;
6203 }
6204
6205
6206 void librados::ObjectCursor::set(rados_object_list_cursor c)
6207 {
6208   delete (hobject_t*)c_cursor;
6209   c_cursor = c;
6210 }
6211
6212 string librados::ObjectCursor::to_str() const
6213 {
6214   stringstream ss;
6215   ss << *(hobject_t *)c_cursor;
6216   return ss.str();
6217 }
6218
6219 bool librados::ObjectCursor::from_str(const string& s)
6220 {
6221   if (s.empty()) {
6222     *(hobject_t *)c_cursor = hobject_t();
6223     return true;
6224   }
6225   return ((hobject_t *)c_cursor)->parse(s);
6226 }
6227
6228 CEPH_RADOS_API std::ostream& librados::operator<<(std::ostream& os, const librados::ObjectCursor& oc)
6229 {
6230   if (oc.c_cursor) {
6231     os << *(hobject_t *)oc.c_cursor;
6232   } else {
6233     os << hobject_t();
6234   }
6235   return os;
6236 }
6237
6238 bool librados::IoCtx::object_list_is_end(const ObjectCursor &oc)
6239 {
6240   hobject_t *h = (hobject_t *)oc.c_cursor;
6241   return h->is_max();
6242 }
6243
6244 int librados::IoCtx::object_list(const ObjectCursor &start,
6245                 const ObjectCursor &finish,
6246                 const size_t result_item_count,
6247                 const bufferlist &filter,
6248                 std::vector<ObjectItem> *result,
6249                 ObjectCursor *next)
6250 {
6251   assert(result != nullptr);
6252   assert(next != nullptr);
6253   result->clear();
6254
6255   C_SaferCond cond;
6256   hobject_t next_hash;
6257   std::list<librados::ListObjectImpl> obj_result;
6258   io_ctx_impl->objecter->enumerate_objects(
6259       io_ctx_impl->poolid,
6260       io_ctx_impl->oloc.nspace,
6261       *((hobject_t*)start.c_cursor),
6262       *((hobject_t*)finish.c_cursor),
6263       result_item_count,
6264       filter,
6265       &obj_result,
6266       &next_hash,
6267       &cond);
6268
6269   int r = cond.wait();
6270   if (r < 0) {
6271     next->set((rados_object_list_cursor)(new hobject_t(hobject_t::get_max())));
6272     return r;
6273   }
6274
6275   next->set((rados_object_list_cursor)(new hobject_t(next_hash)));
6276
6277   for (std::list<librados::ListObjectImpl>::iterator i = obj_result.begin();
6278        i != obj_result.end(); ++i) {
6279     ObjectItem oi;
6280     oi.oid = i->oid;
6281     oi.nspace = i->nspace;
6282     oi.locator = i->locator;
6283     result->push_back(oi);
6284   }
6285
6286   return obj_result.size();
6287 }
6288
6289 void librados::IoCtx::object_list_slice(
6290     const ObjectCursor start,
6291     const ObjectCursor finish,
6292     const size_t n,
6293     const size_t m,
6294     ObjectCursor *split_start,
6295     ObjectCursor *split_finish)
6296 {
6297   assert(split_start != nullptr);
6298   assert(split_finish != nullptr);
6299
6300   io_ctx_impl->object_list_slice(
6301       *((hobject_t*)(start.c_cursor)),
6302       *((hobject_t*)(finish.c_cursor)),
6303       n,
6304       m,
6305       (hobject_t*)(split_start->c_cursor),
6306       (hobject_t*)(split_finish->c_cursor));
6307 }
6308