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