ceph/src/librados/librados.cc

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