--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2012 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#include <limits.h>
+
+#include "IoCtxImpl.h"
+
+#include "librados/AioCompletionImpl.h"
+#include "librados/PoolAsyncCompletionImpl.h"
+#include "librados/RadosClient.h"
+#include "include/assert.h"
+#include "common/valgrind.h"
+#include "common/EventTrace.h"
+
+#define dout_subsys ceph_subsys_rados
+#undef dout_prefix
+#define dout_prefix *_dout << "librados: "
+
+namespace librados {
+namespace {
+
+struct C_notify_Finish : public Context {
+ CephContext *cct;
+ Context *ctx;
+ Objecter *objecter;
+ Objecter::LingerOp *linger_op;
+ bufferlist reply_bl;
+ bufferlist *preply_bl;
+ char **preply_buf;
+ size_t *preply_buf_len;
+
+ C_notify_Finish(CephContext *_cct, Context *_ctx, Objecter *_objecter,
+ Objecter::LingerOp *_linger_op, bufferlist *_preply_bl,
+ char **_preply_buf, size_t *_preply_buf_len)
+ : cct(_cct), ctx(_ctx), objecter(_objecter), linger_op(_linger_op),
+ preply_bl(_preply_bl), preply_buf(_preply_buf),
+ preply_buf_len(_preply_buf_len)
+ {
+ linger_op->on_notify_finish = this;
+ linger_op->notify_result_bl = &reply_bl;
+ }
+
+ void finish(int r) override
+ {
+ ldout(cct, 10) << __func__ << " completed notify (linger op "
+ << linger_op << "), r = " << r << dendl;
+
+ // pass result back to user
+ // NOTE: we do this regardless of what error code we return
+ if (preply_buf) {
+ if (reply_bl.length()) {
+ *preply_buf = (char*)malloc(reply_bl.length());
+ memcpy(*preply_buf, reply_bl.c_str(), reply_bl.length());
+ } else {
+ *preply_buf = NULL;
+ }
+ }
+ if (preply_buf_len)
+ *preply_buf_len = reply_bl.length();
+ if (preply_bl)
+ preply_bl->claim(reply_bl);
+
+ ctx->complete(r);
+ }
+};
+
+struct C_aio_linger_cancel : public Context {
+ Objecter *objecter;
+ Objecter::LingerOp *linger_op;
+
+ C_aio_linger_cancel(Objecter *_objecter, Objecter::LingerOp *_linger_op)
+ : objecter(_objecter), linger_op(_linger_op)
+ {
+ }
+
+ void finish(int r) override
+ {
+ objecter->linger_cancel(linger_op);
+ }
+};
+
+struct C_aio_linger_Complete : public Context {
+ AioCompletionImpl *c;
+ Objecter::LingerOp *linger_op;
+ bool cancel;
+
+ C_aio_linger_Complete(AioCompletionImpl *_c, Objecter::LingerOp *_linger_op, bool _cancel)
+ : c(_c), linger_op(_linger_op), cancel(_cancel)
+ {
+ c->get();
+ }
+
+ void finish(int r) override {
+ if (cancel || r < 0)
+ c->io->client->finisher.queue(new C_aio_linger_cancel(c->io->objecter,
+ linger_op));
+
+ c->lock.Lock();
+ c->rval = r;
+ c->complete = true;
+ c->cond.Signal();
+
+ if (c->callback_complete ||
+ c->callback_safe) {
+ c->io->client->finisher.queue(new C_AioComplete(c));
+ }
+ c->put_unlock();
+ }
+};
+
+struct C_aio_notify_Complete : public C_aio_linger_Complete {
+ Mutex lock;
+ bool acked = false;
+ bool finished = false;
+ int ret_val = 0;
+
+ C_aio_notify_Complete(AioCompletionImpl *_c, Objecter::LingerOp *_linger_op)
+ : C_aio_linger_Complete(_c, _linger_op, false),
+ lock("C_aio_notify_Complete::lock") {
+ }
+
+ void handle_ack(int r) {
+ // invoked by C_aio_notify_Ack
+ lock.Lock();
+ acked = true;
+ complete_unlock(r);
+ }
+
+ void complete(int r) override {
+ // invoked by C_notify_Finish (or C_aio_notify_Ack on failure)
+ lock.Lock();
+ finished = true;
+ complete_unlock(r);
+ }
+
+ void complete_unlock(int r) {
+ if (ret_val == 0 && r < 0) {
+ ret_val = r;
+ }
+
+ if (acked && finished) {
+ lock.Unlock();
+ cancel = true;
+ C_aio_linger_Complete::complete(ret_val);
+ } else {
+ lock.Unlock();
+ }
+ }
+};
+
+struct C_aio_notify_Ack : public Context {
+ CephContext *cct;
+ C_notify_Finish *onfinish;
+ C_aio_notify_Complete *oncomplete;
+
+ C_aio_notify_Ack(CephContext *_cct, C_notify_Finish *_onfinish,
+ C_aio_notify_Complete *_oncomplete)
+ : cct(_cct), onfinish(_onfinish), oncomplete(_oncomplete)
+ {
+ }
+
+ void finish(int r) override
+ {
+ ldout(cct, 10) << __func__ << " linger op " << oncomplete->linger_op << " "
+ << "acked (" << r << ")" << dendl;
+ oncomplete->handle_ack(r);
+ if (r < 0) {
+ // on failure, we won't expect to see a notify_finish callback
+ onfinish->complete(r);
+ }
+ }
+};
+
+struct C_aio_selfmanaged_snap_op_Complete : public Context {
+ librados::RadosClient *client;
+ librados::AioCompletionImpl *c;
+
+ C_aio_selfmanaged_snap_op_Complete(librados::RadosClient *client,
+ librados::AioCompletionImpl *c)
+ : client(client), c(c) {
+ c->get();
+ }
+
+ void finish(int r) override {
+ c->lock.Lock();
+ c->rval = r;
+ c->complete = true;
+ c->cond.Signal();
+
+ if (c->callback_complete || c->callback_safe) {
+ client->finisher.queue(new librados::C_AioComplete(c));
+ }
+ c->put_unlock();
+ }
+};
+
+struct C_aio_selfmanaged_snap_create_Complete : public C_aio_selfmanaged_snap_op_Complete {
+ snapid_t snapid;
+ uint64_t *dest_snapid;
+
+ C_aio_selfmanaged_snap_create_Complete(librados::RadosClient *client,
+ librados::AioCompletionImpl *c,
+ uint64_t *dest_snapid)
+ : C_aio_selfmanaged_snap_op_Complete(client, c),
+ dest_snapid(dest_snapid) {
+ }
+
+ void finish(int r) override {
+ if (r >= 0) {
+ *dest_snapid = snapid;
+ }
+ C_aio_selfmanaged_snap_op_Complete::finish(r);
+ }
+};
+
+} // anonymous namespace
+} // namespace librados
+
+librados::IoCtxImpl::IoCtxImpl() :
+ ref_cnt(0), client(NULL), poolid(0), assert_ver(0), last_objver(0),
+ notify_timeout(30), aio_write_list_lock("librados::IoCtxImpl::aio_write_list_lock"),
+ aio_write_seq(0), objecter(NULL)
+{
+}
+
+librados::IoCtxImpl::IoCtxImpl(RadosClient *c, Objecter *objecter,
+ int64_t poolid, snapid_t s)
+ : ref_cnt(0), client(c), poolid(poolid), snap_seq(s),
+ assert_ver(0), last_objver(0),
+ notify_timeout(c->cct->_conf->client_notify_timeout),
+ oloc(poolid), aio_write_list_lock("librados::IoCtxImpl::aio_write_list_lock"),
+ aio_write_seq(0), objecter(objecter)
+{
+}
+
+void librados::IoCtxImpl::set_snap_read(snapid_t s)
+{
+ if (!s)
+ s = CEPH_NOSNAP;
+ ldout(client->cct, 10) << "set snap read " << snap_seq << " -> " << s << dendl;
+ snap_seq = s;
+}
+
+int librados::IoCtxImpl::set_snap_write_context(snapid_t seq, vector<snapid_t>& snaps)
+{
+ ::SnapContext n;
+ ldout(client->cct, 10) << "set snap write context: seq = " << seq
+ << " and snaps = " << snaps << dendl;
+ n.seq = seq;
+ n.snaps = snaps;
+ if (!n.is_valid())
+ return -EINVAL;
+ snapc = n;
+ return 0;
+}
+
+int librados::IoCtxImpl::get_object_hash_position(
+ const std::string& oid, uint32_t *hash_position)
+{
+ int64_t r = objecter->get_object_hash_position(poolid, oid, oloc.nspace);
+ if (r < 0)
+ return r;
+ *hash_position = (uint32_t)r;
+ return 0;
+}
+
+int librados::IoCtxImpl::get_object_pg_hash_position(
+ const std::string& oid, uint32_t *pg_hash_position)
+{
+ int64_t r = objecter->get_object_pg_hash_position(poolid, oid, oloc.nspace);
+ if (r < 0)
+ return r;
+ *pg_hash_position = (uint32_t)r;
+ return 0;
+}
+
+void librados::IoCtxImpl::queue_aio_write(AioCompletionImpl *c)
+{
+ get();
+ aio_write_list_lock.Lock();
+ assert(c->io == this);
+ c->aio_write_seq = ++aio_write_seq;
+ ldout(client->cct, 20) << "queue_aio_write " << this << " completion " << c
+ << " write_seq " << aio_write_seq << dendl;
+ aio_write_list.push_back(&c->aio_write_list_item);
+ aio_write_list_lock.Unlock();
+}
+
+void librados::IoCtxImpl::complete_aio_write(AioCompletionImpl *c)
+{
+ ldout(client->cct, 20) << "complete_aio_write " << c << dendl;
+ aio_write_list_lock.Lock();
+ assert(c->io == this);
+ c->aio_write_list_item.remove_myself();
+
+ map<ceph_tid_t, std::list<AioCompletionImpl*> >::iterator waiters = aio_write_waiters.begin();
+ while (waiters != aio_write_waiters.end()) {
+ if (!aio_write_list.empty() &&
+ aio_write_list.front()->aio_write_seq <= waiters->first) {
+ ldout(client->cct, 20) << " next outstanding write is " << aio_write_list.front()->aio_write_seq
+ << " <= waiter " << waiters->first
+ << ", stopping" << dendl;
+ break;
+ }
+ ldout(client->cct, 20) << " waking waiters on seq " << waiters->first << dendl;
+ for (std::list<AioCompletionImpl*>::iterator it = waiters->second.begin();
+ it != waiters->second.end(); ++it) {
+ client->finisher.queue(new C_AioCompleteAndSafe(*it));
+ (*it)->put();
+ }
+ aio_write_waiters.erase(waiters++);
+ }
+
+ aio_write_cond.Signal();
+ aio_write_list_lock.Unlock();
+ put();
+}
+
+void librados::IoCtxImpl::flush_aio_writes_async(AioCompletionImpl *c)
+{
+ ldout(client->cct, 20) << "flush_aio_writes_async " << this
+ << " completion " << c << dendl;
+ Mutex::Locker l(aio_write_list_lock);
+ ceph_tid_t seq = aio_write_seq;
+ if (aio_write_list.empty()) {
+ ldout(client->cct, 20) << "flush_aio_writes_async no writes. (tid "
+ << seq << ")" << dendl;
+ client->finisher.queue(new C_AioCompleteAndSafe(c));
+ } else {
+ ldout(client->cct, 20) << "flush_aio_writes_async " << aio_write_list.size()
+ << " writes in flight; waiting on tid " << seq << dendl;
+ c->get();
+ aio_write_waiters[seq].push_back(c);
+ }
+}
+
+void librados::IoCtxImpl::flush_aio_writes()
+{
+ ldout(client->cct, 20) << "flush_aio_writes" << dendl;
+ aio_write_list_lock.Lock();
+ ceph_tid_t seq = aio_write_seq;
+ while (!aio_write_list.empty() &&
+ aio_write_list.front()->aio_write_seq <= seq)
+ aio_write_cond.Wait(aio_write_list_lock);
+ aio_write_list_lock.Unlock();
+}
+
+string librados::IoCtxImpl::get_cached_pool_name()
+{
+ std::string pn;
+ client->pool_get_name(get_id(), &pn);
+ return pn;
+}
+
+// SNAPS
+
+int librados::IoCtxImpl::snap_create(const char *snapName)
+{
+ int reply;
+ string sName(snapName);
+
+ Mutex mylock ("IoCtxImpl::snap_create::mylock");
+ Cond cond;
+ bool done;
+ Context *onfinish = new C_SafeCond(&mylock, &cond, &done, &reply);
+ reply = objecter->create_pool_snap(poolid, sName, onfinish);
+
+ if (reply < 0) {
+ delete onfinish;
+ } else {
+ mylock.Lock();
+ while (!done)
+ cond.Wait(mylock);
+ mylock.Unlock();
+ }
+ return reply;
+}
+
+int librados::IoCtxImpl::selfmanaged_snap_create(uint64_t *psnapid)
+{
+ int reply;
+
+ Mutex mylock("IoCtxImpl::selfmanaged_snap_create::mylock");
+ Cond cond;
+ bool done;
+ Context *onfinish = new C_SafeCond(&mylock, &cond, &done, &reply);
+ snapid_t snapid;
+ reply = objecter->allocate_selfmanaged_snap(poolid, &snapid, onfinish);
+
+ if (reply < 0) {
+ delete onfinish;
+ } else {
+ mylock.Lock();
+ while (!done)
+ cond.Wait(mylock);
+ mylock.Unlock();
+ if (reply == 0)
+ *psnapid = snapid;
+ }
+ return reply;
+}
+
+void librados::IoCtxImpl::aio_selfmanaged_snap_create(uint64_t *snapid,
+ AioCompletionImpl *c)
+{
+ C_aio_selfmanaged_snap_create_Complete *onfinish =
+ new C_aio_selfmanaged_snap_create_Complete(client, c, snapid);
+ int r = objecter->allocate_selfmanaged_snap(poolid, &onfinish->snapid,
+ onfinish);
+ if (r < 0) {
+ onfinish->complete(r);
+ }
+}
+
+int librados::IoCtxImpl::snap_remove(const char *snapName)
+{
+ int reply;
+ string sName(snapName);
+
+ Mutex mylock ("IoCtxImpl::snap_remove::mylock");
+ Cond cond;
+ bool done;
+ Context *onfinish = new C_SafeCond(&mylock, &cond, &done, &reply);
+ reply = objecter->delete_pool_snap(poolid, sName, onfinish);
+
+ if (reply < 0) {
+ delete onfinish;
+ } else {
+ mylock.Lock();
+ while(!done)
+ cond.Wait(mylock);
+ mylock.Unlock();
+ }
+ return reply;
+}
+
+int librados::IoCtxImpl::selfmanaged_snap_rollback_object(const object_t& oid,
+ ::SnapContext& snapc,
+ uint64_t snapid)
+{
+ int reply;
+
+ Mutex mylock("IoCtxImpl::snap_rollback::mylock");
+ Cond cond;
+ bool done;
+ Context *onack = new C_SafeCond(&mylock, &cond, &done, &reply);
+
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.rollback(snapid);
+ objecter->mutate(oid, oloc,
+ op, snapc, ceph::real_clock::now(), 0,
+ onack, NULL);
+
+ mylock.Lock();
+ while (!done) cond.Wait(mylock);
+ mylock.Unlock();
+ return reply;
+}
+
+int librados::IoCtxImpl::rollback(const object_t& oid, const char *snapName)
+{
+ snapid_t snap;
+
+ int r = objecter->pool_snap_by_name(poolid, snapName, &snap);
+ if (r < 0) {
+ return r;
+ }
+
+ return selfmanaged_snap_rollback_object(oid, snapc, snap);
+}
+
+int librados::IoCtxImpl::selfmanaged_snap_remove(uint64_t snapid)
+{
+ int reply;
+
+ Mutex mylock("IoCtxImpl::selfmanaged_snap_remove::mylock");
+ Cond cond;
+ bool done;
+ objecter->delete_selfmanaged_snap(poolid, snapid_t(snapid),
+ new C_SafeCond(&mylock, &cond, &done, &reply));
+
+ mylock.Lock();
+ while (!done) cond.Wait(mylock);
+ mylock.Unlock();
+ return (int)reply;
+}
+
+void librados::IoCtxImpl::aio_selfmanaged_snap_remove(uint64_t snapid,
+ AioCompletionImpl *c)
+{
+ Context *onfinish = new C_aio_selfmanaged_snap_op_Complete(client, c);
+ objecter->delete_selfmanaged_snap(poolid, snapid, onfinish);
+}
+
+int librados::IoCtxImpl::pool_change_auid(unsigned long long auid)
+{
+ int reply;
+
+ Mutex mylock("IoCtxImpl::pool_change_auid::mylock");
+ Cond cond;
+ bool done;
+ objecter->change_pool_auid(poolid,
+ new C_SafeCond(&mylock, &cond, &done, &reply),
+ auid);
+
+ mylock.Lock();
+ while (!done) cond.Wait(mylock);
+ mylock.Unlock();
+ return reply;
+}
+
+int librados::IoCtxImpl::pool_change_auid_async(unsigned long long auid,
+ PoolAsyncCompletionImpl *c)
+{
+ objecter->change_pool_auid(poolid,
+ new C_PoolAsync_Safe(c),
+ auid);
+ return 0;
+}
+
+int librados::IoCtxImpl::snap_list(vector<uint64_t> *snaps)
+{
+ return objecter->pool_snap_list(poolid, snaps);
+}
+
+int librados::IoCtxImpl::snap_lookup(const char *name, uint64_t *snapid)
+{
+ return objecter->pool_snap_by_name(poolid, name, (snapid_t *)snapid);
+}
+
+int librados::IoCtxImpl::snap_get_name(uint64_t snapid, std::string *s)
+{
+ pool_snap_info_t info;
+ int ret = objecter->pool_snap_get_info(poolid, snapid, &info);
+ if (ret < 0) {
+ return ret;
+ }
+ *s = info.name.c_str();
+ return 0;
+}
+
+int librados::IoCtxImpl::snap_get_stamp(uint64_t snapid, time_t *t)
+{
+ pool_snap_info_t info;
+ int ret = objecter->pool_snap_get_info(poolid, snapid, &info);
+ if (ret < 0) {
+ return ret;
+ }
+ *t = info.stamp.sec();
+ return 0;
+}
+
+
+// IO
+
+int librados::IoCtxImpl::nlist(Objecter::NListContext *context, int max_entries)
+{
+ Cond cond;
+ bool done;
+ int r = 0;
+ Mutex mylock("IoCtxImpl::nlist::mylock");
+
+ if (context->at_end())
+ return 0;
+
+ context->max_entries = max_entries;
+ context->nspace = oloc.nspace;
+
+ objecter->list_nobjects(context, new C_SafeCond(&mylock, &cond, &done, &r));
+
+ mylock.Lock();
+ while(!done)
+ cond.Wait(mylock);
+ mylock.Unlock();
+
+ return r;
+}
+
+uint32_t librados::IoCtxImpl::nlist_seek(Objecter::NListContext *context,
+ uint32_t pos)
+{
+ context->list.clear();
+ return objecter->list_nobjects_seek(context, pos);
+}
+
+uint32_t librados::IoCtxImpl::nlist_seek(Objecter::NListContext *context,
+ const rados_object_list_cursor& cursor)
+{
+ context->list.clear();
+ return objecter->list_nobjects_seek(context, *(const hobject_t *)cursor);
+}
+
+rados_object_list_cursor librados::IoCtxImpl::nlist_get_cursor(Objecter::NListContext *context)
+{
+ hobject_t *c = new hobject_t;
+
+ objecter->list_nobjects_get_cursor(context, c);
+ return (rados_object_list_cursor)c;
+}
+
+int librados::IoCtxImpl::create(const object_t& oid, bool exclusive)
+{
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.create(exclusive);
+ return operate(oid, &op, NULL);
+}
+
+/*
+ * add any version assert operations that are appropriate given the
+ * stat in the IoCtx, either the target version assert or any src
+ * object asserts. these affect a single ioctx operation, so clear
+ * the ioctx state when we're doing.
+ *
+ * return a pointer to the ObjectOperation if we added any events;
+ * this is convenient for passing the extra_ops argument into Objecter
+ * methods.
+ */
+::ObjectOperation *librados::IoCtxImpl::prepare_assert_ops(::ObjectOperation *op)
+{
+ ::ObjectOperation *pop = NULL;
+ if (assert_ver) {
+ op->assert_version(assert_ver);
+ assert_ver = 0;
+ pop = op;
+ }
+ return pop;
+}
+
+int librados::IoCtxImpl::write(const object_t& oid, bufferlist& bl,
+ size_t len, uint64_t off)
+{
+ if (len > UINT_MAX/2)
+ return -E2BIG;
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ bufferlist mybl;
+ mybl.substr_of(bl, 0, len);
+ op.write(off, mybl);
+ return operate(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::append(const object_t& oid, bufferlist& bl, size_t len)
+{
+ if (len > UINT_MAX/2)
+ return -E2BIG;
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ bufferlist mybl;
+ mybl.substr_of(bl, 0, len);
+ op.append(mybl);
+ return operate(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::write_full(const object_t& oid, bufferlist& bl)
+{
+ if (bl.length() > UINT_MAX/2)
+ return -E2BIG;
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.write_full(bl);
+ return operate(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::writesame(const object_t& oid, bufferlist& bl,
+ size_t write_len, uint64_t off)
+{
+ if ((bl.length() > UINT_MAX/2) || (write_len > UINT_MAX/2))
+ return -E2BIG;
+ if ((bl.length() == 0) || (write_len % bl.length()))
+ return -EINVAL;
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ bufferlist mybl;
+ mybl.substr_of(bl, 0, bl.length());
+ op.writesame(off, write_len, mybl);
+ return operate(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::operate(const object_t& oid, ::ObjectOperation *o,
+ ceph::real_time *pmtime, int flags)
+{
+ ceph::real_time ut = (pmtime ? *pmtime :
+ ceph::real_clock::now());
+
+ /* can't write to a snapshot */
+ if (snap_seq != CEPH_NOSNAP)
+ return -EROFS;
+
+ if (!o->size())
+ return 0;
+
+ Mutex mylock("IoCtxImpl::operate::mylock");
+ Cond cond;
+ bool done;
+ int r;
+ version_t ver;
+
+ Context *oncommit = new C_SafeCond(&mylock, &cond, &done, &r);
+
+ int op = o->ops[0].op.op;
+ ldout(client->cct, 10) << ceph_osd_op_name(op) << " oid=" << oid
+ << " nspace=" << oloc.nspace << dendl;
+ Objecter::Op *objecter_op = objecter->prepare_mutate_op(oid, oloc,
+ *o, snapc, ut, flags,
+ oncommit, &ver);
+ objecter->op_submit(objecter_op);
+
+ mylock.Lock();
+ while (!done)
+ cond.Wait(mylock);
+ mylock.Unlock();
+ ldout(client->cct, 10) << "Objecter returned from "
+ << ceph_osd_op_name(op) << " r=" << r << dendl;
+
+ set_sync_op_version(ver);
+
+ return r;
+}
+
+int librados::IoCtxImpl::operate_read(const object_t& oid,
+ ::ObjectOperation *o,
+ bufferlist *pbl,
+ int flags)
+{
+ if (!o->size())
+ return 0;
+
+ Mutex mylock("IoCtxImpl::operate_read::mylock");
+ Cond cond;
+ bool done;
+ int r;
+ version_t ver;
+
+ Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
+
+ int op = o->ops[0].op.op;
+ ldout(client->cct, 10) << ceph_osd_op_name(op) << " oid=" << oid << " nspace=" << oloc.nspace << dendl;
+ Objecter::Op *objecter_op = objecter->prepare_read_op(oid, oloc,
+ *o, snap_seq, pbl, flags,
+ onack, &ver);
+ objecter->op_submit(objecter_op);
+
+ mylock.Lock();
+ while (!done)
+ cond.Wait(mylock);
+ mylock.Unlock();
+ ldout(client->cct, 10) << "Objecter returned from "
+ << ceph_osd_op_name(op) << " r=" << r << dendl;
+
+ set_sync_op_version(ver);
+
+ return r;
+}
+
+int librados::IoCtxImpl::aio_operate_read(const object_t &oid,
+ ::ObjectOperation *o,
+ AioCompletionImpl *c,
+ int flags,
+ bufferlist *pbl,
+ const blkin_trace_info *trace_info)
+{
+ FUNCTRACE();
+ Context *oncomplete = new C_aio_Complete(c);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+ c->is_read = true;
+ c->io = this;
+
+ ZTracer::Trace trace;
+ if (trace_info) {
+ ZTracer::Trace parent_trace("", nullptr, trace_info);
+ trace.init("rados operate read", &objecter->trace_endpoint, &parent_trace);
+ }
+
+ trace.event("init root span");
+ Objecter::Op *objecter_op = objecter->prepare_read_op(oid, oloc,
+ *o, snap_seq, pbl, flags,
+ oncomplete, &c->objver, nullptr, 0, &trace);
+ objecter->op_submit(objecter_op, &c->tid);
+ trace.event("rados operate read submitted");
+
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_operate(const object_t& oid,
+ ::ObjectOperation *o, AioCompletionImpl *c,
+ const SnapContext& snap_context, int flags,
+ const blkin_trace_info *trace_info)
+{
+ FUNCTRACE();
+ OID_EVENT_TRACE(oid.name.c_str(), "RADOS_WRITE_OP_BEGIN");
+ auto ut = ceph::real_clock::now();
+ /* can't write to a snapshot */
+ if (snap_seq != CEPH_NOSNAP)
+ return -EROFS;
+
+ Context *oncomplete = new C_aio_Complete(c);
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+
+ c->io = this;
+ queue_aio_write(c);
+
+ ZTracer::Trace trace;
+ if (trace_info) {
+ ZTracer::Trace parent_trace("", nullptr, trace_info);
+ trace.init("rados operate", &objecter->trace_endpoint, &parent_trace);
+ }
+
+ trace.event("init root span");
+ Objecter::Op *op = objecter->prepare_mutate_op(
+ oid, oloc, *o, snap_context, ut, flags,
+ oncomplete, &c->objver, osd_reqid_t(), &trace);
+ objecter->op_submit(op, &c->tid);
+ trace.event("rados operate op submitted");
+
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_read(const object_t oid, AioCompletionImpl *c,
+ bufferlist *pbl, size_t len, uint64_t off,
+ uint64_t snapid, const blkin_trace_info *info)
+{
+ FUNCTRACE();
+ if (len > (size_t) INT_MAX)
+ return -EDOM;
+
+ OID_EVENT_TRACE(oid.name.c_str(), "RADOS_READ_OP_BEGIN");
+ Context *oncomplete = new C_aio_Complete(c);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+ c->is_read = true;
+ c->io = this;
+ c->blp = pbl;
+
+ ZTracer::Trace trace;
+ if (info)
+ trace.init("rados read", &objecter->trace_endpoint, info);
+
+ Objecter::Op *o = objecter->prepare_read_op(
+ oid, oloc,
+ off, len, snapid, pbl, 0,
+ oncomplete, &c->objver, nullptr, 0, &trace);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_read(const object_t oid, AioCompletionImpl *c,
+ char *buf, size_t len, uint64_t off,
+ uint64_t snapid, const blkin_trace_info *info)
+{
+ FUNCTRACE();
+ if (len > (size_t) INT_MAX)
+ return -EDOM;
+
+ OID_EVENT_TRACE(oid.name.c_str(), "RADOS_READ_OP_BEGIN");
+ Context *oncomplete = new C_aio_Complete(c);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+ c->is_read = true;
+ c->io = this;
+ c->bl.clear();
+ c->bl.push_back(buffer::create_static(len, buf));
+ c->blp = &c->bl;
+ c->out_buf = buf;
+
+ ZTracer::Trace trace;
+ if (info)
+ trace.init("rados read", &objecter->trace_endpoint, info);
+
+ Objecter::Op *o = objecter->prepare_read_op(
+ oid, oloc,
+ off, len, snapid, &c->bl, 0,
+ oncomplete, &c->objver, nullptr, 0, &trace);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+class C_ObjectOperation : public Context {
+public:
+ ::ObjectOperation m_ops;
+ explicit C_ObjectOperation(Context *c) : m_ctx(c) {}
+ void finish(int r) override {
+ m_ctx->complete(r);
+ }
+private:
+ Context *m_ctx;
+};
+
+int librados::IoCtxImpl::aio_sparse_read(const object_t oid,
+ AioCompletionImpl *c,
+ std::map<uint64_t,uint64_t> *m,
+ bufferlist *data_bl, size_t len,
+ uint64_t off, uint64_t snapid)
+{
+ FUNCTRACE();
+ if (len > (size_t) INT_MAX)
+ return -EDOM;
+
+ Context *nested = new C_aio_Complete(c);
+ C_ObjectOperation *onack = new C_ObjectOperation(nested);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) nested)->oid = oid;
+#endif
+ c->is_read = true;
+ c->io = this;
+
+ onack->m_ops.sparse_read(off, len, m, data_bl, NULL);
+
+ Objecter::Op *o = objecter->prepare_read_op(
+ oid, oloc,
+ onack->m_ops, snapid, NULL, 0,
+ onack, &c->objver);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_cmpext(const object_t& oid,
+ AioCompletionImpl *c,
+ uint64_t off,
+ bufferlist& cmp_bl)
+{
+ if (cmp_bl.length() > UINT_MAX/2)
+ return -E2BIG;
+
+ Context *onack = new C_aio_Complete(c);
+
+ c->is_read = true;
+ c->io = this;
+
+ Objecter::Op *o = objecter->prepare_cmpext_op(
+ oid, oloc, off, cmp_bl, snap_seq, 0,
+ onack, &c->objver);
+ objecter->op_submit(o, &c->tid);
+
+ return 0;
+}
+
+/* use m_ops.cmpext() + prepare_read_op() for non-bufferlist C API */
+int librados::IoCtxImpl::aio_cmpext(const object_t& oid,
+ AioCompletionImpl *c,
+ const char *cmp_buf,
+ size_t cmp_len,
+ uint64_t off)
+{
+ if (cmp_len > UINT_MAX/2)
+ return -E2BIG;
+
+ bufferlist cmp_bl;
+ cmp_bl.append(cmp_buf, cmp_len);
+
+ Context *nested = new C_aio_Complete(c);
+ C_ObjectOperation *onack = new C_ObjectOperation(nested);
+
+ c->is_read = true;
+ c->io = this;
+
+ onack->m_ops.cmpext(off, cmp_len, cmp_buf, NULL);
+
+ Objecter::Op *o = objecter->prepare_read_op(
+ oid, oloc, onack->m_ops, snap_seq, NULL, 0, onack, &c->objver);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_write(const object_t &oid, AioCompletionImpl *c,
+ const bufferlist& bl, size_t len,
+ uint64_t off, const blkin_trace_info *info)
+{
+ FUNCTRACE();
+ auto ut = ceph::real_clock::now();
+ ldout(client->cct, 20) << "aio_write " << oid << " " << off << "~" << len << " snapc=" << snapc << " snap_seq=" << snap_seq << dendl;
+ OID_EVENT_TRACE(oid.name.c_str(), "RADOS_WRITE_OP_BEGIN");
+
+ if (len > UINT_MAX/2)
+ return -E2BIG;
+ /* can't write to a snapshot */
+ if (snap_seq != CEPH_NOSNAP)
+ return -EROFS;
+
+ Context *oncomplete = new C_aio_Complete(c);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+ ZTracer::Trace trace;
+ if (info)
+ trace.init("rados write", &objecter->trace_endpoint, info);
+
+ c->io = this;
+ queue_aio_write(c);
+
+ Objecter::Op *o = objecter->prepare_write_op(
+ oid, oloc,
+ off, len, snapc, bl, ut, 0,
+ oncomplete, &c->objver, nullptr, 0, &trace);
+ objecter->op_submit(o, &c->tid);
+
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_append(const object_t &oid, AioCompletionImpl *c,
+ const bufferlist& bl, size_t len)
+{
+ FUNCTRACE();
+ auto ut = ceph::real_clock::now();
+
+ if (len > UINT_MAX/2)
+ return -E2BIG;
+ /* can't write to a snapshot */
+ if (snap_seq != CEPH_NOSNAP)
+ return -EROFS;
+
+ Context *oncomplete = new C_aio_Complete(c);
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+
+ c->io = this;
+ queue_aio_write(c);
+
+ Objecter::Op *o = objecter->prepare_append_op(
+ oid, oloc,
+ len, snapc, bl, ut, 0,
+ oncomplete, &c->objver);
+ objecter->op_submit(o, &c->tid);
+
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_write_full(const object_t &oid,
+ AioCompletionImpl *c,
+ const bufferlist& bl)
+{
+ FUNCTRACE();
+ auto ut = ceph::real_clock::now();
+
+ if (bl.length() > UINT_MAX/2)
+ return -E2BIG;
+ /* can't write to a snapshot */
+ if (snap_seq != CEPH_NOSNAP)
+ return -EROFS;
+
+ Context *oncomplete = new C_aio_Complete(c);
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+
+ c->io = this;
+ queue_aio_write(c);
+
+ Objecter::Op *o = objecter->prepare_write_full_op(
+ oid, oloc,
+ snapc, bl, ut, 0,
+ oncomplete, &c->objver);
+ objecter->op_submit(o, &c->tid);
+
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_writesame(const object_t &oid,
+ AioCompletionImpl *c,
+ const bufferlist& bl,
+ size_t write_len,
+ uint64_t off)
+{
+ FUNCTRACE();
+ auto ut = ceph::real_clock::now();
+
+ if ((bl.length() > UINT_MAX/2) || (write_len > UINT_MAX/2))
+ return -E2BIG;
+ if ((bl.length() == 0) || (write_len % bl.length()))
+ return -EINVAL;
+ /* can't write to a snapshot */
+ if (snap_seq != CEPH_NOSNAP)
+ return -EROFS;
+
+ Context *oncomplete = new C_aio_Complete(c);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+ c->io = this;
+ queue_aio_write(c);
+
+ Objecter::Op *o = objecter->prepare_writesame_op(
+ oid, oloc,
+ write_len, off,
+ snapc, bl, ut, 0,
+ oncomplete, &c->objver);
+ objecter->op_submit(o, &c->tid);
+
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_remove(const object_t &oid, AioCompletionImpl *c, int flags)
+{
+ FUNCTRACE();
+ auto ut = ceph::real_clock::now();
+
+ /* can't write to a snapshot */
+ if (snap_seq != CEPH_NOSNAP)
+ return -EROFS;
+
+ Context *oncomplete = new C_aio_Complete(c);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+ c->io = this;
+ queue_aio_write(c);
+
+ Objecter::Op *o = objecter->prepare_remove_op(
+ oid, oloc,
+ snapc, ut, flags,
+ oncomplete, &c->objver);
+ objecter->op_submit(o, &c->tid);
+
+ return 0;
+}
+
+
+int librados::IoCtxImpl::aio_stat(const object_t& oid, AioCompletionImpl *c,
+ uint64_t *psize, time_t *pmtime)
+{
+ C_aio_stat_Ack *onack = new C_aio_stat_Ack(c, pmtime);
+ c->is_read = true;
+ c->io = this;
+ Objecter::Op *o = objecter->prepare_stat_op(
+ oid, oloc,
+ snap_seq, psize, &onack->mtime, 0,
+ onack, &c->objver);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_stat2(const object_t& oid, AioCompletionImpl *c,
+ uint64_t *psize, struct timespec *pts)
+{
+ C_aio_stat2_Ack *onack = new C_aio_stat2_Ack(c, pts);
+ c->is_read = true;
+ c->io = this;
+ Objecter::Op *o = objecter->prepare_stat_op(
+ oid, oloc,
+ snap_seq, psize, &onack->mtime, 0,
+ onack, &c->objver);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_getxattr(const object_t& oid, AioCompletionImpl *c,
+ const char *name, bufferlist& bl)
+{
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.getxattr(name, &bl, NULL);
+ int r = aio_operate_read(oid, &rd, c, 0, &bl);
+ return r;
+}
+
+int librados::IoCtxImpl::aio_rmxattr(const object_t& oid, AioCompletionImpl *c,
+ const char *name)
+{
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.rmxattr(name);
+ return aio_operate(oid, &op, c, snapc, 0);
+}
+
+int librados::IoCtxImpl::aio_setxattr(const object_t& oid, AioCompletionImpl *c,
+ const char *name, bufferlist& bl)
+{
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.setxattr(name, bl);
+ return aio_operate(oid, &op, c, snapc, 0);
+}
+
+namespace {
+struct AioGetxattrsData {
+ AioGetxattrsData(librados::AioCompletionImpl *c, map<string, bufferlist>* attrset,
+ librados::RadosClient *_client) :
+ user_completion(c), user_attrset(attrset), client(_client) {}
+ struct librados::C_AioCompleteAndSafe user_completion;
+ map<string, bufferlist> result_attrset;
+ map<std::string, bufferlist>* user_attrset;
+ librados::RadosClient *client;
+};
+}
+
+static void aio_getxattrs_complete(rados_completion_t c, void *arg) {
+ AioGetxattrsData *cdata = reinterpret_cast<AioGetxattrsData*>(arg);
+ int rc = rados_aio_get_return_value(c);
+ cdata->user_attrset->clear();
+ if (rc >= 0) {
+ for (map<string,bufferlist>::iterator p = cdata->result_attrset.begin();
+ p != cdata->result_attrset.end();
+ ++p) {
+ ldout(cdata->client->cct, 10) << "IoCtxImpl::getxattrs: xattr=" << p->first << dendl;
+ (*cdata->user_attrset)[p->first] = p->second;
+ }
+ }
+ cdata->user_completion.finish(rc);
+ ((librados::AioCompletionImpl*)c)->put();
+ delete cdata;
+}
+
+int librados::IoCtxImpl::aio_getxattrs(const object_t& oid, AioCompletionImpl *c,
+ map<std::string, bufferlist>& attrset)
+{
+ AioGetxattrsData *cdata = new AioGetxattrsData(c, &attrset, client);
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.getxattrs(&cdata->result_attrset, NULL);
+ librados::AioCompletionImpl *comp = new librados::AioCompletionImpl;
+ comp->set_complete_callback(cdata, aio_getxattrs_complete);
+ return aio_operate_read(oid, &rd, comp, 0, NULL);
+}
+
+int librados::IoCtxImpl::aio_cancel(AioCompletionImpl *c)
+{
+ return objecter->op_cancel(c->tid, -ECANCELED);
+}
+
+
+int librados::IoCtxImpl::hit_set_list(uint32_t hash, AioCompletionImpl *c,
+ std::list< std::pair<time_t, time_t> > *pls)
+{
+ Context *oncomplete = new C_aio_Complete(c);
+ c->is_read = true;
+ c->io = this;
+
+ ::ObjectOperation rd;
+ rd.hit_set_ls(pls, NULL);
+ object_locator_t oloc(poolid);
+ Objecter::Op *o = objecter->prepare_pg_read_op(
+ hash, oloc, rd, NULL, 0, oncomplete, NULL, NULL);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::hit_set_get(uint32_t hash, AioCompletionImpl *c,
+ time_t stamp,
+ bufferlist *pbl)
+{
+ Context *oncomplete = new C_aio_Complete(c);
+ c->is_read = true;
+ c->io = this;
+
+ ::ObjectOperation rd;
+ rd.hit_set_get(ceph::real_clock::from_time_t(stamp), pbl, 0);
+ object_locator_t oloc(poolid);
+ Objecter::Op *o = objecter->prepare_pg_read_op(
+ hash, oloc, rd, NULL, 0, oncomplete, NULL, NULL);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::remove(const object_t& oid)
+{
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.remove();
+ return operate(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::remove(const object_t& oid, int flags)
+{
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.remove();
+ return operate(oid, &op, NULL, flags);
+}
+
+int librados::IoCtxImpl::trunc(const object_t& oid, uint64_t size)
+{
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.truncate(size);
+ return operate(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::get_inconsistent_objects(const pg_t& pg,
+ const librados::object_id_t& start_after,
+ uint64_t max_to_get,
+ AioCompletionImpl *c,
+ std::vector<inconsistent_obj_t>* objects,
+ uint32_t* interval)
+{
+ Context *oncomplete = new C_aio_Complete(c);
+ c->is_read = true;
+ c->io = this;
+
+ ::ObjectOperation op;
+ op.scrub_ls(start_after, max_to_get, objects, interval, nullptr);
+ object_locator_t oloc{poolid, pg.ps()};
+ Objecter::Op *o = objecter->prepare_pg_read_op(
+ oloc.hash, oloc, op, nullptr, CEPH_OSD_FLAG_PGOP, oncomplete,
+ nullptr, nullptr);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::get_inconsistent_snapsets(const pg_t& pg,
+ const librados::object_id_t& start_after,
+ uint64_t max_to_get,
+ AioCompletionImpl *c,
+ std::vector<inconsistent_snapset_t>* snapsets,
+ uint32_t* interval)
+{
+ Context *oncomplete = new C_aio_Complete(c);
+ c->is_read = true;
+ c->io = this;
+
+ ::ObjectOperation op;
+ op.scrub_ls(start_after, max_to_get, snapsets, interval, nullptr);
+ object_locator_t oloc{poolid, pg.ps()};
+ Objecter::Op *o = objecter->prepare_pg_read_op(
+ oloc.hash, oloc, op, nullptr, CEPH_OSD_FLAG_PGOP, oncomplete,
+ nullptr, nullptr);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::tmap_update(const object_t& oid, bufferlist& cmdbl)
+{
+ ::ObjectOperation wr;
+ prepare_assert_ops(&wr);
+ wr.tmap_update(cmdbl);
+ return operate(oid, &wr, NULL);
+}
+
+int librados::IoCtxImpl::tmap_put(const object_t& oid, bufferlist& bl)
+{
+ ::ObjectOperation wr;
+ prepare_assert_ops(&wr);
+ wr.tmap_put(bl);
+ return operate(oid, &wr, NULL);
+}
+
+int librados::IoCtxImpl::tmap_get(const object_t& oid, bufferlist& bl)
+{
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.tmap_get(&bl, NULL);
+ return operate_read(oid, &rd, NULL);
+}
+
+int librados::IoCtxImpl::tmap_to_omap(const object_t& oid, bool nullok)
+{
+ ::ObjectOperation wr;
+ prepare_assert_ops(&wr);
+ wr.tmap_to_omap(nullok);
+ return operate(oid, &wr, NULL);
+}
+
+int librados::IoCtxImpl::exec(const object_t& oid,
+ const char *cls, const char *method,
+ bufferlist& inbl, bufferlist& outbl)
+{
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.call(cls, method, inbl);
+ return operate_read(oid, &rd, &outbl);
+}
+
+int librados::IoCtxImpl::aio_exec(const object_t& oid, AioCompletionImpl *c,
+ const char *cls, const char *method,
+ bufferlist& inbl, bufferlist *outbl)
+{
+ FUNCTRACE();
+ Context *oncomplete = new C_aio_Complete(c);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+ c->is_read = true;
+ c->io = this;
+
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.call(cls, method, inbl);
+ Objecter::Op *o = objecter->prepare_read_op(
+ oid, oloc, rd, snap_seq, outbl, 0, oncomplete, &c->objver);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::aio_exec(const object_t& oid, AioCompletionImpl *c,
+ const char *cls, const char *method,
+ bufferlist& inbl, char *buf, size_t out_len)
+{
+ FUNCTRACE();
+ Context *oncomplete = new C_aio_Complete(c);
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ ((C_aio_Complete *) oncomplete)->oid = oid;
+#endif
+ c->is_read = true;
+ c->io = this;
+ c->bl.clear();
+ c->bl.push_back(buffer::create_static(out_len, buf));
+ c->blp = &c->bl;
+ c->out_buf = buf;
+
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.call(cls, method, inbl);
+ Objecter::Op *o = objecter->prepare_read_op(
+ oid, oloc, rd, snap_seq, &c->bl, 0, oncomplete, &c->objver);
+ objecter->op_submit(o, &c->tid);
+ return 0;
+}
+
+int librados::IoCtxImpl::read(const object_t& oid,
+ bufferlist& bl, size_t len, uint64_t off)
+{
+ if (len > (size_t) INT_MAX)
+ return -EDOM;
+ OID_EVENT_TRACE(oid.name.c_str(), "RADOS_READ_OP_BEGIN");
+
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.read(off, len, &bl, NULL, NULL);
+ int r = operate_read(oid, &rd, &bl);
+ if (r < 0)
+ return r;
+
+ if (bl.length() < len) {
+ ldout(client->cct, 10) << "Returned length " << bl.length()
+ << " less than original length "<< len << dendl;
+ }
+
+ return bl.length();
+}
+
+int librados::IoCtxImpl::cmpext(const object_t& oid, uint64_t off,
+ bufferlist& cmp_bl)
+{
+ if (cmp_bl.length() > UINT_MAX/2)
+ return -E2BIG;
+
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.cmpext(off, cmp_bl, NULL);
+ return operate_read(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::mapext(const object_t& oid,
+ uint64_t off, size_t len,
+ std::map<uint64_t,uint64_t>& m)
+{
+ bufferlist bl;
+
+ Mutex mylock("IoCtxImpl::read::mylock");
+ Cond cond;
+ bool done;
+ int r;
+ Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
+
+ objecter->mapext(oid, oloc,
+ off, len, snap_seq, &bl, 0,
+ onack);
+
+ mylock.Lock();
+ while (!done)
+ cond.Wait(mylock);
+ mylock.Unlock();
+ ldout(client->cct, 10) << "Objecter returned from read r=" << r << dendl;
+
+ if (r < 0)
+ return r;
+
+ bufferlist::iterator iter = bl.begin();
+ ::decode(m, iter);
+
+ return m.size();
+}
+
+int librados::IoCtxImpl::sparse_read(const object_t& oid,
+ std::map<uint64_t,uint64_t>& m,
+ bufferlist& data_bl, size_t len,
+ uint64_t off)
+{
+ if (len > (size_t) INT_MAX)
+ return -EDOM;
+
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.sparse_read(off, len, &m, &data_bl, NULL);
+
+ int r = operate_read(oid, &rd, NULL);
+ if (r < 0)
+ return r;
+
+ return m.size();
+}
+
+int librados::IoCtxImpl::checksum(const object_t& oid, uint8_t type,
+ const bufferlist &init_value, size_t len,
+ uint64_t off, size_t chunk_size,
+ bufferlist *pbl)
+{
+ if (len > (size_t) INT_MAX) {
+ return -EDOM;
+ }
+
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.checksum(type, init_value, off, len, chunk_size, pbl, nullptr, nullptr);
+
+ int r = operate_read(oid, &rd, nullptr);
+ if (r < 0) {
+ return r;
+ }
+
+ return 0;
+}
+
+int librados::IoCtxImpl::stat(const object_t& oid, uint64_t *psize, time_t *pmtime)
+{
+ uint64_t size;
+ real_time mtime;
+
+ if (!psize)
+ psize = &size;
+
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.stat(psize, &mtime, NULL);
+ int r = operate_read(oid, &rd, NULL);
+
+ if (r >= 0 && pmtime) {
+ *pmtime = real_clock::to_time_t(mtime);
+ }
+
+ return r;
+}
+
+int librados::IoCtxImpl::stat2(const object_t& oid, uint64_t *psize, struct timespec *pts)
+{
+ uint64_t size;
+ ceph::real_time mtime;
+
+ if (!psize)
+ psize = &size;
+
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.stat(psize, &mtime, NULL);
+ int r = operate_read(oid, &rd, NULL);
+ if (r < 0) {
+ return r;
+ }
+
+ if (pts) {
+ *pts = ceph::real_clock::to_timespec(mtime);
+ }
+
+ return 0;
+}
+
+int librados::IoCtxImpl::getxattr(const object_t& oid,
+ const char *name, bufferlist& bl)
+{
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.getxattr(name, &bl, NULL);
+ int r = operate_read(oid, &rd, &bl);
+ if (r < 0)
+ return r;
+
+ return bl.length();
+}
+
+int librados::IoCtxImpl::rmxattr(const object_t& oid, const char *name)
+{
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.rmxattr(name);
+ return operate(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::setxattr(const object_t& oid,
+ const char *name, bufferlist& bl)
+{
+ ::ObjectOperation op;
+ prepare_assert_ops(&op);
+ op.setxattr(name, bl);
+ return operate(oid, &op, NULL);
+}
+
+int librados::IoCtxImpl::getxattrs(const object_t& oid,
+ map<std::string, bufferlist>& attrset)
+{
+ map<string, bufferlist> aset;
+
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.getxattrs(&aset, NULL);
+ int r = operate_read(oid, &rd, NULL);
+
+ attrset.clear();
+ if (r >= 0) {
+ for (map<string,bufferlist>::iterator p = aset.begin(); p != aset.end(); ++p) {
+ ldout(client->cct, 10) << "IoCtxImpl::getxattrs: xattr=" << p->first << dendl;
+ attrset[p->first.c_str()] = p->second;
+ }
+ }
+
+ return r;
+}
+
+void librados::IoCtxImpl::set_sync_op_version(version_t ver)
+{
+ ANNOTATE_BENIGN_RACE_SIZED(&last_objver, sizeof(last_objver),
+ "IoCtxImpl last_objver");
+ last_objver = ver;
+}
+
+struct WatchInfo : public Objecter::WatchContext {
+ librados::IoCtxImpl *ioctx;
+ object_t oid;
+ librados::WatchCtx *ctx;
+ librados::WatchCtx2 *ctx2;
+ bool internal = false;
+
+ WatchInfo(librados::IoCtxImpl *io, object_t o,
+ librados::WatchCtx *c, librados::WatchCtx2 *c2,
+ bool inter)
+ : ioctx(io), oid(o), ctx(c), ctx2(c2), internal(inter) {
+ ioctx->get();
+ }
+ ~WatchInfo() override {
+ ioctx->put();
+ if (internal) {
+ delete ctx;
+ delete ctx2;
+ }
+ }
+
+ void handle_notify(uint64_t notify_id,
+ uint64_t cookie,
+ uint64_t notifier_id,
+ bufferlist& bl) override {
+ ldout(ioctx->client->cct, 10) << __func__ << " " << notify_id
+ << " cookie " << cookie
+ << " notifier_id " << notifier_id
+ << " len " << bl.length()
+ << dendl;
+
+ if (ctx2)
+ ctx2->handle_notify(notify_id, cookie, notifier_id, bl);
+ if (ctx) {
+ ctx->notify(0, 0, bl);
+
+ // send ACK back to OSD if using legacy protocol
+ bufferlist empty;
+ ioctx->notify_ack(oid, notify_id, cookie, empty);
+ }
+ }
+ void handle_error(uint64_t cookie, int err) override {
+ ldout(ioctx->client->cct, 10) << __func__ << " cookie " << cookie
+ << " err " << err
+ << dendl;
+ if (ctx2)
+ ctx2->handle_error(cookie, err);
+ }
+};
+
+int librados::IoCtxImpl::watch(const object_t& oid, uint64_t *handle,
+ librados::WatchCtx *ctx,
+ librados::WatchCtx2 *ctx2,
+ bool internal)
+{
+ return watch(oid, handle, ctx, ctx2, 0, internal);
+}
+
+int librados::IoCtxImpl::watch(const object_t& oid, uint64_t *handle,
+ librados::WatchCtx *ctx,
+ librados::WatchCtx2 *ctx2,
+ uint32_t timeout,
+ bool internal)
+{
+ ::ObjectOperation wr;
+ version_t objver;
+ C_SaferCond onfinish;
+
+ Objecter::LingerOp *linger_op = objecter->linger_register(oid, oloc, 0);
+ *handle = linger_op->get_cookie();
+ linger_op->watch_context = new WatchInfo(this,
+ oid, ctx, ctx2, internal);
+
+ prepare_assert_ops(&wr);
+ wr.watch(*handle, CEPH_OSD_WATCH_OP_WATCH, timeout);
+ bufferlist bl;
+ objecter->linger_watch(linger_op, wr,
+ snapc, ceph::real_clock::now(), bl,
+ &onfinish,
+ &objver);
+
+ int r = onfinish.wait();
+
+ set_sync_op_version(objver);
+
+ if (r < 0) {
+ objecter->linger_cancel(linger_op);
+ *handle = 0;
+ }
+
+ return r;
+}
+
+int librados::IoCtxImpl::aio_watch(const object_t& oid,
+ AioCompletionImpl *c,
+ uint64_t *handle,
+ librados::WatchCtx *ctx,
+ librados::WatchCtx2 *ctx2,
+ bool internal) {
+ return aio_watch(oid, c, handle, ctx, ctx2, 0, internal);
+}
+
+int librados::IoCtxImpl::aio_watch(const object_t& oid,
+ AioCompletionImpl *c,
+ uint64_t *handle,
+ librados::WatchCtx *ctx,
+ librados::WatchCtx2 *ctx2,
+ uint32_t timeout,
+ bool internal)
+{
+ Objecter::LingerOp *linger_op = objecter->linger_register(oid, oloc, 0);
+ c->io = this;
+ Context *oncomplete = new C_aio_linger_Complete(c, linger_op, false);
+
+ ::ObjectOperation wr;
+ *handle = linger_op->get_cookie();
+ linger_op->watch_context = new WatchInfo(this, oid, ctx, ctx2, internal);
+
+ prepare_assert_ops(&wr);
+ wr.watch(*handle, CEPH_OSD_WATCH_OP_WATCH, timeout);
+ bufferlist bl;
+ objecter->linger_watch(linger_op, wr,
+ snapc, ceph::real_clock::now(), bl,
+ oncomplete, &c->objver);
+
+ return 0;
+}
+
+
+int librados::IoCtxImpl::notify_ack(
+ const object_t& oid,
+ uint64_t notify_id,
+ uint64_t cookie,
+ bufferlist& bl)
+{
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ rd.notify_ack(notify_id, cookie, bl);
+ objecter->read(oid, oloc, rd, snap_seq, (bufferlist*)NULL, 0, 0, 0);
+ return 0;
+}
+
+int librados::IoCtxImpl::watch_check(uint64_t cookie)
+{
+ Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
+ return objecter->linger_check(linger_op);
+}
+
+int librados::IoCtxImpl::unwatch(uint64_t cookie)
+{
+ Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
+ C_SaferCond onfinish;
+ version_t ver = 0;
+
+ ::ObjectOperation wr;
+ prepare_assert_ops(&wr);
+ wr.watch(cookie, CEPH_OSD_WATCH_OP_UNWATCH);
+ objecter->mutate(linger_op->target.base_oid, oloc, wr,
+ snapc, ceph::real_clock::now(), 0,
+ &onfinish, &ver);
+ objecter->linger_cancel(linger_op);
+
+ int r = onfinish.wait();
+ set_sync_op_version(ver);
+ return r;
+}
+
+int librados::IoCtxImpl::aio_unwatch(uint64_t cookie, AioCompletionImpl *c)
+{
+ c->io = this;
+ Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
+ Context *oncomplete = new C_aio_linger_Complete(c, linger_op, true);
+
+ ::ObjectOperation wr;
+ prepare_assert_ops(&wr);
+ wr.watch(cookie, CEPH_OSD_WATCH_OP_UNWATCH);
+ objecter->mutate(linger_op->target.base_oid, oloc, wr,
+ snapc, ceph::real_clock::now(), 0,
+ oncomplete, &c->objver);
+ return 0;
+}
+
+int librados::IoCtxImpl::notify(const object_t& oid, bufferlist& bl,
+ uint64_t timeout_ms,
+ bufferlist *preply_bl,
+ char **preply_buf, size_t *preply_buf_len)
+{
+ Objecter::LingerOp *linger_op = objecter->linger_register(oid, oloc, 0);
+
+ C_SaferCond notify_finish_cond;
+ Context *notify_finish = new C_notify_Finish(client->cct, ¬ify_finish_cond,
+ objecter, linger_op, preply_bl,
+ preply_buf, preply_buf_len);
+
+ uint32_t timeout = notify_timeout;
+ if (timeout_ms)
+ timeout = timeout_ms / 1000;
+
+ // Construct RADOS op
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ bufferlist inbl;
+ rd.notify(linger_op->get_cookie(), 1, timeout, bl, &inbl);
+
+ // Issue RADOS op
+ C_SaferCond onack;
+ version_t objver;
+ objecter->linger_notify(linger_op,
+ rd, snap_seq, inbl, NULL,
+ &onack, &objver);
+
+ ldout(client->cct, 10) << __func__ << " issued linger op " << linger_op << dendl;
+ int r = onack.wait();
+ ldout(client->cct, 10) << __func__ << " linger op " << linger_op
+ << " acked (" << r << ")" << dendl;
+
+ if (r == 0) {
+ ldout(client->cct, 10) << __func__ << " waiting for watch_notify finish "
+ << linger_op << dendl;
+ r = notify_finish_cond.wait();
+
+ } else {
+ ldout(client->cct, 10) << __func__ << " failed to initiate notify, r = "
+ << r << dendl;
+ notify_finish->complete(r);
+ }
+
+ objecter->linger_cancel(linger_op);
+
+ set_sync_op_version(objver);
+ return r;
+}
+
+int librados::IoCtxImpl::aio_notify(const object_t& oid, AioCompletionImpl *c,
+ bufferlist& bl, uint64_t timeout_ms,
+ bufferlist *preply_bl, char **preply_buf,
+ size_t *preply_buf_len)
+{
+ Objecter::LingerOp *linger_op = objecter->linger_register(oid, oloc, 0);
+
+ c->io = this;
+
+ C_aio_notify_Complete *oncomplete = new C_aio_notify_Complete(c, linger_op);
+ C_notify_Finish *onnotify = new C_notify_Finish(client->cct, oncomplete,
+ objecter, linger_op,
+ preply_bl, preply_buf,
+ preply_buf_len);
+ Context *onack = new C_aio_notify_Ack(client->cct, onnotify, oncomplete);
+
+ uint32_t timeout = notify_timeout;
+ if (timeout_ms)
+ timeout = timeout_ms / 1000;
+
+ // Construct RADOS op
+ ::ObjectOperation rd;
+ prepare_assert_ops(&rd);
+ bufferlist inbl;
+ rd.notify(linger_op->get_cookie(), 1, timeout, bl, &inbl);
+
+ // Issue RADOS op
+ objecter->linger_notify(linger_op,
+ rd, snap_seq, inbl, NULL,
+ onack, &c->objver);
+ return 0;
+}
+
+int librados::IoCtxImpl::set_alloc_hint(const object_t& oid,
+ uint64_t expected_object_size,
+ uint64_t expected_write_size,
+ uint32_t flags)
+{
+ ::ObjectOperation wr;
+ prepare_assert_ops(&wr);
+ wr.set_alloc_hint(expected_object_size, expected_write_size, flags);
+ return operate(oid, &wr, NULL);
+}
+
+version_t librados::IoCtxImpl::last_version()
+{
+ return last_objver;
+}
+
+void librados::IoCtxImpl::set_assert_version(uint64_t ver)
+{
+ assert_ver = ver;
+}
+
+void librados::IoCtxImpl::set_notify_timeout(uint32_t timeout)
+{
+ notify_timeout = timeout;
+}
+
+int librados::IoCtxImpl::cache_pin(const object_t& oid)
+{
+ ::ObjectOperation wr;
+ prepare_assert_ops(&wr);
+ wr.cache_pin();
+ return operate(oid, &wr, NULL);
+}
+
+int librados::IoCtxImpl::cache_unpin(const object_t& oid)
+{
+ ::ObjectOperation wr;
+ prepare_assert_ops(&wr);
+ wr.cache_unpin();
+ return operate(oid, &wr, NULL);
+}
+
+
+///////////////////////////// C_aio_stat_Ack ////////////////////////////
+
+librados::IoCtxImpl::C_aio_stat_Ack::C_aio_stat_Ack(AioCompletionImpl *_c,
+ time_t *pm)
+ : c(_c), pmtime(pm)
+{
+ assert(!c->io);
+ c->get();
+}
+
+void librados::IoCtxImpl::C_aio_stat_Ack::finish(int r)
+{
+ c->lock.Lock();
+ c->rval = r;
+ c->complete = true;
+ c->cond.Signal();
+
+ if (r >= 0 && pmtime) {
+ *pmtime = real_clock::to_time_t(mtime);
+ }
+
+ if (c->callback_complete) {
+ c->io->client->finisher.queue(new C_AioComplete(c));
+ }
+
+ c->put_unlock();
+}
+
+///////////////////////////// C_aio_stat2_Ack ////////////////////////////
+
+librados::IoCtxImpl::C_aio_stat2_Ack::C_aio_stat2_Ack(AioCompletionImpl *_c,
+ struct timespec *pt)
+ : c(_c), pts(pt)
+{
+ assert(!c->io);
+ c->get();
+}
+
+void librados::IoCtxImpl::C_aio_stat2_Ack::finish(int r)
+{
+ c->lock.Lock();
+ c->rval = r;
+ c->complete = true;
+ c->cond.Signal();
+
+ if (r >= 0 && pts) {
+ *pts = real_clock::to_timespec(mtime);
+ }
+
+ if (c->callback_complete) {
+ c->io->client->finisher.queue(new C_AioComplete(c));
+ }
+
+ c->put_unlock();
+}
+
+//////////////////////////// C_aio_Complete ////////////////////////////////
+
+librados::IoCtxImpl::C_aio_Complete::C_aio_Complete(AioCompletionImpl *_c)
+ : c(_c)
+{
+ c->get();
+}
+
+void librados::IoCtxImpl::C_aio_Complete::finish(int r)
+{
+ c->lock.Lock();
+ c->rval = r;
+ c->complete = true;
+ c->cond.Signal();
+
+ if (r == 0 && c->blp && c->blp->length() > 0) {
+ if (c->out_buf && !c->blp->is_provided_buffer(c->out_buf))
+ c->blp->copy(0, c->blp->length(), c->out_buf);
+ c->rval = c->blp->length();
+ }
+
+ if (c->callback_complete ||
+ c->callback_safe) {
+ c->io->client->finisher.queue(new C_AioComplete(c));
+ }
+
+ if (c->aio_write_seq) {
+ c->io->complete_aio_write(c);
+ }
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ OID_EVENT_TRACE(oid.name.c_str(), "RADOS_OP_COMPLETE");
+#endif
+ c->put_unlock();
+}
+
+void librados::IoCtxImpl::object_list_slice(
+ const hobject_t start,
+ const hobject_t finish,
+ const size_t n,
+ const size_t m,
+ hobject_t *split_start,
+ hobject_t *split_finish)
+{
+ if (start.is_max()) {
+ *split_start = hobject_t::get_max();
+ *split_finish = hobject_t::get_max();
+ return;
+ }
+
+ uint64_t start_hash = hobject_t::_reverse_bits(start.get_hash());
+ uint64_t finish_hash =
+ finish.is_max() ? 0x100000000 :
+ hobject_t::_reverse_bits(finish.get_hash());
+
+ uint64_t diff = finish_hash - start_hash;
+ uint64_t rev_start = start_hash + (diff * n / m);
+ uint64_t rev_finish = start_hash + (diff * (n + 1) / m);
+ if (n == 0) {
+ *split_start = start;
+ } else {
+ *split_start = hobject_t(
+ object_t(), string(), CEPH_NOSNAP,
+ hobject_t::_reverse_bits(rev_start), poolid, string());
+ }
+
+ if (n == m - 1)
+ *split_finish = finish;
+ else if (rev_finish >= 0x100000000)
+ *split_finish = hobject_t::get_max();
+ else
+ *split_finish = hobject_t(
+ object_t(), string(), CEPH_NOSNAP,
+ hobject_t::_reverse_bits(rev_finish), poolid, string());
+}
+
+int librados::IoCtxImpl::application_enable(const std::string& app_name,
+ bool force)
+{
+ auto c = new PoolAsyncCompletionImpl();
+ application_enable_async(app_name, force, c);
+
+ int r = c->wait();
+ assert(r == 0);
+
+ r = c->get_return_value();
+ c->release();
+ if (r < 0) {
+ return r;
+ }
+
+ return client->wait_for_latest_osdmap();
+}
+
+void librados::IoCtxImpl::application_enable_async(const std::string& app_name,
+ bool force,
+ PoolAsyncCompletionImpl *c)
+{
+ // pre-Luminous clusters will return -EINVAL and application won't be
+ // preserved until Luminous is configured as minimim version.
+ if (!client->get_required_monitor_features().contains_all(
+ ceph::features::mon::FEATURE_LUMINOUS)) {
+ client->finisher.queue(new C_PoolAsync_Safe(c), -EOPNOTSUPP);
+ return;
+ }
+
+ std::stringstream cmd;
+ cmd << "{"
+ << "\"prefix\": \"osd pool application enable\","
+ << "\"pool\": \"" << get_cached_pool_name() << "\","
+ << "\"app\": \"" << app_name << "\"";
+ if (force) {
+ cmd << ",\"force\":\"--yes-i-really-mean-it\"";
+ }
+ cmd << "}";
+
+ std::vector<std::string> cmds;
+ cmds.push_back(cmd.str());
+ bufferlist inbl;
+ client->mon_command_async(cmds, inbl, nullptr, nullptr,
+ new C_PoolAsync_Safe(c));
+}
+
+int librados::IoCtxImpl::application_list(std::set<std::string> *app_names)
+{
+ int r = 0;
+ app_names->clear();
+ objecter->with_osdmap([&](const OSDMap& o) {
+ auto pg_pool = o.get_pg_pool(poolid);
+ if (pg_pool == nullptr) {
+ r = -ENOENT;
+ return;
+ }
+
+ for (auto &pair : pg_pool->application_metadata) {
+ app_names->insert(pair.first);
+ }
+ });
+ return r;
+}
+
+int librados::IoCtxImpl::application_metadata_get(const std::string& app_name,
+ const std::string &key,
+ std::string* value)
+{
+ int r = 0;
+ objecter->with_osdmap([&](const OSDMap& o) {
+ auto pg_pool = o.get_pg_pool(poolid);
+ if (pg_pool == nullptr) {
+ r = -ENOENT;
+ return;
+ }
+
+ auto app_it = pg_pool->application_metadata.find(app_name);
+ if (app_it == pg_pool->application_metadata.end()) {
+ r = -ENOENT;
+ return;
+ }
+
+ auto it = app_it->second.find(key);
+ if (it == app_it->second.end()) {
+ r = -ENOENT;
+ return;
+ }
+
+ *value = it->second;
+ });
+ return r;
+}
+
+int librados::IoCtxImpl::application_metadata_set(const std::string& app_name,
+ const std::string &key,
+ const std::string& value)
+{
+ std::stringstream cmd;
+ cmd << "{"
+ << "\"prefix\":\"osd pool application set\","
+ << "\"pool\":\"" << get_cached_pool_name() << "\","
+ << "\"app\":\"" << app_name << "\","
+ << "\"key\":\"" << key << "\","
+ << "\"value\":\"" << value << "\""
+ << "}";
+
+ std::vector<std::string> cmds;
+ cmds.push_back(cmd.str());
+ bufferlist inbl;
+ int r = client->mon_command(cmds, inbl, nullptr, nullptr);
+ if (r < 0) {
+ return r;
+ }
+
+ // ensure we have the latest osd map epoch before proceeding
+ return client->wait_for_latest_osdmap();
+}
+
+int librados::IoCtxImpl::application_metadata_remove(const std::string& app_name,
+ const std::string &key)
+{
+ std::stringstream cmd;
+ cmd << "{"
+ << "\"prefix\":\"osd pool application rm\","
+ << "\"pool\":\"" << get_cached_pool_name() << "\","
+ << "\"app\":\"" << app_name << "\","
+ << "\"key\":\"" << key << "\""
+ << "}";
+
+ std::vector<std::string> cmds;
+ cmds.push_back(cmd.str());
+ bufferlist inbl;
+ int r = client->mon_command(cmds, inbl, nullptr, nullptr);
+ if (r < 0) {
+ return r;
+ }
+
+ // ensure we have the latest osd map epoch before proceeding
+ return client->wait_for_latest_osdmap();
+}
+
+int librados::IoCtxImpl::application_metadata_list(const std::string& app_name,
+ std::map<std::string, std::string> *values)
+{
+ int r = 0;
+ values->clear();
+ objecter->with_osdmap([&](const OSDMap& o) {
+ auto pg_pool = o.get_pg_pool(poolid);
+ if (pg_pool == nullptr) {
+ r = -ENOENT;
+ return;
+ }
+
+ auto it = pg_pool->application_metadata.find(app_name);
+ if (it == pg_pool->application_metadata.end()) {
+ r = -ENOENT;
+ return;
+ }
+
+ *values = it->second;
+ });
+ return r;
+}
+
Code Review / stor4nfv.git / blobdiff