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.mon_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_pool_list(rados_t cluster, char *buf, size_t len)
3091 {
3092   tracepoint(librados, rados_pool_list_enter, cluster, len);
3093   librados::RadosClient *client = (librados::RadosClient *)cluster;
3094   std::list<std::pair<int64_t, std::string> > pools;
3095   int r = client->pool_list(pools);
3096   if (r < 0) {
3097     tracepoint(librados, rados_pool_list_exit, r);
3098     return r;
3099   }
3100
3101   if (len > 0 && !buf) {
3102     tracepoint(librados, rados_pool_list_exit, -EINVAL);
3103     return -EINVAL;
3104   }
3105
3106   char *b = buf;
3107   if (b)
3108     memset(b, 0, len);
3109   int needed = 0;
3110   std::list<std::pair<int64_t, std::string> >::const_iterator i = pools.begin();
3111   std::list<std::pair<int64_t, std::string> >::const_iterator p_end =
3112     pools.end();
3113   for (; i != p_end; ++i) {
3114     int rl = i->second.length() + 1;
3115     if (len < (unsigned)rl)
3116       break;
3117     const char* pool = i->second.c_str();
3118     tracepoint(librados, rados_pool_list_pool, pool);
3119     if (b) {
3120       strncat(b, pool, rl);
3121       b += rl;
3122     }
3123     needed += rl;
3124     len -= rl;
3125   }
3126   for (; i != p_end; ++i) {
3127     int rl = i->second.length() + 1;
3128     needed += rl;
3129   }
3130   int retval = needed + 1;
3131   tracepoint(librados, rados_pool_list_exit, retval);
3132   return retval;
3133 }
3134
3135 CEPH_RADOS_API int rados_inconsistent_pg_list(rados_t cluster, int64_t pool_id,
3136                                               char *buf, size_t len)
3137 {
3138   tracepoint(librados, rados_inconsistent_pg_list_enter, cluster, pool_id, len);
3139   librados::RadosClient *client = (librados::RadosClient *)cluster;
3140   std::vector<librados::PlacementGroup> pgs;
3141   int r = ::get_inconsistent_pgs(*client, pool_id, &pgs);
3142   if (r < 0) {
3143     tracepoint(librados, rados_inconsistent_pg_list_exit, r);
3144     return r;
3145   }
3146
3147   if (len > 0 && !buf) {
3148     tracepoint(librados, rados_inconsistent_pg_list_exit, -EINVAL);
3149     return -EINVAL;
3150   }
3151
3152   char *b = buf;
3153   if (b)
3154     memset(b, 0, len);
3155   int needed = 0;
3156   for (const auto pg : pgs) {
3157     std::ostringstream ss;
3158     ss << pg;
3159     auto s = ss.str();
3160     unsigned rl = s.length() + 1;
3161     if (b && len >= rl) {
3162       tracepoint(librados, rados_inconsistent_pg_list_pg, s.c_str());
3163       strncat(b, s.c_str(), rl);
3164       b += rl;
3165       len -= rl;
3166     }
3167     needed += rl;
3168   }
3169   int retval = needed + 1;
3170   tracepoint(librados, rados_inconsistent_pg_list_exit, retval);
3171   return retval;
3172 }
3173
3174 static void do_out_buffer(bufferlist& outbl, char **outbuf, size_t *outbuflen)
3175 {
3176   if (outbuf) {
3177     if (outbl.length() > 0) {
3178       *outbuf = (char *)malloc(outbl.length());
3179       memcpy(*outbuf, outbl.c_str(), outbl.length());
3180     } else {
3181       *outbuf = NULL;
3182     }
3183   }
3184   if (outbuflen)
3185     *outbuflen = outbl.length();
3186 }
3187
3188 static void do_out_buffer(string& outbl, char **outbuf, size_t *outbuflen)
3189 {
3190   if (outbuf) {
3191     if (outbl.length() > 0) {
3192       *outbuf = (char *)malloc(outbl.length());
3193       memcpy(*outbuf, outbl.c_str(), outbl.length());
3194     } else {
3195       *outbuf = NULL;
3196     }
3197   }
3198   if (outbuflen)
3199     *outbuflen = outbl.length();
3200 }
3201
3202 extern "C" int rados_ping_monitor(rados_t cluster, const char *mon_id,
3203                                   char **outstr, size_t *outstrlen)
3204 {
3205   tracepoint(librados, rados_ping_monitor_enter, cluster, mon_id);
3206   librados::RadosClient *client = (librados::RadosClient *)cluster;
3207   string str;
3208
3209   if (!mon_id) {
3210     tracepoint(librados, rados_ping_monitor_exit, -EINVAL, NULL, NULL);
3211     return -EINVAL;
3212   }
3213
3214   int ret = client->ping_monitor(mon_id, &str);
3215   if (ret == 0) {
3216     do_out_buffer(str, outstr, outstrlen);
3217   }
3218   tracepoint(librados, rados_ping_monitor_exit, ret, ret < 0 ? NULL : outstr, ret < 0 ? NULL : outstrlen);
3219   return ret;
3220 }
3221
3222 extern "C" int rados_mon_command(rados_t cluster, const char **cmd,
3223                                  size_t cmdlen,
3224                                  const char *inbuf, size_t inbuflen,
3225                                  char **outbuf, size_t *outbuflen,
3226                                  char **outs, size_t *outslen)
3227 {
3228   tracepoint(librados, rados_mon_command_enter, cluster, cmdlen, inbuf, inbuflen);
3229   librados::RadosClient *client = (librados::RadosClient *)cluster;
3230   bufferlist inbl;
3231   bufferlist outbl;
3232   string outstring;
3233   vector<string> cmdvec;
3234
3235   for (size_t i = 0; i < cmdlen; i++) {
3236     tracepoint(librados, rados_mon_command_cmd, cmd[i]);
3237     cmdvec.push_back(cmd[i]);
3238   }
3239
3240   inbl.append(inbuf, inbuflen);
3241   int ret = client->mon_command(cmdvec, inbl, &outbl, &outstring);
3242
3243   do_out_buffer(outbl, outbuf, outbuflen);
3244   do_out_buffer(outstring, outs, outslen);
3245   tracepoint(librados, rados_mon_command_exit, ret, outbuf, outbuflen, outs, outslen);
3246   return ret;
3247 }
3248
3249 extern "C" int rados_mon_command_target(rados_t cluster, const char *name,
3250                                         const char **cmd,
3251                                         size_t cmdlen,
3252                                         const char *inbuf, size_t inbuflen,
3253                                         char **outbuf, size_t *outbuflen,
3254                                         char **outs, size_t *outslen)
3255 {
3256   tracepoint(librados, rados_mon_command_target_enter, cluster, name, cmdlen, inbuf, inbuflen);
3257   librados::RadosClient *client = (librados::RadosClient *)cluster;
3258   bufferlist inbl;
3259   bufferlist outbl;
3260   string outstring;
3261   vector<string> cmdvec;
3262
3263   // is this a numeric id?
3264   char *endptr;
3265   errno = 0;
3266   long rank = strtol(name, &endptr, 10);
3267   if ((errno == ERANGE && (rank == LONG_MAX || rank == LONG_MIN)) ||
3268       (errno != 0 && rank == 0) ||
3269       endptr == name ||    // no digits
3270       *endptr != '\0') {   // extra characters
3271     rank = -1;
3272   }
3273
3274   for (size_t i = 0; i < cmdlen; i++) {
3275     tracepoint(librados, rados_mon_command_target_cmd, cmd[i]);
3276     cmdvec.push_back(cmd[i]);
3277   }
3278
3279   inbl.append(inbuf, inbuflen);
3280   int ret;
3281   if (rank >= 0)
3282     ret = client->mon_command(rank, cmdvec, inbl, &outbl, &outstring);
3283   else
3284     ret = client->mon_command(name, cmdvec, inbl, &outbl, &outstring);
3285
3286   do_out_buffer(outbl, outbuf, outbuflen);
3287   do_out_buffer(outstring, outs, outslen);
3288   tracepoint(librados, rados_mon_command_target_exit, ret, outbuf, outbuflen, outs, outslen);
3289   return ret;
3290 }
3291
3292 extern "C" int rados_osd_command(rados_t cluster, int osdid, const char **cmd,
3293                                  size_t cmdlen,
3294                                  const char *inbuf, size_t inbuflen,
3295                                  char **outbuf, size_t *outbuflen,
3296                                  char **outs, size_t *outslen)
3297 {
3298   tracepoint(librados, rados_osd_command_enter, cluster, osdid, cmdlen, inbuf, inbuflen);
3299   librados::RadosClient *client = (librados::RadosClient *)cluster;
3300   bufferlist inbl;
3301   bufferlist outbl;
3302   string outstring;
3303   vector<string> cmdvec;
3304
3305   for (size_t i = 0; i < cmdlen; i++) {
3306     tracepoint(librados, rados_osd_command_cmd, cmd[i]);
3307     cmdvec.push_back(cmd[i]);
3308   }
3309
3310   inbl.append(inbuf, inbuflen);
3311   int ret = client->osd_command(osdid, cmdvec, inbl, &outbl, &outstring);
3312
3313   do_out_buffer(outbl, outbuf, outbuflen);
3314   do_out_buffer(outstring, outs, outslen);
3315   tracepoint(librados, rados_osd_command_exit, ret, outbuf, outbuflen, outs, outslen);
3316   return ret;
3317 }
3318
3319 extern "C" int rados_mgr_command(rados_t cluster, const char **cmd,
3320                                  size_t cmdlen,
3321                                  const char *inbuf, size_t inbuflen,
3322                                  char **outbuf, size_t *outbuflen,
3323                                  char **outs, size_t *outslen)
3324 {
3325   tracepoint(librados, rados_mgr_command_enter, cluster, cmdlen, inbuf,
3326       inbuflen);
3327
3328   librados::RadosClient *client = (librados::RadosClient *)cluster;
3329   bufferlist inbl;
3330   bufferlist outbl;
3331   string outstring;
3332   vector<string> cmdvec;
3333
3334   for (size_t i = 0; i < cmdlen; i++) {
3335     tracepoint(librados, rados_mgr_command_cmd, cmd[i]);
3336     cmdvec.push_back(cmd[i]);
3337   }
3338
3339   inbl.append(inbuf, inbuflen);
3340   int ret = client->mgr_command(cmdvec, inbl, &outbl, &outstring);
3341
3342   do_out_buffer(outbl, outbuf, outbuflen);
3343   do_out_buffer(outstring, outs, outslen);
3344   tracepoint(librados, rados_mgr_command_exit, ret, outbuf, outbuflen, outs,
3345       outslen);
3346   return ret;
3347 }
3348
3349 extern "C" int rados_pg_command(rados_t cluster, const char *pgstr,
3350                                 const char **cmd, size_t cmdlen,
3351                                 const char *inbuf, size_t inbuflen,
3352                                 char **outbuf, size_t *outbuflen,
3353                                 char **outs, size_t *outslen)
3354 {
3355   tracepoint(librados, rados_pg_command_enter, cluster, pgstr, cmdlen, inbuf, inbuflen);
3356   librados::RadosClient *client = (librados::RadosClient *)cluster;
3357   bufferlist inbl;
3358   bufferlist outbl;
3359   string outstring;
3360   pg_t pgid;
3361   vector<string> cmdvec;
3362
3363   for (size_t i = 0; i < cmdlen; i++) {
3364     tracepoint(librados, rados_pg_command_cmd, cmd[i]);
3365     cmdvec.push_back(cmd[i]);
3366   }
3367
3368   inbl.append(inbuf, inbuflen);
3369   if (!pgid.parse(pgstr))
3370     return -EINVAL;
3371
3372   int ret = client->pg_command(pgid, cmdvec, inbl, &outbl, &outstring);
3373
3374   do_out_buffer(outbl, outbuf, outbuflen);
3375   do_out_buffer(outstring, outs, outslen);
3376   tracepoint(librados, rados_pg_command_exit, ret, outbuf, outbuflen, outs, outslen);
3377   return ret;
3378 }
3379
3380 extern "C" void rados_buffer_free(char *buf)
3381 {
3382   tracepoint(librados, rados_buffer_free_enter, buf);
3383   if (buf)
3384     free(buf);
3385   tracepoint(librados, rados_buffer_free_exit);
3386 }
3387
3388 extern "C" int rados_monitor_log(rados_t cluster, const char *level, rados_log_callback_t cb, void *arg)
3389 {
3390   tracepoint(librados, rados_monitor_log_enter, cluster, level, cb, arg);
3391   librados::RadosClient *client = (librados::RadosClient *)cluster;
3392   int retval = client->monitor_log(level, cb, nullptr, arg);
3393   tracepoint(librados, rados_monitor_log_exit, retval);
3394   return retval;
3395 }
3396
3397 extern "C" int rados_monitor_log2(rados_t cluster, const char *level,
3398                                   rados_log_callback2_t cb, void *arg)
3399 {
3400   tracepoint(librados, rados_monitor_log2_enter, cluster, level, cb, arg);
3401   librados::RadosClient *client = (librados::RadosClient *)cluster;
3402   int retval = client->monitor_log(level, nullptr, cb, arg);
3403   tracepoint(librados, rados_monitor_log2_exit, retval);
3404   return retval;
3405 }
3406
3407 extern "C" int rados_ioctx_create(rados_t cluster, const char *name, rados_ioctx_t *io)
3408 {
3409   tracepoint(librados, rados_ioctx_create_enter, cluster, name);
3410   librados::RadosClient *client = (librados::RadosClient *)cluster;
3411   librados::IoCtxImpl *ctx;
3412
3413   int r = client->create_ioctx(name, &ctx);
3414   if (r < 0) {
3415     tracepoint(librados, rados_ioctx_create_exit, r, NULL);
3416     return r;
3417   }
3418
3419   *io = ctx;
3420   ctx->get();
3421   tracepoint(librados, rados_ioctx_create_exit, 0, ctx);
3422   return 0;
3423 }
3424
3425 extern "C" int rados_ioctx_create2(rados_t cluster, int64_t pool_id,
3426                                    rados_ioctx_t *io)
3427 {
3428   tracepoint(librados, rados_ioctx_create2_enter, cluster, pool_id);
3429   librados::RadosClient *client = (librados::RadosClient *)cluster;
3430   librados::IoCtxImpl *ctx;
3431
3432   int r = client->create_ioctx(pool_id, &ctx);
3433   if (r < 0) {
3434     tracepoint(librados, rados_ioctx_create2_exit, r, NULL);
3435     return r;
3436   }
3437
3438   *io = ctx;
3439   ctx->get();
3440   tracepoint(librados, rados_ioctx_create2_exit, 0, ctx);
3441   return 0;
3442 }
3443
3444 extern "C" void rados_ioctx_destroy(rados_ioctx_t io)
3445 {
3446   tracepoint(librados, rados_ioctx_destroy_enter, io);
3447   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3448   ctx->put();
3449   tracepoint(librados, rados_ioctx_destroy_exit);
3450 }
3451
3452 extern "C" int rados_ioctx_pool_stat(rados_ioctx_t io, struct rados_pool_stat_t *stats)
3453 {
3454   tracepoint(librados, rados_ioctx_pool_stat_enter, io);
3455   librados::IoCtxImpl *io_ctx_impl = (librados::IoCtxImpl *)io;
3456   list<string> ls;
3457   std::string pool_name;
3458
3459   int err = io_ctx_impl->client->pool_get_name(io_ctx_impl->get_id(), &pool_name);
3460   if (err) {
3461     tracepoint(librados, rados_ioctx_pool_stat_exit, err, stats);
3462     return err;
3463   }
3464   ls.push_back(pool_name);
3465
3466   map<string, ::pool_stat_t> rawresult;
3467   err = io_ctx_impl->client->get_pool_stats(ls, rawresult);
3468   if (err) {
3469     tracepoint(librados, rados_ioctx_pool_stat_exit, err, stats);
3470     return err;
3471   }
3472
3473   ::pool_stat_t& r = rawresult[pool_name];
3474   stats->num_kb = SHIFT_ROUND_UP(r.stats.sum.num_bytes, 10);
3475   stats->num_bytes = r.stats.sum.num_bytes;
3476   stats->num_objects = r.stats.sum.num_objects;
3477   stats->num_object_clones = r.stats.sum.num_object_clones;
3478   stats->num_object_copies = r.stats.sum.num_object_copies;
3479   stats->num_objects_missing_on_primary = r.stats.sum.num_objects_missing_on_primary;
3480   stats->num_objects_unfound = r.stats.sum.num_objects_unfound;
3481   stats->num_objects_degraded =
3482     r.stats.sum.num_objects_degraded +
3483     r.stats.sum.num_objects_misplaced; // FIXME: this is imprecise
3484   stats->num_rd = r.stats.sum.num_rd;
3485   stats->num_rd_kb = r.stats.sum.num_rd_kb;
3486   stats->num_wr = r.stats.sum.num_wr;
3487   stats->num_wr_kb = r.stats.sum.num_wr_kb;
3488   tracepoint(librados, rados_ioctx_pool_stat_exit, 0, stats);
3489   return 0;
3490 }
3491
3492 extern "C" rados_config_t rados_ioctx_cct(rados_ioctx_t io)
3493 {
3494   tracepoint(librados, rados_ioctx_cct_enter, io);
3495   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3496   rados_config_t retval = (rados_config_t)ctx->client->cct;
3497   tracepoint(librados, rados_ioctx_cct_exit, retval);
3498   return retval;
3499 }
3500
3501 extern "C" void rados_ioctx_snap_set_read(rados_ioctx_t io, rados_snap_t seq)
3502 {
3503   tracepoint(librados, rados_ioctx_snap_set_read_enter, io, seq);
3504   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3505   ctx->set_snap_read((snapid_t)seq);
3506   tracepoint(librados, rados_ioctx_snap_set_read_exit);
3507 }
3508
3509 extern "C" int rados_ioctx_selfmanaged_snap_set_write_ctx(rados_ioctx_t io,
3510             rados_snap_t seq, rados_snap_t *snaps, int num_snaps)
3511 {
3512   tracepoint(librados, rados_ioctx_selfmanaged_snap_set_write_ctx_enter, io, seq, snaps, num_snaps);
3513   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3514   vector<snapid_t> snv;
3515   snv.resize(num_snaps);
3516   for (int i=0; i<num_snaps; i++) {
3517     snv[i] = (snapid_t)snaps[i];
3518   }
3519   int retval = ctx->set_snap_write_context((snapid_t)seq, snv);
3520   tracepoint(librados, rados_ioctx_selfmanaged_snap_set_write_ctx_exit, retval);
3521   return retval;
3522 }
3523
3524 extern "C" int rados_write(rados_ioctx_t io, const char *o, const char *buf, size_t len, uint64_t off)
3525 {
3526   tracepoint(librados, rados_write_enter, io, o, buf, len, off);
3527   if (len > UINT_MAX/2)
3528     return -E2BIG;
3529   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3530   object_t oid(o);
3531   bufferlist bl;
3532   bl.append(buf, len);
3533   int retval = ctx->write(oid, bl, len, off);
3534   tracepoint(librados, rados_write_exit, retval);
3535   return retval;
3536 }
3537
3538 extern "C" int rados_append(rados_ioctx_t io, const char *o, const char *buf, size_t len)
3539 {
3540   tracepoint(librados, rados_append_enter, io, o, buf, len);
3541   if (len > UINT_MAX/2)
3542     return -E2BIG;
3543   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3544   object_t oid(o);
3545   bufferlist bl;
3546   bl.append(buf, len);
3547   int retval = ctx->append(oid, bl, len);
3548   tracepoint(librados, rados_append_exit, retval);
3549   return retval;
3550 }
3551
3552 extern "C" int rados_write_full(rados_ioctx_t io, const char *o, const char *buf, size_t len)
3553 {
3554   tracepoint(librados, rados_write_full_enter, io, o, buf, len);
3555   if (len > UINT_MAX/2)
3556     return -E2BIG;
3557   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3558   object_t oid(o);
3559   bufferlist bl;
3560   bl.append(buf, len);
3561   int retval = ctx->write_full(oid, bl);
3562   tracepoint(librados, rados_write_full_exit, retval);
3563   return retval;
3564 }
3565
3566 extern "C" int rados_writesame(rados_ioctx_t io,
3567                                 const char *o,
3568                                 const char *buf,
3569                                 size_t data_len,
3570                                 size_t write_len,
3571                                 uint64_t off)
3572 {
3573   tracepoint(librados, rados_writesame_enter, io, o, buf, data_len, write_len, off);
3574   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3575   object_t oid(o);
3576   bufferlist bl;
3577   bl.append(buf, data_len);
3578   int retval = ctx->writesame(oid, bl, write_len, off);
3579   tracepoint(librados, rados_writesame_exit, retval);
3580   return retval;
3581 }
3582
3583 extern "C" int rados_trunc(rados_ioctx_t io, const char *o, uint64_t size)
3584 {
3585   tracepoint(librados, rados_trunc_enter, io, o, size);
3586   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3587   object_t oid(o);
3588   int retval = ctx->trunc(oid, size);
3589   tracepoint(librados, rados_trunc_exit, retval);
3590   return retval;
3591 }
3592
3593 extern "C" int rados_remove(rados_ioctx_t io, const char *o)
3594 {
3595   tracepoint(librados, rados_remove_enter, io, o);
3596   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3597   object_t oid(o);
3598   int retval = ctx->remove(oid);
3599   tracepoint(librados, rados_remove_exit, retval);
3600   return retval;
3601 }
3602
3603 extern "C" int rados_read(rados_ioctx_t io, const char *o, char *buf, size_t len, uint64_t off)
3604 {
3605   tracepoint(librados, rados_read_enter, io, o, buf, len, off);
3606   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3607   int ret;
3608   object_t oid(o);
3609
3610   bufferlist bl;
3611   bufferptr bp = buffer::create_static(len, buf);
3612   bl.push_back(bp);
3613
3614   ret = ctx->read(oid, bl, len, off);
3615   if (ret >= 0) {
3616     if (bl.length() > len) {
3617       tracepoint(librados, rados_read_exit, -ERANGE, NULL);
3618       return -ERANGE;
3619     }
3620     if (!bl.is_provided_buffer(buf))
3621       bl.copy(0, bl.length(), buf);
3622     ret = bl.length();    // hrm :/
3623   }
3624
3625   tracepoint(librados, rados_read_exit, ret, buf);
3626   return ret;
3627 }
3628
3629 extern "C" int rados_checksum(rados_ioctx_t io, const char *o,
3630                               rados_checksum_type_t type,
3631                               const char *init_value, size_t init_value_len,
3632                               size_t len, uint64_t off, size_t chunk_size,
3633                               char *pchecksum, size_t checksum_len)
3634 {
3635   tracepoint(librados, rados_checksum_enter, io, o, type, init_value,
3636              init_value_len, len, off, chunk_size);
3637   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3638   object_t oid(o);
3639
3640   bufferlist init_value_bl;
3641   init_value_bl.append(init_value, init_value_len);
3642
3643   bufferlist checksum_bl;
3644
3645   int retval = ctx->checksum(oid, get_checksum_op_type(type), init_value_bl,
3646                              len, off, chunk_size, &checksum_bl);
3647   if (retval >= 0) {
3648     if (checksum_bl.length() > checksum_len) {
3649       tracepoint(librados, rados_checksum_exit, -ERANGE, NULL, 0);
3650       return -ERANGE;
3651     }
3652
3653     checksum_bl.copy(0, checksum_bl.length(), pchecksum);
3654   }
3655   tracepoint(librados, rados_checksum_exit, retval, pchecksum, checksum_len);
3656   return retval;
3657 }
3658
3659 extern "C" uint64_t rados_get_last_version(rados_ioctx_t io)
3660 {
3661   tracepoint(librados, rados_get_last_version_enter, io);
3662   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3663   uint64_t retval = ctx->last_version();
3664   tracepoint(librados, rados_get_last_version_exit, retval);
3665   return retval;
3666 }
3667
3668 extern "C" int rados_pool_create(rados_t cluster, const char *name)
3669 {
3670   tracepoint(librados, rados_pool_create_enter, cluster, name);
3671   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3672   string sname(name);
3673   int retval = radosp->pool_create(sname);
3674   tracepoint(librados, rados_pool_create_exit, retval);
3675   return retval;
3676 }
3677
3678 extern "C" int rados_pool_create_with_auid(rados_t cluster, const char *name,
3679                                            uint64_t auid)
3680 {
3681   tracepoint(librados, rados_pool_create_with_auid_enter, cluster, name, auid);
3682   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3683   string sname(name);
3684   int retval = radosp->pool_create(sname, auid);
3685   tracepoint(librados, rados_pool_create_with_auid_exit, retval);
3686   return retval;
3687 }
3688
3689 extern "C" int rados_pool_create_with_crush_rule(rados_t cluster, const char *name,
3690                                                  __u8 crush_rule_num)
3691 {
3692   tracepoint(librados, rados_pool_create_with_crush_rule_enter, cluster, name, crush_rule_num);
3693   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3694   string sname(name);
3695   int retval = radosp->pool_create(sname, 0, crush_rule_num);
3696   tracepoint(librados, rados_pool_create_with_crush_rule_exit, retval);
3697   return retval;
3698 }
3699
3700 extern "C" int rados_pool_create_with_all(rados_t cluster, const char *name,
3701                                           uint64_t auid, __u8 crush_rule_num)
3702 {
3703   tracepoint(librados, rados_pool_create_with_all_enter, cluster, name, auid, crush_rule_num);
3704   librados::RadosClient *radosp = (librados::RadosClient *)cluster;
3705   string sname(name);
3706   int retval = radosp->pool_create(sname, auid, crush_rule_num);
3707   tracepoint(librados, rados_pool_create_with_all_exit, retval);
3708   return retval;
3709 }
3710
3711 extern "C" int rados_pool_get_base_tier(rados_t cluster, int64_t pool_id, int64_t* base_tier)
3712 {
3713   tracepoint(librados, rados_pool_get_base_tier_enter, cluster, pool_id);
3714   librados::RadosClient *client = (librados::RadosClient *)cluster;
3715   int retval = client->pool_get_base_tier(pool_id, base_tier);
3716   tracepoint(librados, rados_pool_get_base_tier_exit, retval, *base_tier);
3717   return retval;
3718 }
3719
3720 extern "C" int rados_pool_delete(rados_t cluster, const char *pool_name)
3721 {
3722   tracepoint(librados, rados_pool_delete_enter, cluster, pool_name);
3723   librados::RadosClient *client = (librados::RadosClient *)cluster;
3724   int retval = client->pool_delete(pool_name);
3725   tracepoint(librados, rados_pool_delete_exit, retval);
3726   return retval;
3727 }
3728
3729 extern "C" int rados_ioctx_pool_set_auid(rados_ioctx_t io, uint64_t auid)
3730 {
3731   tracepoint(librados, rados_ioctx_pool_set_auid_enter, io, auid);
3732   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3733   int retval = ctx->pool_change_auid(auid);
3734   tracepoint(librados, rados_ioctx_pool_set_auid_exit, retval);
3735   return retval;
3736 }
3737
3738 extern "C" int rados_ioctx_pool_get_auid(rados_ioctx_t io, uint64_t *auid)
3739 {
3740   tracepoint(librados, rados_ioctx_pool_get_auid_enter, io);
3741   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3742   int retval = ctx->client->pool_get_auid(ctx->get_id(), (unsigned long long *)auid);
3743   tracepoint(librados, rados_ioctx_pool_get_auid_exit, retval, *auid);
3744   return retval;
3745 }
3746
3747 extern "C" int rados_ioctx_pool_requires_alignment(rados_ioctx_t io)
3748 {
3749   tracepoint(librados, rados_ioctx_pool_requires_alignment_enter, io);
3750   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3751   int retval = ctx->client->pool_requires_alignment(ctx->get_id());
3752   tracepoint(librados, rados_ioctx_pool_requires_alignment_exit, retval);
3753   return retval;
3754 }
3755
3756 extern "C" int rados_ioctx_pool_requires_alignment2(rados_ioctx_t io,
3757         int *requires)
3758 {
3759   tracepoint(librados, rados_ioctx_pool_requires_alignment_enter2, io);
3760   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3761   bool requires_alignment;
3762   int retval = ctx->client->pool_requires_alignment2(ctx->get_id(),
3763         &requires_alignment);
3764   tracepoint(librados, rados_ioctx_pool_requires_alignment_exit2, retval,
3765         requires_alignment);
3766   if (requires)
3767     *requires = requires_alignment;
3768   return retval;
3769 }
3770
3771 extern "C" uint64_t rados_ioctx_pool_required_alignment(rados_ioctx_t io)
3772 {
3773   tracepoint(librados, rados_ioctx_pool_required_alignment_enter, io);
3774   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3775   uint64_t retval = ctx->client->pool_required_alignment(ctx->get_id());
3776   tracepoint(librados, rados_ioctx_pool_required_alignment_exit, retval);
3777   return retval;
3778 }
3779
3780 extern "C" int rados_ioctx_pool_required_alignment2(rados_ioctx_t io,
3781         uint64_t *alignment)
3782 {
3783   tracepoint(librados, rados_ioctx_pool_required_alignment_enter2, io);
3784   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3785   int retval = ctx->client->pool_required_alignment2(ctx->get_id(),
3786         alignment);
3787   tracepoint(librados, rados_ioctx_pool_required_alignment_exit2, retval,
3788         *alignment);
3789   return retval;
3790 }
3791
3792 extern "C" void rados_ioctx_locator_set_key(rados_ioctx_t io, const char *key)
3793 {
3794   tracepoint(librados, rados_ioctx_locator_set_key_enter, io, key);
3795   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3796   if (key)
3797     ctx->oloc.key = key;
3798   else
3799     ctx->oloc.key = "";
3800   tracepoint(librados, rados_ioctx_locator_set_key_exit);
3801 }
3802
3803 extern "C" void rados_ioctx_set_namespace(rados_ioctx_t io, const char *nspace)
3804 {
3805   tracepoint(librados, rados_ioctx_set_namespace_enter, io, nspace);
3806   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3807   if (nspace)
3808     ctx->oloc.nspace = nspace;
3809   else
3810     ctx->oloc.nspace = "";
3811   tracepoint(librados, rados_ioctx_set_namespace_exit);
3812 }
3813
3814 extern "C" rados_t rados_ioctx_get_cluster(rados_ioctx_t io)
3815 {
3816   tracepoint(librados, rados_ioctx_get_cluster_enter, io);
3817   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3818   rados_t retval = (rados_t)ctx->client;
3819   tracepoint(librados, rados_ioctx_get_cluster_exit, retval);
3820   return retval;
3821 }
3822
3823 extern "C" int64_t rados_ioctx_get_id(rados_ioctx_t io)
3824 {
3825   tracepoint(librados, rados_ioctx_get_id_enter, io);
3826   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3827   int64_t retval = ctx->get_id();
3828   tracepoint(librados, rados_ioctx_get_id_exit, retval);
3829   return retval;
3830 }
3831
3832 extern "C" int rados_ioctx_get_pool_name(rados_ioctx_t io, char *s, unsigned maxlen)
3833 {
3834   tracepoint(librados, rados_ioctx_get_pool_name_enter, io, maxlen);
3835   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3836   std::string pool_name;
3837
3838   int err = ctx->client->pool_get_name(ctx->get_id(), &pool_name);
3839   if (err) {
3840     tracepoint(librados, rados_ioctx_get_pool_name_exit, err, "");
3841     return err;
3842   }
3843   if (pool_name.length() >= maxlen) {
3844     tracepoint(librados, rados_ioctx_get_pool_name_exit, -ERANGE, "");
3845     return -ERANGE;
3846   }
3847   strcpy(s, pool_name.c_str());
3848   int retval = pool_name.length();
3849   tracepoint(librados, rados_ioctx_get_pool_name_exit, retval, s);
3850   return retval;
3851 }
3852
3853 // snaps
3854
3855 extern "C" int rados_ioctx_snap_create(rados_ioctx_t io, const char *snapname)
3856 {
3857   tracepoint(librados, rados_ioctx_snap_create_enter, io, snapname);
3858   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3859   int retval = ctx->snap_create(snapname);
3860   tracepoint(librados, rados_ioctx_snap_create_exit, retval);
3861   return retval;
3862 }
3863
3864 extern "C" int rados_ioctx_snap_remove(rados_ioctx_t io, const char *snapname)
3865 {
3866   tracepoint(librados, rados_ioctx_snap_remove_enter, io, snapname);
3867   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3868   int retval = ctx->snap_remove(snapname);
3869   tracepoint(librados, rados_ioctx_snap_remove_exit, retval);
3870   return retval;
3871 }
3872
3873 extern "C" int rados_ioctx_snap_rollback(rados_ioctx_t io, const char *oid,
3874                               const char *snapname)
3875 {
3876   tracepoint(librados, rados_ioctx_snap_rollback_enter, io, oid, snapname);
3877   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3878   int retval = ctx->rollback(oid, snapname);
3879   tracepoint(librados, rados_ioctx_snap_rollback_exit, retval);
3880   return retval;
3881 }
3882
3883 // Deprecated name kept for backward compatibility
3884 extern "C" int rados_rollback(rados_ioctx_t io, const char *oid,
3885                               const char *snapname)
3886 {
3887   return rados_ioctx_snap_rollback(io, oid, snapname);
3888 }
3889
3890 extern "C" int rados_ioctx_selfmanaged_snap_create(rados_ioctx_t io,
3891                                              uint64_t *snapid)
3892 {
3893   tracepoint(librados, rados_ioctx_selfmanaged_snap_create_enter, io);
3894   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3895   int retval = ctx->selfmanaged_snap_create(snapid);
3896   tracepoint(librados, rados_ioctx_selfmanaged_snap_create_exit, retval, *snapid);
3897   return retval;
3898 }
3899
3900 extern "C" void
3901 rados_aio_ioctx_selfmanaged_snap_create(rados_ioctx_t io,
3902                                         rados_snap_t *snapid,
3903                                         rados_completion_t completion)
3904 {
3905   tracepoint(librados, rados_ioctx_selfmanaged_snap_create_enter, io);
3906   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3907   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
3908   ctx->aio_selfmanaged_snap_create(snapid, c);
3909   tracepoint(librados, rados_ioctx_selfmanaged_snap_create_exit, 0, 0);
3910 }
3911
3912 extern "C" int rados_ioctx_selfmanaged_snap_remove(rados_ioctx_t io,
3913                                              uint64_t snapid)
3914 {
3915   tracepoint(librados, rados_ioctx_selfmanaged_snap_remove_enter, io, snapid);
3916   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3917   int retval = ctx->selfmanaged_snap_remove(snapid);
3918   tracepoint(librados, rados_ioctx_selfmanaged_snap_remove_exit, retval);
3919   return retval;
3920 }
3921
3922 extern "C" void
3923 rados_aio_ioctx_selfmanaged_snap_remove(rados_ioctx_t io,
3924                                         rados_snap_t snapid,
3925                                         rados_completion_t completion)
3926 {
3927   tracepoint(librados, rados_ioctx_selfmanaged_snap_remove_enter, io, snapid);
3928   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3929   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
3930   ctx->aio_selfmanaged_snap_remove(snapid, c);
3931   tracepoint(librados, rados_ioctx_selfmanaged_snap_remove_exit, 0);
3932 }
3933
3934 extern "C" int rados_ioctx_selfmanaged_snap_rollback(rados_ioctx_t io,
3935                                                      const char *oid,
3936                                                      uint64_t snapid)
3937 {
3938   tracepoint(librados, rados_ioctx_selfmanaged_snap_rollback_enter, io, oid, snapid);
3939   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3940   int retval = ctx->selfmanaged_snap_rollback_object(oid, ctx->snapc, snapid);
3941   tracepoint(librados, rados_ioctx_selfmanaged_snap_rollback_exit, retval);
3942   return retval;
3943 }
3944
3945 extern "C" int rados_ioctx_snap_list(rados_ioctx_t io, rados_snap_t *snaps,
3946                                     int maxlen)
3947 {
3948   tracepoint(librados, rados_ioctx_snap_list_enter, io, maxlen);
3949   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3950   vector<uint64_t> snapvec;
3951   int r = ctx->snap_list(&snapvec);
3952   if (r < 0) {
3953     tracepoint(librados, rados_ioctx_snap_list_exit, r, snaps, 0);
3954     return r;
3955   }
3956   if ((int)snapvec.size() <= maxlen) {
3957     for (unsigned i=0; i<snapvec.size(); i++) {
3958       snaps[i] = snapvec[i];
3959     }
3960     int retval = snapvec.size();
3961     tracepoint(librados, rados_ioctx_snap_list_exit, retval, snaps, retval);
3962     return retval;
3963   }
3964   int retval = -ERANGE;
3965   tracepoint(librados, rados_ioctx_snap_list_exit, retval, snaps, 0);
3966   return retval;
3967 }
3968
3969 extern "C" int rados_ioctx_snap_lookup(rados_ioctx_t io, const char *name,
3970                                       rados_snap_t *id)
3971 {
3972   tracepoint(librados, rados_ioctx_snap_lookup_enter, io, name);
3973   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3974   int retval = ctx->snap_lookup(name, (uint64_t *)id);
3975   tracepoint(librados, rados_ioctx_snap_lookup_exit, retval, *id);
3976   return retval;
3977 }
3978
3979 extern "C" int rados_ioctx_snap_get_name(rados_ioctx_t io, rados_snap_t id,
3980                                         char *name, int maxlen)
3981 {
3982   tracepoint(librados, rados_ioctx_snap_get_name_enter, io, id, maxlen);
3983   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
3984   std::string sname;
3985   int r = ctx->snap_get_name(id, &sname);
3986   if (r < 0) {
3987     tracepoint(librados, rados_ioctx_snap_get_name_exit, r, "");
3988     return r;
3989   }
3990   if ((int)sname.length() >= maxlen) {
3991     int retval = -ERANGE;
3992     tracepoint(librados, rados_ioctx_snap_get_name_exit, retval, "");
3993     return retval;
3994   }
3995   strncpy(name, sname.c_str(), maxlen);
3996   tracepoint(librados, rados_ioctx_snap_get_name_exit, 0, name);
3997   return 0;
3998 }
3999
4000 extern "C" int rados_ioctx_snap_get_stamp(rados_ioctx_t io, rados_snap_t id, time_t *t)
4001 {
4002   tracepoint(librados, rados_ioctx_snap_get_stamp_enter, io, id);
4003   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4004   int retval = ctx->snap_get_stamp(id, t);
4005   tracepoint(librados, rados_ioctx_snap_get_stamp_exit, retval, *t);
4006   return retval;
4007 }
4008
4009 extern "C" int rados_cmpext(rados_ioctx_t io, const char *o,
4010                             const char *cmp_buf, size_t cmp_len, uint64_t off)
4011 {
4012   tracepoint(librados, rados_cmpext_enter, io, o, cmp_buf, cmp_len, off);
4013   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4014   int ret;
4015   object_t oid(o);
4016
4017   bufferlist cmp_bl;
4018   cmp_bl.append(cmp_buf, cmp_len);
4019
4020   ret = ctx->cmpext(oid, off, cmp_bl);
4021   tracepoint(librados, rados_cmpext_exit, ret);
4022
4023   return ret;
4024 }
4025
4026 extern "C" int rados_getxattr(rados_ioctx_t io, const char *o, const char *name,
4027                               char *buf, size_t len)
4028 {
4029   tracepoint(librados, rados_getxattr_enter, io, o, name, len);
4030   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4031   int ret;
4032   object_t oid(o);
4033   bufferlist bl;
4034   bl.push_back(buffer::create_static(len, buf));
4035   ret = ctx->getxattr(oid, name, bl);
4036   if (ret >= 0) {
4037     if (bl.length() > len) {
4038       tracepoint(librados, rados_getxattr_exit, -ERANGE, buf, 0);
4039       return -ERANGE;
4040     }
4041     if (!bl.is_provided_buffer(buf))
4042       bl.copy(0, bl.length(), buf);
4043     ret = bl.length();
4044   }
4045
4046   tracepoint(librados, rados_getxattr_exit, ret, buf, ret);
4047   return ret;
4048 }
4049
4050 extern "C" int rados_getxattrs(rados_ioctx_t io, const char *oid,
4051                                rados_xattrs_iter_t *iter)
4052 {
4053   tracepoint(librados, rados_getxattrs_enter, io, oid);
4054   librados::RadosXattrsIter *it = new librados::RadosXattrsIter();
4055   if (!it) {
4056     tracepoint(librados, rados_getxattrs_exit, -ENOMEM, NULL);
4057     return -ENOMEM;
4058   }
4059   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4060   object_t obj(oid);
4061   int ret = ctx->getxattrs(obj, it->attrset);
4062   if (ret) {
4063     delete it;
4064     tracepoint(librados, rados_getxattrs_exit, ret, NULL);
4065     return ret;
4066   }
4067   it->i = it->attrset.begin();
4068
4069   librados::RadosXattrsIter **iret = (librados::RadosXattrsIter**)iter;
4070   *iret = it;
4071   *iter = it;
4072   tracepoint(librados, rados_getxattrs_exit, 0, *iter);
4073   return 0;
4074 }
4075
4076 extern "C" int rados_getxattrs_next(rados_xattrs_iter_t iter,
4077                                     const char **name, const char **val, size_t *len)
4078 {
4079   tracepoint(librados, rados_getxattrs_next_enter, iter);
4080   librados::RadosXattrsIter *it = static_cast<librados::RadosXattrsIter*>(iter);
4081   if (it->i == it->attrset.end()) {
4082     *name = NULL;
4083     *val = NULL;
4084     *len = 0;
4085     tracepoint(librados, rados_getxattrs_next_exit, 0, NULL, NULL, 0);
4086     return 0;
4087   }
4088   free(it->val);
4089   const std::string &s(it->i->first);
4090   *name = s.c_str();
4091   bufferlist &bl(it->i->second);
4092   size_t bl_len = bl.length();
4093   if (!bl_len) {
4094     // malloc(0) is not guaranteed to return a valid pointer
4095     *val = (char *)NULL;
4096   } else {
4097     it->val = (char*)malloc(bl_len);
4098     if (!it->val) {
4099       tracepoint(librados, rados_getxattrs_next_exit, -ENOMEM, *name, NULL, 0);
4100       return -ENOMEM;
4101     }
4102     memcpy(it->val, bl.c_str(), bl_len);
4103     *val = it->val;
4104   }
4105   *len = bl_len;
4106   ++it->i;
4107   tracepoint(librados, rados_getxattrs_next_exit, 0, *name, *val, *len);
4108   return 0;
4109 }
4110
4111 extern "C" void rados_getxattrs_end(rados_xattrs_iter_t iter)
4112 {
4113   tracepoint(librados, rados_getxattrs_end_enter, iter);
4114   librados::RadosXattrsIter *it = static_cast<librados::RadosXattrsIter*>(iter);
4115   delete it;
4116   tracepoint(librados, rados_getxattrs_end_exit);
4117 }
4118
4119 extern "C" int rados_setxattr(rados_ioctx_t io, const char *o, const char *name, const char *buf, size_t len)
4120 {
4121   tracepoint(librados, rados_setxattr_enter, io, o, name, buf, len);
4122   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4123   object_t oid(o);
4124   bufferlist bl;
4125   bl.append(buf, len);
4126   int retval = ctx->setxattr(oid, name, bl);
4127   tracepoint(librados, rados_setxattr_exit, retval);
4128   return retval;
4129 }
4130
4131 extern "C" int rados_rmxattr(rados_ioctx_t io, const char *o, const char *name)
4132 {
4133   tracepoint(librados, rados_rmxattr_enter, io, o, name);
4134   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4135   object_t oid(o);
4136   int retval = ctx->rmxattr(oid, name);
4137   tracepoint(librados, rados_rmxattr_exit, retval);
4138   return retval;
4139 }
4140
4141 extern "C" int rados_stat(rados_ioctx_t io, const char *o, uint64_t *psize, time_t *pmtime)
4142 {
4143   tracepoint(librados, rados_stat_enter, io, o);
4144   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4145   object_t oid(o);
4146   int retval = ctx->stat(oid, psize, pmtime);
4147   tracepoint(librados, rados_stat_exit, retval, psize, pmtime);
4148   return retval;
4149 }
4150
4151 extern "C" int rados_tmap_update(rados_ioctx_t io, const char *o, const char *cmdbuf, size_t cmdbuflen)
4152 {
4153   tracepoint(librados, rados_tmap_update_enter, io, o, cmdbuf, cmdbuflen);
4154   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4155   object_t oid(o);
4156   bufferlist cmdbl;
4157   cmdbl.append(cmdbuf, cmdbuflen);
4158   int retval = ctx->tmap_update(oid, cmdbl);
4159   tracepoint(librados, rados_tmap_update_exit, retval);
4160   return retval;
4161 }
4162
4163 extern "C" int rados_tmap_put(rados_ioctx_t io, const char *o, const char *buf, size_t buflen)
4164 {
4165   tracepoint(librados, rados_tmap_put_enter, io, o, buf, buflen);
4166   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4167   object_t oid(o);
4168   bufferlist bl;
4169   bl.append(buf, buflen);
4170   int retval = ctx->tmap_put(oid, bl);
4171   tracepoint(librados, rados_tmap_put_exit, retval);
4172   return retval;
4173 }
4174
4175 extern "C" int rados_tmap_get(rados_ioctx_t io, const char *o, char *buf, size_t buflen)
4176 {
4177   tracepoint(librados, rados_tmap_get_enter, io, o, buflen);
4178   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4179   object_t oid(o);
4180   bufferlist bl;
4181   int r = ctx->tmap_get(oid, bl);
4182   if (r < 0) {
4183     tracepoint(librados, rados_tmap_get_exit, r, buf, 0);
4184     return r;
4185   }
4186   if (bl.length() > buflen) {
4187     tracepoint(librados, rados_tmap_get_exit, -ERANGE, buf, 0);
4188     return -ERANGE;
4189   }
4190   bl.copy(0, bl.length(), buf);
4191   int retval = bl.length();
4192   tracepoint(librados, rados_tmap_get_exit, retval, buf, retval);
4193   return retval;
4194 }
4195
4196 extern "C" int rados_tmap_to_omap(rados_ioctx_t io, const char *o, bool nullok)
4197 {
4198   tracepoint(librados, rados_tmap_to_omap_enter, io, o, nullok);
4199   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4200   object_t oid(o);
4201   int retval = ctx->tmap_to_omap(oid, nullok);
4202   tracepoint(librados, rados_tmap_to_omap_exit, retval);
4203   return retval;
4204 }
4205
4206 extern "C" int rados_exec(rados_ioctx_t io, const char *o, const char *cls, const char *method,
4207                          const char *inbuf, size_t in_len, char *buf, size_t out_len)
4208 {
4209   tracepoint(librados, rados_exec_enter, io, o, cls, method, inbuf, in_len, out_len);
4210   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4211   object_t oid(o);
4212   bufferlist inbl, outbl;
4213   int ret;
4214   inbl.append(inbuf, in_len);
4215   ret = ctx->exec(oid, cls, method, inbl, outbl);
4216   if (ret >= 0) {
4217     if (outbl.length()) {
4218       if (outbl.length() > out_len) {
4219         tracepoint(librados, rados_exec_exit, -ERANGE, buf, 0);
4220         return -ERANGE;
4221       }
4222       outbl.copy(0, outbl.length(), buf);
4223       ret = outbl.length();   // hrm :/
4224     }
4225   }
4226   tracepoint(librados, rados_exec_exit, ret, buf, ret);
4227   return ret;
4228 }
4229
4230 extern "C" rados_object_list_cursor rados_object_list_begin(rados_ioctx_t io)
4231 {
4232   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4233
4234   hobject_t *result = new hobject_t(ctx->objecter->enumerate_objects_begin());
4235   return (rados_object_list_cursor)result;
4236 }
4237
4238 extern "C" rados_object_list_cursor rados_object_list_end(rados_ioctx_t io)
4239 {
4240   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4241
4242   hobject_t *result = new hobject_t(ctx->objecter->enumerate_objects_end());
4243   return (rados_object_list_cursor)result;
4244 }
4245
4246 extern "C" int rados_object_list_is_end(
4247     rados_ioctx_t io, rados_object_list_cursor cur)
4248 {
4249   hobject_t *hobj = (hobject_t*)cur;
4250   return hobj->is_max();
4251 }
4252
4253 extern "C" void rados_object_list_cursor_free(
4254     rados_ioctx_t io, rados_object_list_cursor cur)
4255 {
4256   hobject_t *hobj = (hobject_t*)cur;
4257   delete hobj;
4258 }
4259
4260 extern "C" int rados_object_list_cursor_cmp(
4261     rados_ioctx_t io,
4262     rados_object_list_cursor lhs_cur,
4263     rados_object_list_cursor rhs_cur)
4264 {
4265   hobject_t *lhs = (hobject_t*)lhs_cur;
4266   hobject_t *rhs = (hobject_t*)rhs_cur;
4267   return cmp(*lhs, *rhs);
4268 }
4269
4270 extern "C" int rados_object_list(rados_ioctx_t io,
4271     const rados_object_list_cursor start,
4272     const rados_object_list_cursor finish,
4273     const size_t result_item_count,
4274     const char *filter_buf,
4275     const size_t filter_buf_len,
4276     rados_object_list_item *result_items,
4277     rados_object_list_cursor *next)
4278 {
4279   assert(next);
4280
4281   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4282
4283   // Zero out items so that they will be safe to free later
4284   memset(result_items, 0, sizeof(rados_object_list_item) * result_item_count);
4285
4286   std::list<librados::ListObjectImpl> result;
4287   hobject_t next_hash;
4288
4289   bufferlist filter_bl;
4290   if (filter_buf != nullptr) {
4291     filter_bl.append(filter_buf, filter_buf_len);
4292   }
4293
4294   C_SaferCond cond;
4295   ctx->objecter->enumerate_objects(
4296       ctx->poolid,
4297       ctx->oloc.nspace,
4298       *((hobject_t*)start),
4299       *((hobject_t*)finish),
4300       result_item_count,
4301       filter_bl,
4302       &result,
4303       &next_hash,
4304       &cond);
4305
4306   hobject_t *next_hobj = (hobject_t*)(*next);
4307   assert(next_hobj);
4308
4309   int r = cond.wait();
4310   if (r < 0) {
4311     *next_hobj = hobject_t::get_max();
4312     return r;
4313   }
4314
4315   assert(result.size() <= result_item_count);  // Don't overflow!
4316
4317   int k = 0;
4318   for (std::list<librados::ListObjectImpl>::iterator i = result.begin();
4319        i != result.end(); ++i) {
4320     rados_object_list_item &item = result_items[k++];
4321     do_out_buffer(i->oid, &item.oid, &item.oid_length);
4322     do_out_buffer(i->nspace, &item.nspace, &item.nspace_length);
4323     do_out_buffer(i->locator, &item.locator, &item.locator_length);
4324   }
4325
4326   *next_hobj = next_hash;
4327
4328   return result.size();
4329 }
4330
4331 extern "C" void rados_object_list_free(
4332     const size_t result_size,
4333     rados_object_list_item *results)
4334 {
4335   assert(results);
4336
4337   for (unsigned int i = 0; i < result_size; ++i) {
4338     rados_buffer_free(results[i].oid);
4339     rados_buffer_free(results[i].locator);
4340     rados_buffer_free(results[i].nspace);
4341   }
4342 }
4343
4344 /* list objects */
4345
4346 extern "C" int rados_nobjects_list_open(rados_ioctx_t io, rados_list_ctx_t *listh)
4347 {
4348   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4349
4350   tracepoint(librados, rados_nobjects_list_open_enter, io);
4351
4352   Objecter::NListContext *h = new Objecter::NListContext;
4353   h->pool_id = ctx->poolid;
4354   h->pool_snap_seq = ctx->snap_seq;
4355   h->nspace = ctx->oloc.nspace; // After dropping compatibility need nspace
4356   *listh = (void *)new librados::ObjListCtx(ctx, h);
4357   tracepoint(librados, rados_nobjects_list_open_exit, 0, *listh);
4358   return 0;
4359 }
4360
4361 extern "C" void rados_nobjects_list_close(rados_list_ctx_t h)
4362 {
4363   tracepoint(librados, rados_nobjects_list_close_enter, h);
4364   librados::ObjListCtx *lh = (librados::ObjListCtx *)h;
4365   delete lh;
4366   tracepoint(librados, rados_nobjects_list_close_exit);
4367 }
4368
4369 extern "C" uint32_t rados_nobjects_list_seek(rados_list_ctx_t listctx,
4370                                             uint32_t pos)
4371 {
4372   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4373   tracepoint(librados, rados_nobjects_list_seek_enter, listctx, pos);
4374   uint32_t r = lh->ctx->nlist_seek(lh->nlc, pos);
4375   tracepoint(librados, rados_nobjects_list_seek_exit, r);
4376   return r;
4377 }
4378
4379 extern "C" uint32_t rados_nobjects_list_seek_cursor(rados_list_ctx_t listctx,
4380                                                     rados_object_list_cursor cursor)
4381 {
4382   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4383
4384   tracepoint(librados, rados_nobjects_list_seek_cursor_enter, listctx);
4385   uint32_t r = lh->ctx->nlist_seek(lh->nlc, cursor);
4386   tracepoint(librados, rados_nobjects_list_seek_cursor_exit, r);
4387   return r;
4388 }
4389
4390 extern "C" int rados_nobjects_list_get_cursor(rados_list_ctx_t listctx,
4391                                               rados_object_list_cursor *cursor)
4392 {
4393   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4394
4395   tracepoint(librados, rados_nobjects_list_get_cursor_enter, listctx);
4396   *cursor = lh->ctx->nlist_get_cursor(lh->nlc);
4397   tracepoint(librados, rados_nobjects_list_get_cursor_exit, 0);
4398   return 0;
4399 }
4400
4401 extern "C" uint32_t rados_nobjects_list_get_pg_hash_position(
4402   rados_list_ctx_t listctx)
4403 {
4404   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4405   tracepoint(librados, rados_nobjects_list_get_pg_hash_position_enter, listctx);
4406   uint32_t retval = lh->nlc->get_pg_hash_position();
4407   tracepoint(librados, rados_nobjects_list_get_pg_hash_position_exit, retval);
4408   return retval;
4409 }
4410
4411 extern "C" int rados_nobjects_list_next(rados_list_ctx_t listctx, const char **entry, const char **key, const char **nspace)
4412 {
4413   tracepoint(librados, rados_nobjects_list_next_enter, listctx);
4414   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
4415   Objecter::NListContext *h = lh->nlc;
4416
4417   // if the list is non-empty, this method has been called before
4418   if (!h->list.empty())
4419     // so let's kill the previously-returned object
4420     h->list.pop_front();
4421
4422   if (h->list.empty()) {
4423     int ret = lh->ctx->nlist(lh->nlc, RADOS_LIST_MAX_ENTRIES);
4424     if (ret < 0) {
4425       tracepoint(librados, rados_nobjects_list_next_exit, ret, NULL, NULL, NULL);
4426       return ret;
4427     }
4428     if (h->list.empty()) {
4429       tracepoint(librados, rados_nobjects_list_next_exit, -ENOENT, NULL, NULL, NULL);
4430       return -ENOENT;
4431     }
4432   }
4433
4434   *entry = h->list.front().oid.c_str();
4435
4436   if (key) {
4437     if (h->list.front().locator.size())
4438       *key = h->list.front().locator.c_str();
4439     else
4440       *key = NULL;
4441   }
4442   if (nspace)
4443     *nspace = h->list.front().nspace.c_str();
4444   tracepoint(librados, rados_nobjects_list_next_exit, 0, *entry, key, nspace);
4445   return 0;
4446 }
4447
4448
4449 /*
4450  * removed legacy v2 list objects stubs
4451  *
4452  * thse return -ENOTSUP where possible.
4453  */
4454 extern "C" int rados_objects_list_open(
4455   rados_ioctx_t io,
4456   rados_list_ctx_t *ctx)
4457 {
4458   return -ENOTSUP;
4459 }
4460
4461 extern "C" uint32_t rados_objects_list_get_pg_hash_position(
4462   rados_list_ctx_t ctx)
4463 {
4464   return 0;
4465 }
4466
4467 extern "C" uint32_t rados_objects_list_seek(
4468   rados_list_ctx_t ctx,
4469   uint32_t pos)
4470 {
4471   return 0;
4472 }
4473
4474 extern "C" int rados_objects_list_next(
4475   rados_list_ctx_t ctx,
4476   const char **entry,
4477   const char **key)
4478 {
4479   return -ENOTSUP;
4480 }
4481
4482 extern "C" void rados_objects_list_close(
4483   rados_list_ctx_t ctx)
4484 {
4485 }
4486
4487
4488 // -------------------------
4489 // aio
4490
4491 extern "C" int rados_aio_create_completion(void *cb_arg,
4492                                            rados_callback_t cb_complete,
4493                                            rados_callback_t cb_safe,
4494                                            rados_completion_t *pc)
4495 {
4496   tracepoint(librados, rados_aio_create_completion_enter, cb_arg, cb_complete, cb_safe);
4497   librados::AioCompletionImpl *c = new librados::AioCompletionImpl;
4498   if (cb_complete)
4499     c->set_complete_callback(cb_arg, cb_complete);
4500   if (cb_safe)
4501     c->set_safe_callback(cb_arg, cb_safe);
4502   *pc = c;
4503   tracepoint(librados, rados_aio_create_completion_exit, 0, *pc);
4504   return 0;
4505 }
4506
4507 extern "C" int rados_aio_wait_for_complete(rados_completion_t c)
4508 {
4509   tracepoint(librados, rados_aio_wait_for_complete_enter, c);
4510   int retval = ((librados::AioCompletionImpl*)c)->wait_for_complete();
4511   tracepoint(librados, rados_aio_wait_for_complete_exit, retval);
4512   return retval;
4513 }
4514
4515 extern "C" int rados_aio_wait_for_safe(rados_completion_t c)
4516 {
4517   tracepoint(librados, rados_aio_wait_for_safe_enter, c);
4518   int retval = ((librados::AioCompletionImpl*)c)->wait_for_safe();
4519   tracepoint(librados, rados_aio_wait_for_safe_exit, retval);
4520   return retval;
4521 }
4522
4523 extern "C" int rados_aio_is_complete(rados_completion_t c)
4524 {
4525   tracepoint(librados, rados_aio_is_complete_enter, c);
4526   int retval = ((librados::AioCompletionImpl*)c)->is_complete();
4527   tracepoint(librados, rados_aio_is_complete_exit, retval);
4528   return retval;
4529 }
4530
4531 extern "C" int rados_aio_is_safe(rados_completion_t c)
4532 {
4533   tracepoint(librados, rados_aio_is_safe_enter, c);
4534   int retval = ((librados::AioCompletionImpl*)c)->is_safe();
4535   tracepoint(librados, rados_aio_is_safe_exit, retval);
4536   return retval;
4537 }
4538
4539 extern "C" int rados_aio_wait_for_complete_and_cb(rados_completion_t c)
4540 {
4541   tracepoint(librados, rados_aio_wait_for_complete_and_cb_enter, c);
4542   int retval = ((librados::AioCompletionImpl*)c)->wait_for_complete_and_cb();
4543   tracepoint(librados, rados_aio_wait_for_complete_and_cb_exit, retval);
4544   return retval;
4545 }
4546
4547 extern "C" int rados_aio_wait_for_safe_and_cb(rados_completion_t c)
4548 {
4549   tracepoint(librados, rados_aio_wait_for_safe_and_cb_enter, c);
4550   int retval = ((librados::AioCompletionImpl*)c)->wait_for_safe_and_cb();
4551   tracepoint(librados, rados_aio_wait_for_safe_and_cb_exit, retval);
4552   return retval;
4553 }
4554
4555 extern "C" int rados_aio_is_complete_and_cb(rados_completion_t c)
4556 {
4557   tracepoint(librados, rados_aio_is_complete_and_cb_enter, c);
4558   int retval = ((librados::AioCompletionImpl*)c)->is_complete_and_cb();
4559   tracepoint(librados, rados_aio_is_complete_and_cb_exit, retval);
4560   return retval;
4561 }
4562
4563 extern "C" int rados_aio_is_safe_and_cb(rados_completion_t c)
4564 {
4565   tracepoint(librados, rados_aio_is_safe_and_cb_enter, c);
4566   int retval = ((librados::AioCompletionImpl*)c)->is_safe_and_cb();
4567   tracepoint(librados, rados_aio_is_safe_and_cb_exit, retval);
4568   return retval;
4569 }
4570
4571 extern "C" int rados_aio_get_return_value(rados_completion_t c)
4572 {
4573   tracepoint(librados, rados_aio_get_return_value_enter, c);
4574   int retval = ((librados::AioCompletionImpl*)c)->get_return_value();
4575   tracepoint(librados, rados_aio_get_return_value_exit, retval);
4576   return retval;
4577 }
4578
4579 extern "C" uint64_t rados_aio_get_version(rados_completion_t c)
4580 {
4581   tracepoint(librados, rados_aio_get_version_enter, c);
4582   uint64_t retval = ((librados::AioCompletionImpl*)c)->get_version();
4583   tracepoint(librados, rados_aio_get_version_exit, retval);
4584   return retval;
4585 }
4586
4587 extern "C" void rados_aio_release(rados_completion_t c)
4588 {
4589   tracepoint(librados, rados_aio_release_enter, c);
4590   ((librados::AioCompletionImpl*)c)->put();
4591   tracepoint(librados, rados_aio_release_exit);
4592 }
4593
4594 extern "C" int rados_aio_read(rados_ioctx_t io, const char *o,
4595                                rados_completion_t completion,
4596                                char *buf, size_t len, uint64_t off)
4597 {
4598   tracepoint(librados, rados_aio_read_enter, io, o, completion, len, off);
4599   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4600   object_t oid(o);
4601   int retval = ctx->aio_read(oid, (librados::AioCompletionImpl*)completion,
4602                        buf, len, off, ctx->snap_seq);
4603   tracepoint(librados, rados_aio_read_exit, retval);
4604   return retval;
4605 }
4606
4607 #ifdef WITH_BLKIN
4608 extern "C" int rados_aio_read_traced(rados_ioctx_t io, const char *o,
4609                                      rados_completion_t completion,
4610                                      char *buf, size_t len, uint64_t off,
4611                                      struct blkin_trace_info *info)
4612 {
4613   tracepoint(librados, rados_aio_read_enter, io, o, completion, len, off);
4614   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4615   object_t oid(o);
4616   int retval = ctx->aio_read(oid, (librados::AioCompletionImpl*)completion,
4617                              buf, len, off, ctx->snap_seq, info);
4618   tracepoint(librados, rados_aio_read_exit, retval);
4619   return retval;
4620 }
4621 #endif
4622
4623 extern "C" int rados_aio_write(rados_ioctx_t io, const char *o,
4624                                 rados_completion_t completion,
4625                                 const char *buf, size_t len, uint64_t off)
4626 {
4627   tracepoint(librados, rados_aio_write_enter, io, o, completion, buf, len, off);
4628   if (len > UINT_MAX/2)
4629     return -E2BIG;
4630   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4631   object_t oid(o);
4632   bufferlist bl;
4633   bl.append(buf, len);
4634   int retval = ctx->aio_write(oid, (librados::AioCompletionImpl*)completion,
4635                         bl, len, off);
4636   tracepoint(librados, rados_aio_write_exit, retval);
4637   return retval;
4638 }
4639
4640 #ifdef WITH_BLKIN
4641 extern "C" int rados_aio_write_traced(rados_ioctx_t io, const char *o,
4642                                       rados_completion_t completion,
4643                                       const char *buf, size_t len, uint64_t off,
4644                                       struct blkin_trace_info *info)
4645 {
4646   tracepoint(librados, rados_aio_write_enter, io, o, completion, buf, len, off);
4647   if (len > UINT_MAX/2)
4648     return -E2BIG;
4649   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4650   object_t oid(o);
4651   bufferlist bl;
4652   bl.append(buf, len);
4653   int retval = ctx->aio_write(oid, (librados::AioCompletionImpl*)completion,
4654                               bl, len, off, info);
4655   tracepoint(librados, rados_aio_write_exit, retval);
4656   return retval;
4657 }
4658 #endif
4659
4660 extern "C" int rados_aio_append(rados_ioctx_t io, const char *o,
4661                                 rados_completion_t completion,
4662                                 const char *buf, size_t len)
4663 {
4664   tracepoint(librados, rados_aio_append_enter, io, o, completion, buf, len);
4665   if (len > UINT_MAX/2)
4666     return -E2BIG;
4667   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4668   object_t oid(o);
4669   bufferlist bl;
4670   bl.append(buf, len);
4671   int retval = ctx->aio_append(oid, (librados::AioCompletionImpl*)completion,
4672                          bl, len);
4673   tracepoint(librados, rados_aio_append_exit, retval);
4674   return retval;
4675 }
4676
4677 extern "C" int rados_aio_write_full(rados_ioctx_t io, const char *o,
4678                                     rados_completion_t completion,
4679                                     const char *buf, size_t len)
4680 {
4681   tracepoint(librados, rados_aio_write_full_enter, io, o, completion, buf, len);
4682   if (len > UINT_MAX/2)
4683     return -E2BIG;
4684   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4685   object_t oid(o);
4686   bufferlist bl;
4687   bl.append(buf, len);
4688   int retval = ctx->aio_write_full(oid, (librados::AioCompletionImpl*)completion, bl);
4689   tracepoint(librados, rados_aio_write_full_exit, retval);
4690   return retval;
4691 }
4692
4693 extern "C" int rados_aio_writesame(rados_ioctx_t io, const char *o,
4694                                    rados_completion_t completion,
4695                                    const char *buf, size_t data_len,
4696                                    size_t write_len, uint64_t off)
4697 {
4698   tracepoint(librados, rados_aio_writesame_enter, io, o, completion, buf,
4699                                                 data_len, write_len, off);
4700   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4701   object_t oid(o);
4702   bufferlist bl;
4703   bl.append(buf, data_len);
4704   int retval = ctx->aio_writesame(o, (librados::AioCompletionImpl*)completion,
4705                                   bl, write_len, off);
4706   tracepoint(librados, rados_aio_writesame_exit, retval);
4707   return retval;
4708 }
4709
4710 extern "C" int rados_aio_remove(rados_ioctx_t io, const char *o,
4711                                 rados_completion_t completion)
4712 {
4713   tracepoint(librados, rados_aio_remove_enter, io, o, completion);
4714   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4715   object_t oid(o);
4716   int retval = ctx->aio_remove(oid, (librados::AioCompletionImpl*)completion);
4717   tracepoint(librados, rados_aio_remove_exit, retval);
4718   return retval;
4719 }
4720
4721 extern "C" int rados_aio_flush_async(rados_ioctx_t io,
4722                                      rados_completion_t completion)
4723 {
4724   tracepoint(librados, rados_aio_flush_async_enter, io, completion);
4725   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4726   ctx->flush_aio_writes_async((librados::AioCompletionImpl*)completion);
4727   tracepoint(librados, rados_aio_flush_async_exit, 0);
4728   return 0;
4729 }
4730
4731 extern "C" int rados_aio_flush(rados_ioctx_t io)
4732 {
4733   tracepoint(librados, rados_aio_flush_enter, io);
4734   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4735   ctx->flush_aio_writes();
4736   tracepoint(librados, rados_aio_flush_exit, 0);
4737   return 0;
4738 }
4739
4740 struct AioGetxattrData {
4741   AioGetxattrData(char* buf, rados_completion_t c, size_t l) :
4742     user_buf(buf), len(l), user_completion((librados::AioCompletionImpl*)c) {}
4743   bufferlist bl;
4744   char* user_buf;
4745   size_t len;
4746   struct librados::C_AioCompleteAndSafe user_completion;
4747 };
4748
4749 static void rados_aio_getxattr_complete(rados_completion_t c, void *arg) {
4750   AioGetxattrData *cdata = reinterpret_cast<AioGetxattrData*>(arg);
4751   int rc = rados_aio_get_return_value(c);
4752   if (rc >= 0) {
4753     if (cdata->bl.length() > cdata->len) {
4754       rc = -ERANGE;
4755     } else {
4756       if (!cdata->bl.is_provided_buffer(cdata->user_buf))
4757         cdata->bl.copy(0, cdata->bl.length(), cdata->user_buf);
4758       rc = cdata->bl.length();
4759     }
4760   }
4761   cdata->user_completion.finish(rc);
4762   delete cdata;
4763 }
4764
4765 extern "C" int rados_aio_getxattr(rados_ioctx_t io, const char *o,
4766                                   rados_completion_t completion,
4767                                   const char *name, char *buf, size_t len)
4768 {
4769   tracepoint(librados, rados_aio_getxattr_enter, io, o, completion, name, len);
4770   // create data object to be passed to async callback
4771   AioGetxattrData *cdata = new AioGetxattrData(buf, completion, len);
4772   if (!cdata) {
4773     tracepoint(librados, rados_aio_getxattr_exit, -ENOMEM, NULL, 0);
4774     return -ENOMEM;
4775   }
4776   cdata->bl.push_back(buffer::create_static(len, buf));
4777   // create completion callback
4778   librados::AioCompletionImpl *c = new librados::AioCompletionImpl;
4779   c->set_complete_callback(cdata, rados_aio_getxattr_complete);
4780   // call async getxattr of IoCtx
4781   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4782   object_t oid(o);
4783   int ret = ctx->aio_getxattr(oid, c, name, cdata->bl);
4784   tracepoint(librados, rados_aio_getxattr_exit, ret, buf, ret);
4785   return ret;
4786 }
4787
4788 namespace {
4789 struct AioGetxattrsData {
4790   AioGetxattrsData(rados_completion_t c, rados_xattrs_iter_t *_iter) :
4791     iter(_iter), user_completion((librados::AioCompletionImpl*)c) {
4792     it = new librados::RadosXattrsIter();
4793   }
4794   ~AioGetxattrsData() {
4795     if (it) delete it;
4796   }
4797   librados::RadosXattrsIter *it;
4798   rados_xattrs_iter_t *iter;
4799   struct librados::C_AioCompleteAndSafe user_completion;
4800 };
4801 }
4802
4803 static void rados_aio_getxattrs_complete(rados_completion_t c, void *arg) {
4804   AioGetxattrsData *cdata = reinterpret_cast<AioGetxattrsData*>(arg);
4805   int rc = rados_aio_get_return_value(c);
4806   if (rc) {
4807     cdata->user_completion.finish(rc);
4808   } else {
4809     cdata->it->i = cdata->it->attrset.begin();
4810     *cdata->iter = cdata->it;
4811     cdata->it = 0;
4812     cdata->user_completion.finish(0);
4813   }
4814   delete cdata;
4815 }
4816
4817 extern "C" int rados_aio_getxattrs(rados_ioctx_t io, const char *oid,
4818                                    rados_completion_t completion,
4819                                    rados_xattrs_iter_t *iter)
4820 {
4821   tracepoint(librados, rados_aio_getxattrs_enter, io, oid, completion);
4822   // create data object to be passed to async callback
4823   AioGetxattrsData *cdata = new AioGetxattrsData(completion, iter);
4824   if (!cdata) {
4825     tracepoint(librados, rados_getxattrs_exit, -ENOMEM, NULL);
4826     return -ENOMEM;
4827   }
4828   // create completion callback
4829   librados::AioCompletionImpl *c = new librados::AioCompletionImpl;
4830   c->set_complete_callback(cdata, rados_aio_getxattrs_complete);
4831   // call async getxattrs of IoCtx
4832   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4833   object_t obj(oid);
4834   int ret = ctx->aio_getxattrs(obj, c, cdata->it->attrset);
4835   tracepoint(librados, rados_aio_getxattrs_exit, ret, cdata->it);
4836   return ret;
4837 }
4838
4839 extern "C" int rados_aio_setxattr(rados_ioctx_t io, const char *o,
4840                                   rados_completion_t completion,
4841                                   const char *name, const char *buf, size_t len)
4842 {
4843   tracepoint(librados, rados_aio_setxattr_enter, io, o, completion, name, buf, len);
4844   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4845   object_t oid(o);
4846   bufferlist bl;
4847   bl.append(buf, len);
4848   int retval = ctx->aio_setxattr(oid, (librados::AioCompletionImpl*)completion, name, bl);
4849   tracepoint(librados, rados_aio_setxattr_exit, retval);
4850   return retval;
4851 }
4852
4853 extern "C" int rados_aio_rmxattr(rados_ioctx_t io, const char *o,
4854                                  rados_completion_t completion,
4855                                  const char *name)
4856 {
4857   tracepoint(librados, rados_aio_rmxattr_enter, io, o, completion, name);
4858   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4859   object_t oid(o);
4860   int retval = ctx->aio_rmxattr(oid, (librados::AioCompletionImpl*)completion, name);
4861   tracepoint(librados, rados_aio_rmxattr_exit, retval);
4862   return retval;
4863 }
4864
4865 extern "C" int rados_aio_stat(rados_ioctx_t io, const char *o,
4866                               rados_completion_t completion,
4867                               uint64_t *psize, time_t *pmtime)
4868 {
4869   tracepoint(librados, rados_aio_stat_enter, io, o, completion);
4870   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4871   object_t oid(o);
4872   int retval = ctx->aio_stat(oid, (librados::AioCompletionImpl*)completion,
4873                        psize, pmtime);
4874   tracepoint(librados, rados_aio_stat_exit, retval);
4875   return retval;
4876 }
4877
4878 extern "C" int rados_aio_cmpext(rados_ioctx_t io, const char *o,
4879                                 rados_completion_t completion, const char *cmp_buf,
4880                                 size_t cmp_len, uint64_t off)
4881 {
4882   tracepoint(librados, rados_aio_cmpext_enter, io, o, completion, cmp_buf,
4883              cmp_len, off);
4884   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4885   object_t oid(o);
4886   int retval = ctx->aio_cmpext(oid, (librados::AioCompletionImpl*)completion,
4887                                cmp_buf, cmp_len, off);
4888   tracepoint(librados, rados_aio_cmpext_exit, retval);
4889   return retval;
4890 }
4891
4892 extern "C" int rados_aio_cancel(rados_ioctx_t io, rados_completion_t completion)
4893 {
4894   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4895   return ctx->aio_cancel((librados::AioCompletionImpl*)completion);
4896 }
4897
4898 extern "C" int rados_aio_exec(rados_ioctx_t io, const char *o,
4899                               rados_completion_t completion,
4900                               const char *cls, const char *method,
4901                               const char *inbuf, size_t in_len,
4902                               char *buf, size_t out_len)
4903 {
4904   tracepoint(librados, rados_aio_exec_enter, io, o, completion);
4905   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4906   object_t oid(o);
4907   bufferlist inbl;
4908   inbl.append(inbuf, in_len);
4909   int retval = ctx->aio_exec(oid, (librados::AioCompletionImpl*)completion,
4910                        cls, method, inbl, buf, out_len);
4911   tracepoint(librados, rados_aio_exec_exit, retval);
4912   return retval;
4913 }
4914
4915 struct C_WatchCB : public librados::WatchCtx {
4916   rados_watchcb_t wcb;
4917   void *arg;
4918   C_WatchCB(rados_watchcb_t _wcb, void *_arg) : wcb(_wcb), arg(_arg) {}
4919   void notify(uint8_t opcode, uint64_t ver, bufferlist& bl) override {
4920     wcb(opcode, ver, arg);
4921   }
4922 };
4923
4924 extern "C" int rados_watch(rados_ioctx_t io, const char *o, uint64_t ver,
4925                            uint64_t *handle,
4926                            rados_watchcb_t watchcb, void *arg)
4927 {
4928   tracepoint(librados, rados_watch_enter, io, o, ver, watchcb, arg);
4929   uint64_t *cookie = handle;
4930   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4931   object_t oid(o);
4932   C_WatchCB *wc = new C_WatchCB(watchcb, arg);
4933   int retval = ctx->watch(oid, cookie, wc, NULL, true);
4934   tracepoint(librados, rados_watch_exit, retval, *handle);
4935   return retval;
4936 }
4937
4938 struct C_WatchCB2 : public librados::WatchCtx2 {
4939   rados_watchcb2_t wcb;
4940   rados_watcherrcb_t errcb;
4941   void *arg;
4942   C_WatchCB2(rados_watchcb2_t _wcb,
4943              rados_watcherrcb_t _errcb,
4944              void *_arg) : wcb(_wcb), errcb(_errcb), arg(_arg) {}
4945   void handle_notify(uint64_t notify_id,
4946                      uint64_t cookie,
4947                      uint64_t notifier_gid,
4948                      bufferlist& bl) override {
4949     wcb(arg, notify_id, cookie, notifier_gid, bl.c_str(), bl.length());
4950   }
4951   void handle_error(uint64_t cookie, int err) override {
4952     if (errcb)
4953       errcb(arg, cookie, err);
4954   }
4955 };
4956
4957 extern "C" int rados_watch2(rados_ioctx_t io, const char *o, uint64_t *handle,
4958                             rados_watchcb2_t watchcb,
4959                             rados_watcherrcb_t watcherrcb,
4960                             void *arg) {
4961   return rados_watch3(io, o, handle, watchcb, watcherrcb, 0, arg);
4962 }
4963
4964 extern "C" int rados_watch3(rados_ioctx_t io, const char *o, uint64_t *handle,
4965                             rados_watchcb2_t watchcb,
4966                             rados_watcherrcb_t watcherrcb,
4967                             uint32_t timeout,
4968                             void *arg)
4969 {
4970   tracepoint(librados, rados_watch3_enter, io, o, handle, watchcb, timeout, arg);
4971   int ret;
4972   if (!watchcb || !o || !handle) {
4973     ret = -EINVAL;
4974   } else {
4975     uint64_t *cookie = handle;
4976     librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
4977     object_t oid(o);
4978     C_WatchCB2 *wc = new C_WatchCB2(watchcb, watcherrcb, arg);
4979     ret = ctx->watch(oid, cookie, NULL, wc, timeout, true);
4980   }
4981   tracepoint(librados, rados_watch3_exit, ret, handle ? *handle : 0);
4982   return ret;
4983 }
4984
4985 extern "C" int rados_aio_watch(rados_ioctx_t io, const char *o,
4986                                rados_completion_t completion,
4987                                uint64_t *handle,
4988                                rados_watchcb2_t watchcb,
4989                                rados_watcherrcb_t watcherrcb, void *arg) {
4990   return rados_aio_watch2(io, o, completion, handle, watchcb, watcherrcb, 0, arg);
4991 }
4992
4993 extern "C" int rados_aio_watch2(rados_ioctx_t io, const char *o,
4994                                 rados_completion_t completion,
4995                                 uint64_t *handle,
4996                                 rados_watchcb2_t watchcb,
4997                                 rados_watcherrcb_t watcherrcb,
4998                                 uint32_t timeout, void *arg)
4999 {
5000   tracepoint(librados, rados_aio_watch2_enter, io, o, completion, handle, watchcb, timeout, arg);
5001   int ret;
5002   if (!completion || !watchcb || !o || !handle) {
5003     ret = -EINVAL;
5004   } else {
5005     uint64_t *cookie = handle;
5006     librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5007     object_t oid(o);
5008     librados::AioCompletionImpl *c =
5009       reinterpret_cast<librados::AioCompletionImpl*>(completion);
5010     C_WatchCB2 *wc = new C_WatchCB2(watchcb, watcherrcb, arg);
5011     ret = ctx->aio_watch(oid, c, cookie, NULL, wc, timeout, true);
5012   }
5013   tracepoint(librados, rados_aio_watch2_exit, ret, handle ? *handle : 0);
5014   return ret;
5015 }
5016
5017
5018 extern "C" int rados_unwatch(rados_ioctx_t io, const char *o, uint64_t handle)
5019 {
5020   tracepoint(librados, rados_unwatch_enter, io, o, handle);
5021   uint64_t cookie = handle;
5022   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5023   int retval = ctx->unwatch(cookie);
5024   tracepoint(librados, rados_unwatch_exit, retval);
5025   return retval;
5026 }
5027
5028 extern "C" int rados_unwatch2(rados_ioctx_t io, uint64_t handle)
5029 {
5030   tracepoint(librados, rados_unwatch2_enter, io, handle);
5031   uint64_t cookie = handle;
5032   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5033   int retval = ctx->unwatch(cookie);
5034   tracepoint(librados, rados_unwatch2_exit, retval);
5035   return retval;
5036 }
5037
5038 extern "C" int rados_aio_unwatch(rados_ioctx_t io, uint64_t handle,
5039                                  rados_completion_t completion)
5040 {
5041   tracepoint(librados, rados_aio_unwatch_enter, io, handle, completion);
5042   uint64_t cookie = handle;
5043   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5044   librados::AioCompletionImpl *c =
5045     reinterpret_cast<librados::AioCompletionImpl*>(completion);
5046   int retval = ctx->aio_unwatch(cookie, c);
5047   tracepoint(librados, rados_aio_unwatch_exit, retval);
5048   return retval;
5049 }
5050
5051 extern "C" int rados_watch_check(rados_ioctx_t io, uint64_t handle)
5052 {
5053   tracepoint(librados, rados_watch_check_enter, io, handle);
5054   uint64_t cookie = handle;
5055   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5056   int retval = ctx->watch_check(cookie);
5057   tracepoint(librados, rados_watch_check_exit, retval);
5058   return retval;
5059 }
5060
5061 extern "C" int rados_notify(rados_ioctx_t io, const char *o,
5062                             uint64_t ver, const char *buf, int buf_len)
5063 {
5064   tracepoint(librados, rados_notify_enter, io, o, ver, buf, buf_len);
5065   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5066   object_t oid(o);
5067   bufferlist bl;
5068   if (buf) {
5069     bufferptr p = buffer::create(buf_len);
5070     memcpy(p.c_str(), buf, buf_len);
5071     bl.push_back(p);
5072   }
5073   int retval = ctx->notify(oid, bl, 0, NULL, NULL, NULL);
5074   tracepoint(librados, rados_notify_exit, retval);
5075   return retval;
5076 }
5077
5078 extern "C" int rados_notify2(rados_ioctx_t io, const char *o,
5079                              const char *buf, int buf_len,
5080                              uint64_t timeout_ms,
5081                              char **reply_buffer,
5082                              size_t *reply_buffer_len)
5083 {
5084   tracepoint(librados, rados_notify2_enter, io, o, buf, buf_len, timeout_ms);
5085   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5086   object_t oid(o);
5087   bufferlist bl;
5088   if (buf) {
5089     bufferptr p = buffer::create(buf_len);
5090     memcpy(p.c_str(), buf, buf_len);
5091     bl.push_back(p);
5092   }
5093   int ret = ctx->notify(oid, bl, timeout_ms, NULL, reply_buffer, reply_buffer_len);
5094   tracepoint(librados, rados_notify2_exit, ret);
5095   return ret;
5096 }
5097
5098 extern "C" int rados_aio_notify(rados_ioctx_t io, const char *o,
5099                                 rados_completion_t completion,
5100                                 const char *buf, int buf_len,
5101                                 uint64_t timeout_ms, char **reply_buffer,
5102                                 size_t *reply_buffer_len)
5103 {
5104   tracepoint(librados, rados_aio_notify_enter, io, o, completion, buf, buf_len,
5105              timeout_ms);
5106   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5107   object_t oid(o);
5108   bufferlist bl;
5109   if (buf) {
5110     bl.push_back(buffer::copy(buf, buf_len));
5111   }
5112   librados::AioCompletionImpl *c =
5113     reinterpret_cast<librados::AioCompletionImpl*>(completion);
5114   int ret = ctx->aio_notify(oid, c, bl, timeout_ms, NULL, reply_buffer,
5115                             reply_buffer_len);
5116   tracepoint(librados, rados_aio_notify_exit, ret);
5117   return ret;
5118 }
5119
5120 extern "C" int rados_notify_ack(rados_ioctx_t io, const char *o,
5121                                 uint64_t notify_id, uint64_t handle,
5122                                 const char *buf, int buf_len)
5123 {
5124   tracepoint(librados, rados_notify_ack_enter, io, o, notify_id, handle, buf, buf_len);
5125   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5126   object_t oid(o);
5127   bufferlist bl;
5128   if (buf) {
5129     bufferptr p = buffer::create(buf_len);
5130     memcpy(p.c_str(), buf, buf_len);
5131     bl.push_back(p);
5132   }
5133   ctx->notify_ack(oid, notify_id, handle, bl);
5134   tracepoint(librados, rados_notify_ack_exit, 0);
5135   return 0;
5136 }
5137
5138 extern "C" int rados_watch_flush(rados_t cluster)
5139 {
5140   tracepoint(librados, rados_watch_flush_enter, cluster);
5141   librados::RadosClient *client = (librados::RadosClient *)cluster;
5142   int retval = client->watch_flush();
5143   tracepoint(librados, rados_watch_flush_exit, retval);
5144   return retval;
5145 }
5146
5147 extern "C" int rados_aio_watch_flush(rados_t cluster, rados_completion_t completion)
5148 {
5149   tracepoint(librados, rados_aio_watch_flush_enter, cluster, completion);
5150   librados::RadosClient *client = (librados::RadosClient *)cluster;
5151   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
5152   int retval = client->async_watch_flush(c);
5153   tracepoint(librados, rados_aio_watch_flush_exit, retval);
5154   return retval;
5155 }
5156
5157 extern "C" int rados_set_alloc_hint(rados_ioctx_t io, const char *o,
5158                                     uint64_t expected_object_size,
5159                                     uint64_t expected_write_size)
5160 {
5161   tracepoint(librados, rados_set_alloc_hint_enter, io, o, expected_object_size, expected_write_size);
5162   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5163   object_t oid(o);
5164   int retval = ctx->set_alloc_hint(oid, expected_object_size,
5165                                    expected_write_size, 0);
5166   tracepoint(librados, rados_set_alloc_hint_exit, retval);
5167   return retval;
5168 }
5169
5170 extern "C" int rados_set_alloc_hint2(rados_ioctx_t io, const char *o,
5171                                      uint64_t expected_object_size,
5172                                      uint64_t expected_write_size,
5173                                      uint32_t flags)
5174 {
5175   tracepoint(librados, rados_set_alloc_hint2_enter, io, o, expected_object_size, expected_write_size, flags);
5176   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5177   object_t oid(o);
5178   int retval = ctx->set_alloc_hint(oid, expected_object_size,
5179                                    expected_write_size, flags);
5180   tracepoint(librados, rados_set_alloc_hint2_exit, retval);
5181   return retval;
5182 }
5183
5184 extern "C" int rados_lock_exclusive(rados_ioctx_t io, const char * o,
5185                           const char * name, const char * cookie,
5186                           const char * desc, struct timeval * duration,
5187                           uint8_t flags)
5188 {
5189   tracepoint(librados, rados_lock_exclusive_enter, io, o, name, cookie, desc, duration, flags);
5190   librados::IoCtx ctx;
5191   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5192
5193   int retval = ctx.lock_exclusive(o, name, cookie, desc, duration, flags);
5194   tracepoint(librados, rados_lock_exclusive_exit, retval);
5195   return retval;
5196 }
5197
5198 extern "C" int rados_lock_shared(rados_ioctx_t io, const char * o,
5199                           const char * name, const char * cookie,
5200                           const char * tag, const char * desc,
5201                           struct timeval * duration, uint8_t flags)
5202 {
5203   tracepoint(librados, rados_lock_shared_enter, io, o, name, cookie, tag, desc, duration, flags);
5204   librados::IoCtx ctx;
5205   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5206
5207   int retval = ctx.lock_shared(o, name, cookie, tag, desc, duration, flags);
5208   tracepoint(librados, rados_lock_shared_exit, retval);
5209   return retval;
5210 }
5211 extern "C" int rados_unlock(rados_ioctx_t io, const char *o, const char *name,
5212                             const char *cookie)
5213 {
5214   tracepoint(librados, rados_unlock_enter, io, o, name, cookie);
5215   librados::IoCtx ctx;
5216   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5217
5218   int retval = ctx.unlock(o, name, cookie);
5219   tracepoint(librados, rados_unlock_exit, retval);
5220   return retval;
5221 }
5222
5223 extern "C" int rados_aio_unlock(rados_ioctx_t io, const char *o, const char *name,
5224                                 const char *cookie, rados_completion_t completion)
5225 {
5226   tracepoint(librados, rados_aio_unlock_enter, io, o, name, cookie, completion);
5227   librados::IoCtx ctx;
5228   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5229   librados::AioCompletionImpl *comp = (librados::AioCompletionImpl*)completion;
5230   librados::AioCompletion c(comp);
5231   int retval = ctx.aio_unlock(o, name, cookie, &c);
5232   tracepoint(librados, rados_aio_unlock_exit, retval);
5233   return retval;
5234 }
5235
5236 extern "C" ssize_t rados_list_lockers(rados_ioctx_t io, const char *o,
5237                                       const char *name, int *exclusive,
5238                                       char *tag, size_t *tag_len,
5239                                       char *clients, size_t *clients_len,
5240                                       char *cookies, size_t *cookies_len,
5241                                       char *addrs, size_t *addrs_len)
5242 {
5243   tracepoint(librados, rados_list_lockers_enter, io, o, name, *tag_len, *clients_len, *cookies_len, *addrs_len);
5244   librados::IoCtx ctx;
5245   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5246   std::string name_str = name;
5247   std::string oid = o;
5248   std::string tag_str;
5249   int tmp_exclusive;
5250   std::list<librados::locker_t> lockers;
5251   int r = ctx.list_lockers(oid, name_str, &tmp_exclusive, &tag_str, &lockers);
5252   if (r < 0) {
5253     tracepoint(librados, rados_list_lockers_exit, r, *exclusive, "", *tag_len, *clients_len, *cookies_len, *addrs_len);
5254           return r;
5255   }
5256
5257   size_t clients_total = 0;
5258   size_t cookies_total = 0;
5259   size_t addrs_total = 0;
5260   list<librados::locker_t>::const_iterator it;
5261   for (it = lockers.begin(); it != lockers.end(); ++it) {
5262     clients_total += it->client.length() + 1;
5263     cookies_total += it->cookie.length() + 1;
5264     addrs_total += it->address.length() + 1;
5265   }
5266
5267   bool too_short = ((clients_total > *clients_len) ||
5268                     (cookies_total > *cookies_len) ||
5269                     (addrs_total > *addrs_len) ||
5270                     (tag_str.length() + 1 > *tag_len));
5271   *clients_len = clients_total;
5272   *cookies_len = cookies_total;
5273   *addrs_len = addrs_total;
5274   *tag_len = tag_str.length() + 1;
5275   if (too_short) {
5276     tracepoint(librados, rados_list_lockers_exit, -ERANGE, *exclusive, "", *tag_len, *clients_len, *cookies_len, *addrs_len);
5277     return -ERANGE;
5278   }
5279
5280   strcpy(tag, tag_str.c_str());
5281   char *clients_p = clients;
5282   char *cookies_p = cookies;
5283   char *addrs_p = addrs;
5284   for (it = lockers.begin(); it != lockers.end(); ++it) {
5285     strcpy(clients_p, it->client.c_str());
5286     strcpy(cookies_p, it->cookie.c_str());
5287     strcpy(addrs_p, it->address.c_str());
5288     tracepoint(librados, rados_list_lockers_locker, clients_p, cookies_p, addrs_p);
5289     clients_p += it->client.length() + 1;
5290     cookies_p += it->cookie.length() + 1;
5291     addrs_p += it->address.length() + 1;
5292   }
5293   if (tmp_exclusive)
5294     *exclusive = 1;
5295   else
5296     *exclusive = 0;
5297
5298   int retval = lockers.size();
5299   tracepoint(librados, rados_list_lockers_exit, retval, *exclusive, tag, *tag_len, *clients_len, *cookies_len, *addrs_len);
5300   return retval;
5301 }
5302
5303 extern "C" int rados_break_lock(rados_ioctx_t io, const char *o,
5304                                 const char *name, const char *client,
5305                                 const char *cookie)
5306 {
5307   tracepoint(librados, rados_break_lock_enter, io, o, name, client, cookie);
5308   librados::IoCtx ctx;
5309   librados::IoCtx::from_rados_ioctx_t(io, ctx);
5310
5311   int retval = ctx.break_lock(o, name, client, cookie);
5312   tracepoint(librados, rados_break_lock_exit, retval);
5313   return retval;
5314 }
5315
5316 extern "C" rados_write_op_t rados_create_write_op()
5317 {
5318   tracepoint(librados, rados_create_write_op_enter);
5319   rados_write_op_t retval = new (std::nothrow)::ObjectOperation;
5320   tracepoint(librados, rados_create_write_op_exit, retval);
5321   return retval;
5322 }
5323
5324 extern "C" void rados_release_write_op(rados_write_op_t write_op)
5325 {
5326   tracepoint(librados, rados_release_write_op_enter, write_op);
5327   delete (::ObjectOperation*)write_op;
5328   tracepoint(librados, rados_release_write_op_exit);
5329 }
5330
5331 extern "C" void rados_write_op_set_flags(rados_write_op_t write_op, int flags)
5332 {
5333   tracepoint(librados, rados_write_op_set_flags_enter, write_op, flags);
5334   set_op_flags((::ObjectOperation *)write_op, flags);
5335   tracepoint(librados, rados_write_op_set_flags_exit);
5336 }
5337
5338 extern "C" void rados_write_op_assert_version(rados_write_op_t write_op, uint64_t ver)
5339 {
5340   tracepoint(librados, rados_write_op_assert_version_enter, write_op, ver);
5341   ((::ObjectOperation *)write_op)->assert_version(ver);
5342   tracepoint(librados, rados_write_op_assert_version_exit);
5343 }
5344
5345 extern "C" void rados_write_op_assert_exists(rados_write_op_t write_op)
5346 {
5347   tracepoint(librados, rados_write_op_assert_exists_enter, write_op);
5348   ((::ObjectOperation *)write_op)->stat(NULL, (ceph::real_time *)NULL, NULL);
5349   tracepoint(librados, rados_write_op_assert_exists_exit);
5350 }
5351
5352 extern "C" void rados_write_op_cmpext(rados_write_op_t write_op,
5353                                       const char *cmp_buf,
5354                                       size_t cmp_len,
5355                                       uint64_t off,
5356                                       int *prval)
5357 {
5358   tracepoint(librados, rados_write_op_cmpext_enter, write_op, cmp_buf,
5359              cmp_len, off, prval);
5360   ((::ObjectOperation *)write_op)->cmpext(off, cmp_len, cmp_buf, prval);
5361   tracepoint(librados, rados_write_op_cmpext_exit);
5362 }
5363
5364 extern "C" void rados_write_op_cmpxattr(rados_write_op_t write_op,
5365                                        const char *name,
5366                                        uint8_t comparison_operator,
5367                                        const char *value,
5368                                        size_t value_len)
5369 {
5370   tracepoint(librados, rados_write_op_cmpxattr_enter, write_op, name, comparison_operator, value, value_len);
5371   bufferlist bl;
5372   bl.append(value, value_len);
5373   ((::ObjectOperation *)write_op)->cmpxattr(name,
5374                                             comparison_operator,
5375                                             CEPH_OSD_CMPXATTR_MODE_STRING,
5376                                             bl);
5377   tracepoint(librados, rados_write_op_cmpxattr_exit);
5378 }
5379
5380 static void rados_c_omap_cmp(ObjectOperation *op,
5381                              const char *key,
5382                              uint8_t comparison_operator,
5383                              const char *val,
5384                              size_t val_len,
5385                              int *prval)
5386 {
5387   bufferlist bl;
5388   bl.append(val, val_len);
5389   std::map<std::string, pair<bufferlist, int> > assertions;
5390   assertions[key] = std::make_pair(bl, comparison_operator);
5391   op->omap_cmp(assertions, prval);
5392 }
5393
5394 extern "C" void rados_write_op_omap_cmp(rados_write_op_t write_op,
5395                                         const char *key,
5396                                         uint8_t comparison_operator,
5397                                         const char *val,
5398                                         size_t val_len,
5399                                         int *prval)
5400 {
5401   tracepoint(librados, rados_write_op_omap_cmp_enter, write_op, key, comparison_operator, val, val_len, prval);
5402   rados_c_omap_cmp((::ObjectOperation *)write_op, key, comparison_operator,
5403                    val, val_len, prval);
5404   tracepoint(librados, rados_write_op_omap_cmp_exit);
5405 }
5406
5407 extern "C" void rados_write_op_setxattr(rados_write_op_t write_op,
5408                                        const char *name,
5409                                        const char *value,
5410                                        size_t value_len)
5411 {
5412   tracepoint(librados, rados_write_op_setxattr_enter, write_op, name, value, value_len);
5413   bufferlist bl;
5414   bl.append(value, value_len);
5415   ((::ObjectOperation *)write_op)->setxattr(name, bl);
5416   tracepoint(librados, rados_write_op_setxattr_exit);
5417 }
5418
5419 extern "C" void rados_write_op_rmxattr(rados_write_op_t write_op,
5420                                        const char *name)
5421 {
5422   tracepoint(librados, rados_write_op_rmxattr_enter, write_op, name);
5423   ((::ObjectOperation *)write_op)->rmxattr(name);
5424   tracepoint(librados, rados_write_op_rmxattr_exit);
5425 }
5426
5427 extern "C" void rados_write_op_create(rados_write_op_t write_op,
5428                                       int exclusive,
5429                                       const char* category) // unused
5430 {
5431   tracepoint(librados, rados_write_op_create_enter, write_op, exclusive);
5432   ::ObjectOperation *oo = (::ObjectOperation *) write_op;
5433   oo->create(!!exclusive);
5434   tracepoint(librados, rados_write_op_create_exit);
5435 }
5436
5437 extern "C" void rados_write_op_write(rados_write_op_t write_op,
5438                                      const char *buffer,
5439                                      size_t len,
5440                                      uint64_t offset)
5441 {
5442   tracepoint(librados, rados_write_op_write_enter, write_op, buffer, len, offset);
5443   bufferlist bl;
5444   bl.append(buffer,len);
5445   ((::ObjectOperation *)write_op)->write(offset, bl);
5446   tracepoint(librados, rados_write_op_write_exit);
5447 }
5448
5449 extern "C" void rados_write_op_write_full(rados_write_op_t write_op,
5450                                           const char *buffer,
5451                                           size_t len)
5452 {
5453   tracepoint(librados, rados_write_op_write_full_enter, write_op, buffer, len);
5454   bufferlist bl;
5455   bl.append(buffer,len);
5456   ((::ObjectOperation *)write_op)->write_full(bl);
5457   tracepoint(librados, rados_write_op_write_full_exit);
5458 }
5459
5460 extern "C" void rados_write_op_writesame(rados_write_op_t write_op,
5461                                          const char *buffer,
5462                                          size_t data_len,
5463                                          size_t write_len,
5464                                          uint64_t offset)
5465 {
5466   tracepoint(librados, rados_write_op_writesame_enter, write_op, buffer, data_len, write_len, offset);
5467   bufferlist bl;
5468   bl.append(buffer, data_len);
5469   ((::ObjectOperation *)write_op)->writesame(offset, write_len, bl);
5470   tracepoint(librados, rados_write_op_writesame_exit);
5471 }
5472
5473 extern "C" void rados_write_op_append(rados_write_op_t write_op,
5474                                       const char *buffer,
5475                                       size_t len)
5476 {
5477   tracepoint(librados, rados_write_op_append_enter, write_op, buffer, len);
5478   bufferlist bl;
5479   bl.append(buffer,len);
5480   ((::ObjectOperation *)write_op)->append(bl);
5481   tracepoint(librados, rados_write_op_append_exit);
5482 }
5483
5484 extern "C" void rados_write_op_remove(rados_write_op_t write_op)
5485 {
5486   tracepoint(librados, rados_write_op_remove_enter, write_op);
5487   ((::ObjectOperation *)write_op)->remove();
5488   tracepoint(librados, rados_write_op_remove_exit);
5489 }
5490
5491 extern "C" void rados_write_op_truncate(rados_write_op_t write_op,
5492                                         uint64_t offset)
5493 {
5494   tracepoint(librados, rados_write_op_truncate_enter, write_op, offset);
5495   ((::ObjectOperation *)write_op)->truncate(offset);
5496   tracepoint(librados, rados_write_op_truncate_exit);
5497 }
5498
5499 extern "C" void rados_write_op_zero(rados_write_op_t write_op,
5500                                     uint64_t offset,
5501                                     uint64_t len)
5502 {
5503   tracepoint(librados, rados_write_op_zero_enter, write_op, offset, len);
5504   ((::ObjectOperation *)write_op)->zero(offset, len);
5505   tracepoint(librados, rados_write_op_zero_exit);
5506 }
5507
5508 extern "C" void rados_write_op_exec(rados_write_op_t write_op,
5509                                     const char *cls,
5510                                     const char *method,
5511                                     const char *in_buf,
5512                                     size_t in_len,
5513                                     int *prval)
5514 {
5515   tracepoint(librados, rados_write_op_exec_enter, write_op, cls, method, in_buf, in_len, prval);
5516   bufferlist inbl;
5517   inbl.append(in_buf, in_len);
5518   ((::ObjectOperation *)write_op)->call(cls, method, inbl, NULL, NULL, prval);
5519   tracepoint(librados, rados_write_op_exec_exit);
5520 }
5521
5522 extern "C" void rados_write_op_omap_set(rados_write_op_t write_op,
5523                                         char const* const* keys,
5524                                         char const* const* vals,
5525                                         const size_t *lens,
5526                                         size_t num)
5527 {
5528   tracepoint(librados, rados_write_op_omap_set_enter, write_op, num);
5529   std::map<std::string, bufferlist> entries;
5530   for (size_t i = 0; i < num; ++i) {
5531     tracepoint(librados, rados_write_op_omap_set_entry, keys[i], vals[i], lens[i]);
5532     bufferlist bl(lens[i]);
5533     bl.append(vals[i], lens[i]);
5534     entries[keys[i]] = bl;
5535   }
5536   ((::ObjectOperation *)write_op)->omap_set(entries);
5537   tracepoint(librados, rados_write_op_omap_set_exit);
5538 }
5539
5540 extern "C" void rados_write_op_omap_rm_keys(rados_write_op_t write_op,
5541                                             char const* const* keys,
5542                                             size_t keys_len)
5543 {
5544   tracepoint(librados, rados_write_op_omap_rm_keys_enter, write_op, keys_len);
5545   for(size_t i = 0; i < keys_len; i++) {
5546     tracepoint(librados, rados_write_op_omap_rm_keys_entry, keys[i]);
5547   }
5548   std::set<std::string> to_remove(keys, keys + keys_len);
5549   ((::ObjectOperation *)write_op)->omap_rm_keys(to_remove);
5550   tracepoint(librados, rados_write_op_omap_rm_keys_exit);
5551 }
5552
5553 extern "C" void rados_write_op_omap_clear(rados_write_op_t write_op)
5554 {
5555   tracepoint(librados, rados_write_op_omap_clear_enter, write_op);
5556   ((::ObjectOperation *)write_op)->omap_clear();
5557   tracepoint(librados, rados_write_op_omap_clear_exit);
5558 }
5559
5560 extern "C" void rados_write_op_set_alloc_hint(rados_write_op_t write_op,
5561                                             uint64_t expected_object_size,
5562                                             uint64_t expected_write_size)
5563 {
5564   tracepoint(librados, rados_write_op_set_alloc_hint_enter, write_op, expected_object_size, expected_write_size);
5565   ((::ObjectOperation *)write_op)->set_alloc_hint(expected_object_size,
5566                                                   expected_write_size, 0);
5567   tracepoint(librados, rados_write_op_set_alloc_hint_exit);
5568 }
5569
5570 extern "C" void rados_write_op_set_alloc_hint2(rados_write_op_t write_op,
5571                                                uint64_t expected_object_size,
5572                                                uint64_t expected_write_size,
5573                                                uint32_t flags)
5574 {
5575   tracepoint(librados, rados_write_op_set_alloc_hint2_enter, write_op, expected_object_size, expected_write_size, flags);
5576   ((::ObjectOperation *)write_op)->set_alloc_hint(expected_object_size,
5577                                                   expected_write_size,
5578                                                   flags);
5579   tracepoint(librados, rados_write_op_set_alloc_hint2_exit);
5580 }
5581
5582 extern "C" int rados_write_op_operate(rados_write_op_t write_op,
5583                                       rados_ioctx_t io,
5584                                       const char *oid,
5585                                       time_t *mtime,
5586                                       int flags)
5587 {
5588   tracepoint(librados, rados_write_op_operate_enter, write_op, io, oid, mtime, flags);
5589   object_t obj(oid);
5590   ::ObjectOperation *oo = (::ObjectOperation *) write_op;
5591   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5592
5593   ceph::real_time *prt = NULL;
5594   ceph::real_time rt;
5595
5596   if (mtime) {
5597     rt = ceph::real_clock::from_time_t(*mtime);
5598     prt = &rt;
5599   }
5600
5601   int retval = ctx->operate(obj, oo, prt, translate_flags(flags));
5602   tracepoint(librados, rados_write_op_operate_exit, retval);
5603   return retval;
5604 }
5605
5606 extern "C" int rados_write_op_operate2(rados_write_op_t write_op,
5607                                        rados_ioctx_t io,
5608                                        const char *oid,
5609                                        struct timespec *ts,
5610                                        int flags)
5611 {
5612   tracepoint(librados, rados_write_op_operate2_enter, write_op, io, oid, ts, flags);
5613   object_t obj(oid);
5614   ::ObjectOperation *oo = (::ObjectOperation *) write_op;
5615   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5616
5617   ceph::real_time *prt = NULL;
5618   ceph::real_time rt;
5619
5620   if (ts) {
5621     rt = ceph::real_clock::from_timespec(*ts);
5622     prt = &rt;
5623   }
5624
5625   int retval = ctx->operate(obj, oo, prt, translate_flags(flags));
5626   tracepoint(librados, rados_write_op_operate_exit, retval);
5627   return retval;
5628 }
5629
5630 extern "C" int rados_aio_write_op_operate(rados_write_op_t write_op,
5631                                           rados_ioctx_t io,
5632                                           rados_completion_t completion,
5633                                           const char *oid,
5634                                           time_t *mtime,
5635                                           int flags)
5636 {
5637   tracepoint(librados, rados_aio_write_op_operate_enter, write_op, io, completion, oid, mtime, flags);
5638   object_t obj(oid);
5639   ::ObjectOperation *oo = (::ObjectOperation *) write_op;
5640   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
5641   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
5642   int retval = ctx->aio_operate(obj, oo, c, ctx->snapc, translate_flags(flags));
5643   tracepoint(librados, rados_aio_write_op_operate_exit, retval);
5644   return retval;
5645 }
5646
5647 extern "C" rados_read_op_t rados_create_read_op()
5648 {
5649   tracepoint(librados, rados_create_read_op_enter);
5650   rados_read_op_t retval = new (std::nothrow)::ObjectOperation;
5651   tracepoint(librados, rados_create_read_op_exit, retval);
5652   return retval;
5653 }
5654
5655 extern "C" void rados_release_read_op(rados_read_op_t read_op)
5656 {
5657   tracepoint(librados, rados_release_read_op_enter, read_op);
5658   delete (::ObjectOperation *)read_op;
5659   tracepoint(librados, rados_release_read_op_exit);
5660 }
5661
5662 extern "C" void rados_read_op_set_flags(rados_read_op_t read_op, int flags)
5663 {
5664   tracepoint(librados, rados_read_op_set_flags_enter, read_op, flags);
5665   set_op_flags((::ObjectOperation *)read_op, flags);
5666   tracepoint(librados, rados_read_op_set_flags_exit);
5667 }
5668
5669 extern "C" void rados_read_op_assert_version(rados_read_op_t read_op, uint64_t ver)
5670 {
5671   tracepoint(librados, rados_read_op_assert_version_enter, read_op, ver);
5672   ((::ObjectOperation *)read_op)->assert_version(ver);
5673   tracepoint(librados, rados_read_op_assert_version_exit);
5674 }
5675
5676 extern "C" void rados_read_op_assert_exists(rados_read_op_t read_op)
5677 {
5678   tracepoint(librados, rados_read_op_assert_exists_enter, read_op);
5679   ((::ObjectOperation *)read_op)->stat(NULL, (ceph::real_time *)NULL, NULL);
5680   tracepoint(librados, rados_read_op_assert_exists_exit);
5681 }
5682
5683 extern "C" void rados_read_op_cmpext(rados_read_op_t read_op,
5684                                      const char *cmp_buf,
5685                                      size_t cmp_len,
5686                                      uint64_t off,
5687                                      int *prval)
5688 {
5689   tracepoint(librados, rados_read_op_cmpext_enter, read_op, cmp_buf,
5690              cmp_len, off, prval);
5691   ((::ObjectOperation *)read_op)->cmpext(off, cmp_len, cmp_buf, prval);
5692   tracepoint(librados, rados_read_op_cmpext_exit);
5693 }
5694
5695 extern "C" void rados_read_op_cmpxattr(rados_read_op_t read_op,
5696                                        const char *name,
5697                                        uint8_t comparison_operator,
5698                                        const char *value,
5699                                        size_t value_len)
5700 {
5701   tracepoint(librados, rados_read_op_cmpxattr_enter, read_op, name, comparison_operator, value, value_len);
5702   bufferlist bl;
5703   bl.append(value, value_len);
5704   ((::ObjectOperation *)read_op)->cmpxattr(name,
5705                                            comparison_operator,
5706                                            CEPH_OSD_CMPXATTR_MODE_STRING,
5707                                            bl);
5708   tracepoint(librados, rados_read_op_cmpxattr_exit);
5709 }
5710
5711 extern "C" void rados_read_op_omap_cmp(rados_read_op_t read_op,
5712                                        const char *key,
5713                                        uint8_t comparison_operator,
5714                                        const char *val,
5715                                        size_t val_len,
5716                                        int *prval)
5717 {
5718   tracepoint(librados, rados_read_op_omap_cmp_enter, read_op, key, comparison_operator, val, val_len, prval);
5719   rados_c_omap_cmp((::ObjectOperation *)read_op, key, comparison_operator,
5720                    val, val_len, prval);
5721   tracepoint(librados, rados_read_op_omap_cmp_exit);
5722 }
5723
5724 extern "C" void rados_read_op_stat(rados_read_op_t read_op,
5725                                    uint64_t *psize,
5726                                    time_t *pmtime,
5727                                    int *prval)
5728 {
5729   tracepoint(librados, rados_read_op_stat_enter, read_op, psize, pmtime, prval);
5730   ((::ObjectOperation *)read_op)->stat(psize, pmtime, prval);
5731   tracepoint(librados, rados_read_op_stat_exit);
5732 }
5733
5734 class C_bl_to_buf : public Context {
5735   char *out_buf;
5736   size_t out_len;
5737   size_t *bytes_read;
5738   int *prval;
5739 public:
5740   bufferlist out_bl;
5741   C_bl_to_buf(char *out_buf,
5742               size_t out_len,
5743               size_t *bytes_read,
5744               int *prval) : out_buf(out_buf), out_len(out_len),
5745                             bytes_read(bytes_read), prval(prval) {}
5746   void finish(int r) override {
5747     if (out_bl.length() > out_len) {
5748       if (prval)
5749         *prval = -ERANGE;
5750       if (bytes_read)
5751         *bytes_read = 0;
5752       return;
5753     }
5754     if (bytes_read)
5755       *bytes_read = out_bl.length();
5756     if (out_buf && !out_bl.is_provided_buffer(out_buf))
5757       out_bl.copy(0, out_bl.length(), out_buf);
5758   }
5759 };
5760
5761 extern "C" void rados_read_op_read(rados_read_op_t read_op,
5762                                    uint64_t offset,
5763                                    size_t len,
5764                                    char *buf,
5765                                    size_t *bytes_read,
5766                                    int *prval)
5767 {
5768   tracepoint(librados, rados_read_op_read_enter, read_op, offset, len, buf, bytes_read, prval);
5769   C_bl_to_buf *ctx = new C_bl_to_buf(buf, len, bytes_read, prval);
5770   ctx->out_bl.push_back(buffer::create_static(len, buf));
5771   ((::ObjectOperation *)read_op)->read(offset, len, &ctx->out_bl, prval, ctx);
5772   tracepoint(librados, rados_read_op_read_exit);
5773 }
5774
5775 extern "C" void rados_read_op_checksum(rados_read_op_t read_op,
5776                                        rados_checksum_type_t type,
5777                                        const char *init_value,
5778                                        size_t init_value_len,
5779                                        uint64_t offset, size_t len,
5780                                        size_t chunk_size, char *pchecksum,
5781                                        size_t checksum_len, int *prval)
5782 {
5783   tracepoint(librados, rados_read_op_checksum_enter, read_op, type, init_value,
5784              init_value_len, offset, len, chunk_size);
5785   bufferlist init_value_bl;
5786   init_value_bl.append(init_value, init_value_len);
5787
5788   C_bl_to_buf *ctx = nullptr;
5789   if (pchecksum != nullptr) {
5790     ctx = new C_bl_to_buf(pchecksum, checksum_len, nullptr, prval);
5791   }
5792   ((::ObjectOperation *)read_op)->checksum(get_checksum_op_type(type),
5793                                            init_value_bl, offset, len,
5794                                            chunk_size,
5795                                            (ctx ? &ctx->out_bl : nullptr),
5796                                            prval, ctx);
5797   tracepoint(librados, rados_read_op_checksum_exit);
5798 }
5799
5800 class C_out_buffer : public Context {
5801   char **out_buf;
5802   size_t *out_len;
5803 public:
5804   bufferlist out_bl;
5805   C_out_buffer(char **out_buf, size_t *out_len) : out_buf(out_buf),
5806                                                   out_len(out_len) {}
5807   void finish(int r) override {
5808     // ignore r since we don't know the meaning of return values
5809     // from custom class methods
5810     do_out_buffer(out_bl, out_buf, out_len);
5811   }
5812 };
5813
5814 extern "C" void rados_read_op_exec(rados_read_op_t read_op,
5815                                    const char *cls,
5816                                    const char *method,
5817                                    const char *in_buf,
5818                                    size_t in_len,
5819                                    char **out_buf,
5820                                    size_t *out_len,
5821                                    int *prval)
5822 {
5823   tracepoint(librados, rados_read_op_exec_enter, read_op, cls, method, in_buf, in_len, out_buf, out_len, prval);
5824   bufferlist inbl;
5825   inbl.append(in_buf, in_len);
5826   C_out_buffer *ctx = new C_out_buffer(out_buf, out_len);
5827   ((::ObjectOperation *)read_op)->call(cls, method, inbl, &ctx->out_bl, ctx,
5828                                        prval);
5829   tracepoint(librados, rados_read_op_exec_exit);
5830 }
5831
5832 extern "C" void rados_read_op_exec_user_buf(rados_read_op_t read_op,
5833                                             const char *cls,
5834                                             const char *method,
5835                                             const char *in_buf,
5836                                             size_t in_len,
5837                                             char *out_buf,
5838                                             size_t out_len,
5839                                             size_t *used_len,
5840                                             int *prval)
5841 {
5842   tracepoint(librados, rados_read_op_exec_user_buf_enter, read_op, cls, method, in_buf, in_len, out_buf, out_len, used_len, prval);
5843   C_bl_to_buf *ctx = new C_bl_to_buf(out_buf, out_len, used_len, prval);
5844   bufferlist inbl;
5845   inbl.append(in_buf, in_len);
5846   ((::ObjectOperation *)read_op)->call(cls, method, inbl, &ctx->out_bl, ctx,
5847                                        prval);
5848   tracepoint(librados, rados_read_op_exec_user_buf_exit);
5849 }
5850
5851 struct RadosOmapIter {
5852   std::map<std::string, bufferlist> values;
5853   std::map<std::string, bufferlist>::iterator i;
5854 };
5855
5856 class C_OmapIter : public Context {
5857   RadosOmapIter *iter;
5858 public:
5859   explicit C_OmapIter(RadosOmapIter *iter) : iter(iter) {}
5860   void finish(int r) override {
5861     iter->i = iter->values.begin();
5862   }
5863 };
5864
5865 class C_XattrsIter : public Context {
5866   librados::RadosXattrsIter *iter;
5867 public:
5868   explicit C_XattrsIter(librados::RadosXattrsIter *iter) : iter(iter) {}
5869   void finish(int r) override {
5870     iter->i = iter->attrset.begin();
5871   }
5872 };
5873
5874 extern "C" void rados_read_op_getxattrs(rados_read_op_t read_op,
5875                                         rados_xattrs_iter_t *iter,
5876                                         int *prval)
5877 {
5878   tracepoint(librados, rados_read_op_getxattrs_enter, read_op, prval);
5879   librados::RadosXattrsIter *xattrs_iter = new librados::RadosXattrsIter;
5880   ((::ObjectOperation *)read_op)->getxattrs(&xattrs_iter->attrset, prval);
5881   ((::ObjectOperation *)read_op)->add_handler(new C_XattrsIter(xattrs_iter));
5882   *iter = xattrs_iter;
5883   tracepoint(librados, rados_read_op_getxattrs_exit, *iter);
5884 }
5885
5886 extern "C" void rados_read_op_omap_get_vals(rados_read_op_t read_op,
5887                                             const char *start_after,
5888                                             const char *filter_prefix,
5889                                             uint64_t max_return,
5890                                             rados_omap_iter_t *iter,
5891                                             int *prval)
5892 {
5893   tracepoint(librados, rados_read_op_omap_get_vals_enter, read_op, start_after, filter_prefix, max_return, prval);
5894   RadosOmapIter *omap_iter = new RadosOmapIter;
5895   const char *start = start_after ? start_after : "";
5896   const char *filter = filter_prefix ? filter_prefix : "";
5897   ((::ObjectOperation *)read_op)->omap_get_vals(
5898     start,
5899     filter,
5900     max_return,
5901     &omap_iter->values,
5902     nullptr,
5903     prval);
5904   ((::ObjectOperation *)read_op)->add_handler(new C_OmapIter(omap_iter));
5905   *iter = omap_iter;
5906   tracepoint(librados, rados_read_op_omap_get_vals_exit, *iter);
5907 }
5908
5909 extern "C" void rados_read_op_omap_get_vals2(rados_read_op_t read_op,
5910                                              const char *start_after,
5911                                              const char *filter_prefix,
5912                                              uint64_t max_return,
5913                                              rados_omap_iter_t *iter,
5914                                              unsigned char *pmore,
5915                                              int *prval)
5916 {
5917   tracepoint(librados, rados_read_op_omap_get_vals_enter, read_op, start_after, filter_prefix, max_return, prval);
5918   RadosOmapIter *omap_iter = new RadosOmapIter;
5919   const char *start = start_after ? start_after : "";
5920   const char *filter = filter_prefix ? filter_prefix : "";
5921   ((::ObjectOperation *)read_op)->omap_get_vals(
5922     start,
5923     filter,
5924     max_return,
5925     &omap_iter->values,
5926     (bool*)pmore,
5927     prval);
5928   ((::ObjectOperation *)read_op)->add_handler(new C_OmapIter(omap_iter));
5929   *iter = omap_iter;
5930   tracepoint(librados, rados_read_op_omap_get_vals_exit, *iter);
5931 }
5932
5933 struct C_OmapKeysIter : public Context {
5934   RadosOmapIter *iter;
5935   std::set<std::string> keys;
5936   explicit C_OmapKeysIter(RadosOmapIter *iter) : iter(iter) {}
5937   void finish(int r) override {
5938     // map each key to an empty bl
5939     for (std::set<std::string>::const_iterator i = keys.begin();
5940          i != keys.end(); ++i) {
5941       iter->values[*i];
5942     }
5943     iter->i = iter->values.begin();
5944   }
5945 };
5946
5947 extern "C" void rados_read_op_omap_get_keys(rados_read_op_t read_op,
5948                                             const char *start_after,
5949                                             uint64_t max_return,
5950                                             rados_omap_iter_t *iter,
5951                                             int *prval)
5952 {
5953   tracepoint(librados, rados_read_op_omap_get_keys_enter, read_op, start_after, max_return, prval);
5954   RadosOmapIter *omap_iter = new RadosOmapIter;
5955   C_OmapKeysIter *ctx = new C_OmapKeysIter(omap_iter);
5956   ((::ObjectOperation *)read_op)->omap_get_keys(
5957     start_after ? start_after : "",
5958     max_return, &ctx->keys, nullptr, prval);
5959   ((::ObjectOperation *)read_op)->add_handler(ctx);
5960   *iter = omap_iter;
5961   tracepoint(librados, rados_read_op_omap_get_keys_exit, *iter);
5962 }
5963
5964 extern "C" void rados_read_op_omap_get_keys2(rados_read_op_t read_op,
5965                                              const char *start_after,
5966                                              uint64_t max_return,
5967                                              rados_omap_iter_t *iter,
5968                                              unsigned char *pmore,
5969                                              int *prval)
5970 {
5971   tracepoint(librados, rados_read_op_omap_get_keys_enter, read_op, start_after, max_return, prval);
5972   RadosOmapIter *omap_iter = new RadosOmapIter;
5973   C_OmapKeysIter *ctx = new C_OmapKeysIter(omap_iter);
5974   ((::ObjectOperation *)read_op)->omap_get_keys(
5975     start_after ? start_after : "",
5976     max_return, &ctx->keys,
5977     (bool*)pmore, prval);
5978   ((::ObjectOperation *)read_op)->add_handler(ctx);
5979   *iter = omap_iter;
5980   tracepoint(librados, rados_read_op_omap_get_keys_exit, *iter);
5981 }
5982
5983 extern "C" void rados_read_op_omap_get_vals_by_keys(rados_read_op_t read_op,
5984                                                     char const* const* keys,
5985                                                     size_t keys_len,
5986                                                     rados_omap_iter_t *iter,
5987                                                     int *prval)
5988 {
5989   tracepoint(librados, rados_read_op_omap_get_vals_by_keys_enter, read_op, keys, keys_len, iter, prval);
5990   std::set<std::string> to_get(keys, keys + keys_len);
5991
5992   RadosOmapIter *omap_iter = new RadosOmapIter;
5993   ((::ObjectOperation *)read_op)->omap_get_vals_by_keys(to_get,
5994                                                         &omap_iter->values,
5995                                                         prval);
5996   ((::ObjectOperation *)read_op)->add_handler(new C_OmapIter(omap_iter));
5997   *iter = omap_iter;
5998   tracepoint(librados, rados_read_op_omap_get_vals_by_keys_exit, *iter);
5999 }
6000
6001 extern "C" int rados_omap_get_next(rados_omap_iter_t iter,
6002                                    char **key,
6003                                    char **val,
6004                                    size_t *len)
6005 {
6006   tracepoint(librados, rados_omap_get_next_enter, iter);
6007   RadosOmapIter *it = static_cast<RadosOmapIter *>(iter);
6008   if (it->i == it->values.end()) {
6009     *key = NULL;
6010     *val = NULL;
6011     *len = 0;
6012     tracepoint(librados, rados_omap_get_next_exit, 0, key, val, len);
6013     return 0;
6014   }
6015   if (key)
6016     *key = (char*)it->i->first.c_str();
6017   if (val)
6018     *val = it->i->second.c_str();
6019   if (len)
6020     *len = it->i->second.length();
6021   ++it->i;
6022   tracepoint(librados, rados_omap_get_next_exit, 0, key, val, len);
6023   return 0;
6024 }
6025
6026 extern "C" void rados_omap_get_end(rados_omap_iter_t iter)
6027 {
6028   tracepoint(librados, rados_omap_get_end_enter, iter);
6029   RadosOmapIter *it = static_cast<RadosOmapIter *>(iter);
6030   delete it;
6031   tracepoint(librados, rados_omap_get_end_exit);
6032 }
6033
6034 extern "C" int rados_read_op_operate(rados_read_op_t read_op,
6035                                      rados_ioctx_t io,
6036                                      const char *oid,
6037                                      int flags)
6038 {
6039   tracepoint(librados, rados_read_op_operate_enter, read_op, io, oid, flags);
6040   object_t obj(oid);
6041   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6042   int retval = ctx->operate_read(obj, (::ObjectOperation *)read_op, NULL,
6043                                  translate_flags(flags));
6044   tracepoint(librados, rados_read_op_operate_exit, retval);
6045   return retval;
6046 }
6047
6048 extern "C" int rados_aio_read_op_operate(rados_read_op_t read_op,
6049                                          rados_ioctx_t io,
6050                                          rados_completion_t completion,
6051                                          const char *oid,
6052                                          int flags)
6053 {
6054   tracepoint(librados, rados_aio_read_op_operate_enter, read_op, io, completion, oid, flags);
6055   object_t obj(oid);
6056   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6057   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
6058   int retval = ctx->aio_operate_read(obj, (::ObjectOperation *)read_op,
6059                                      c, translate_flags(flags), NULL);
6060   tracepoint(librados, rados_aio_read_op_operate_exit, retval);
6061   return retval;
6062 }
6063
6064 extern "C" int rados_cache_pin(rados_ioctx_t io, const char *o)
6065 {
6066   tracepoint(librados, rados_cache_pin_enter, io, o);
6067   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6068   object_t oid(o);
6069   int retval = ctx->cache_pin(oid);
6070   tracepoint(librados, rados_cache_pin_exit, retval);
6071   return retval;
6072 }
6073
6074 extern "C" int rados_cache_unpin(rados_ioctx_t io, const char *o)
6075 {
6076   tracepoint(librados, rados_cache_unpin_enter, io, o);
6077   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6078   object_t oid(o);
6079   int retval = ctx->cache_unpin(oid);
6080   tracepoint(librados, rados_cache_unpin_exit, retval);
6081   return retval;
6082 }
6083
6084
6085 ///////////////////////////// ListObject //////////////////////////////
6086 librados::ListObject::ListObject() : impl(NULL)
6087 {
6088 }
6089
6090 librados::ListObject::ListObject(librados::ListObjectImpl *i): impl(i)
6091 {
6092 }
6093
6094 librados::ListObject::ListObject(const ListObject& rhs)
6095 {
6096   if (rhs.impl == NULL) {
6097     impl = NULL;
6098     return;
6099   }
6100   impl = new ListObjectImpl();
6101   *impl = *(rhs.impl);
6102 }
6103
6104 librados::ListObject& librados::ListObject::operator=(const ListObject& rhs)
6105 {
6106   if (rhs.impl == NULL) {
6107     delete impl;
6108     impl = NULL;
6109     return *this;
6110   }
6111   if (impl == NULL)
6112     impl = new ListObjectImpl();
6113   *impl = *(rhs.impl);
6114   return *this;
6115 }
6116
6117 librados::ListObject::~ListObject()
6118 {
6119   if (impl)
6120     delete impl;
6121   impl = NULL;
6122 }
6123
6124 const std::string& librados::ListObject::get_nspace() const
6125 {
6126   return impl->get_nspace();
6127 }
6128
6129 const std::string& librados::ListObject::get_oid() const
6130 {
6131   return impl->get_oid();
6132 }
6133
6134 const std::string& librados::ListObject::get_locator() const
6135 {
6136   return impl->get_locator();
6137 }
6138
6139 std::ostream& librados::operator<<(std::ostream& out, const librados::ListObject& lop)
6140 {
6141   out << *(lop.impl);
6142   return out;
6143 }
6144
6145 CEPH_RADOS_API void rados_object_list_slice(
6146     rados_ioctx_t io,
6147     const rados_object_list_cursor start,
6148     const rados_object_list_cursor finish,
6149     const size_t n,
6150     const size_t m,
6151     rados_object_list_cursor *split_start,
6152     rados_object_list_cursor *split_finish)
6153 {
6154   librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
6155
6156   assert(split_start);
6157   assert(split_finish);
6158   hobject_t *split_start_hobj = (hobject_t*)(*split_start);
6159   hobject_t *split_finish_hobj = (hobject_t*)(*split_finish);
6160   assert(split_start_hobj);
6161   assert(split_finish_hobj);
6162   hobject_t *start_hobj = (hobject_t*)(start);
6163   hobject_t *finish_hobj = (hobject_t*)(finish);
6164
6165   ctx->object_list_slice(
6166       *start_hobj,
6167       *finish_hobj,
6168       n,
6169       m,
6170       split_start_hobj,
6171       split_finish_hobj);
6172 }
6173
6174 librados::ObjectCursor::ObjectCursor()
6175 {
6176   c_cursor = (rados_object_list_cursor)new hobject_t();
6177 }
6178
6179 librados::ObjectCursor::~ObjectCursor()
6180 {
6181   hobject_t *h = (hobject_t *)c_cursor;
6182   delete h;
6183 }
6184
6185 librados::ObjectCursor::ObjectCursor(rados_object_list_cursor c)
6186 {
6187   if (!c) {
6188     c_cursor = nullptr;
6189   } else {
6190     c_cursor = (rados_object_list_cursor)new hobject_t(*(hobject_t *)c);
6191   }
6192 }
6193
6194 librados::ObjectCursor& librados::ObjectCursor::operator=(const librados::ObjectCursor& rhs)
6195 {
6196   if (rhs.c_cursor != nullptr) {
6197     hobject_t *h = (hobject_t*)rhs.c_cursor;
6198     c_cursor = (rados_object_list_cursor)(new hobject_t(*h));
6199   } else {
6200     c_cursor = nullptr;
6201   }
6202   return *this;
6203 }
6204
6205 bool librados::ObjectCursor::operator<(const librados::ObjectCursor &rhs) const
6206 {
6207   const hobject_t lhs_hobj = (c_cursor == nullptr) ? hobject_t() : *((hobject_t*)c_cursor);
6208   const hobject_t rhs_hobj = (rhs.c_cursor == nullptr) ? hobject_t() : *((hobject_t*)(rhs.c_cursor));
6209   return lhs_hobj < rhs_hobj;
6210 }
6211
6212 bool librados::ObjectCursor::operator==(const librados::ObjectCursor &rhs) const
6213 {
6214   const hobject_t lhs_hobj = (c_cursor == nullptr) ? hobject_t() : *((hobject_t*)c_cursor);
6215   const hobject_t rhs_hobj = (rhs.c_cursor == nullptr) ? hobject_t() : *((hobject_t*)(rhs.c_cursor));
6216   return cmp(lhs_hobj, rhs_hobj) == 0;
6217 }
6218 librados::ObjectCursor::ObjectCursor(const librados::ObjectCursor &rhs)
6219 {
6220   *this = rhs;
6221 }
6222
6223 librados::ObjectCursor librados::IoCtx::object_list_begin()
6224 {
6225   hobject_t *h = new hobject_t(io_ctx_impl->objecter->enumerate_objects_begin());
6226   ObjectCursor oc;
6227   oc.set((rados_object_list_cursor)h);
6228   return oc;
6229 }
6230
6231
6232 librados::ObjectCursor librados::IoCtx::object_list_end()
6233 {
6234   hobject_t *h = new hobject_t(io_ctx_impl->objecter->enumerate_objects_end());
6235   librados::ObjectCursor oc;
6236   oc.set((rados_object_list_cursor)h);
6237   return oc;
6238 }
6239
6240
6241 void librados::ObjectCursor::set(rados_object_list_cursor c)
6242 {
6243   delete (hobject_t*)c_cursor;
6244   c_cursor = c;
6245 }
6246
6247 string librados::ObjectCursor::to_str() const
6248 {
6249   stringstream ss;
6250   ss << *(hobject_t *)c_cursor;
6251   return ss.str();
6252 }
6253
6254 bool librados::ObjectCursor::from_str(const string& s)
6255 {
6256   if (s.empty()) {
6257     *(hobject_t *)c_cursor = hobject_t();
6258     return true;
6259   }
6260   return ((hobject_t *)c_cursor)->parse(s);
6261 }
6262
6263 CEPH_RADOS_API std::ostream& librados::operator<<(std::ostream& os, const librados::ObjectCursor& oc)
6264 {
6265   if (oc.c_cursor) {
6266     os << *(hobject_t *)oc.c_cursor;
6267   } else {
6268     os << hobject_t();
6269   }
6270   return os;
6271 }
6272
6273 bool librados::IoCtx::object_list_is_end(const ObjectCursor &oc)
6274 {
6275   hobject_t *h = (hobject_t *)oc.c_cursor;
6276   return h->is_max();
6277 }
6278
6279 int librados::IoCtx::object_list(const ObjectCursor &start,
6280                 const ObjectCursor &finish,
6281                 const size_t result_item_count,
6282                 const bufferlist &filter,
6283                 std::vector<ObjectItem> *result,
6284                 ObjectCursor *next)
6285 {
6286   assert(result != nullptr);
6287   assert(next != nullptr);
6288   result->clear();
6289
6290   C_SaferCond cond;
6291   hobject_t next_hash;
6292   std::list<librados::ListObjectImpl> obj_result;
6293   io_ctx_impl->objecter->enumerate_objects(
6294       io_ctx_impl->poolid,
6295       io_ctx_impl->oloc.nspace,
6296       *((hobject_t*)start.c_cursor),
6297       *((hobject_t*)finish.c_cursor),
6298       result_item_count,
6299       filter,
6300       &obj_result,
6301       &next_hash,
6302       &cond);
6303
6304   int r = cond.wait();
6305   if (r < 0) {
6306     next->set((rados_object_list_cursor)(new hobject_t(hobject_t::get_max())));
6307     return r;
6308   }
6309
6310   next->set((rados_object_list_cursor)(new hobject_t(next_hash)));
6311
6312   for (std::list<librados::ListObjectImpl>::iterator i = obj_result.begin();
6313        i != obj_result.end(); ++i) {
6314     ObjectItem oi;
6315     oi.oid = i->oid;
6316     oi.nspace = i->nspace;
6317     oi.locator = i->locator;
6318     result->push_back(oi);
6319   }
6320
6321   return obj_result.size();
6322 }
6323
6324 void librados::IoCtx::object_list_slice(
6325     const ObjectCursor start,
6326     const ObjectCursor finish,
6327     const size_t n,
6328     const size_t m,
6329     ObjectCursor *split_start,
6330     ObjectCursor *split_finish)
6331 {
6332   assert(split_start != nullptr);
6333   assert(split_finish != nullptr);
6334
6335   io_ctx_impl->object_list_slice(
6336       *((hobject_t*)(start.c_cursor)),
6337       *((hobject_t*)(finish.c_cursor)),
6338       n,
6339       m,
6340       (hobject_t*)(split_start->c_cursor),
6341       (hobject_t*)(split_finish->c_cursor));
6342 }
6343