]> git.proxmox.com Git - ceph.git/blame - ceph/src/librados/IoCtxImpl.cc
bump version to 12.0.3-pve3
[ceph.git] / ceph / src / librados / IoCtxImpl.cc
CommitLineData
7c673cae
FG
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 "IoCtxImpl.h"
18
19#include "librados/AioCompletionImpl.h"
20#include "librados/PoolAsyncCompletionImpl.h"
21#include "librados/RadosClient.h"
22#include "include/assert.h"
23#include "common/valgrind.h"
24#include "common/EventTrace.h"
25
26#define dout_subsys ceph_subsys_rados
27#undef dout_prefix
28#define dout_prefix *_dout << "librados: "
29
30namespace librados {
31namespace {
32
33struct C_notify_Finish : public Context {
34 CephContext *cct;
35 Context *ctx;
36 Objecter *objecter;
37 Objecter::LingerOp *linger_op;
38 bufferlist reply_bl;
39 bufferlist *preply_bl;
40 char **preply_buf;
41 size_t *preply_buf_len;
42
43 C_notify_Finish(CephContext *_cct, Context *_ctx, Objecter *_objecter,
44 Objecter::LingerOp *_linger_op, bufferlist *_preply_bl,
45 char **_preply_buf, size_t *_preply_buf_len)
46 : cct(_cct), ctx(_ctx), objecter(_objecter), linger_op(_linger_op),
47 preply_bl(_preply_bl), preply_buf(_preply_buf),
48 preply_buf_len(_preply_buf_len)
49 {
50 linger_op->on_notify_finish = this;
51 linger_op->notify_result_bl = &reply_bl;
52 }
53
54 void finish(int r) override
55 {
56 ldout(cct, 10) << __func__ << " completed notify (linger op "
57 << linger_op << "), r = " << r << dendl;
58
59 // pass result back to user
60 // NOTE: we do this regardless of what error code we return
61 if (preply_buf) {
62 if (reply_bl.length()) {
63 *preply_buf = (char*)malloc(reply_bl.length());
64 memcpy(*preply_buf, reply_bl.c_str(), reply_bl.length());
65 } else {
66 *preply_buf = NULL;
67 }
68 }
69 if (preply_buf_len)
70 *preply_buf_len = reply_bl.length();
71 if (preply_bl)
72 preply_bl->claim(reply_bl);
73
74 ctx->complete(r);
75 }
76};
77
78struct C_aio_linger_cancel : public Context {
79 Objecter *objecter;
80 Objecter::LingerOp *linger_op;
81
82 C_aio_linger_cancel(Objecter *_objecter, Objecter::LingerOp *_linger_op)
83 : objecter(_objecter), linger_op(_linger_op)
84 {
85 }
86
87 void finish(int r) override
88 {
89 objecter->linger_cancel(linger_op);
90 }
91};
92
93struct C_aio_linger_Complete : public Context {
94 AioCompletionImpl *c;
95 Objecter::LingerOp *linger_op;
96 bool cancel;
97
98 C_aio_linger_Complete(AioCompletionImpl *_c, Objecter::LingerOp *_linger_op, bool _cancel)
99 : c(_c), linger_op(_linger_op), cancel(_cancel)
100 {
101 c->get();
102 }
103
104 void finish(int r) override {
105 if (cancel || r < 0)
106 c->io->client->finisher.queue(new C_aio_linger_cancel(c->io->objecter,
107 linger_op));
108
109 c->lock.Lock();
110 c->rval = r;
111 c->complete = true;
112 c->cond.Signal();
113
114 if (c->callback_complete ||
115 c->callback_safe) {
116 c->io->client->finisher.queue(new C_AioComplete(c));
117 }
118 c->put_unlock();
119 }
120};
121
122struct C_aio_notify_Complete : public C_aio_linger_Complete {
123 Mutex lock;
124 bool acked = false;
125 bool finished = false;
126 int ret_val = 0;
127
128 C_aio_notify_Complete(AioCompletionImpl *_c, Objecter::LingerOp *_linger_op)
129 : C_aio_linger_Complete(_c, _linger_op, false),
130 lock("C_aio_notify_Complete::lock") {
131 }
132
133 void handle_ack(int r) {
134 // invoked by C_aio_notify_Ack
135 lock.Lock();
136 acked = true;
137 complete_unlock(r);
138 }
139
140 void complete(int r) override {
141 // invoked by C_notify_Finish (or C_aio_notify_Ack on failure)
142 lock.Lock();
143 finished = true;
144 complete_unlock(r);
145 }
146
147 void complete_unlock(int r) {
148 if (ret_val == 0 && r < 0) {
149 ret_val = r;
150 }
151
152 if (acked && finished) {
153 lock.Unlock();
154 cancel = true;
155 C_aio_linger_Complete::complete(ret_val);
156 } else {
157 lock.Unlock();
158 }
159 }
160};
161
162struct C_aio_notify_Ack : public Context {
163 CephContext *cct;
164 C_notify_Finish *onfinish;
165 C_aio_notify_Complete *oncomplete;
166
167 C_aio_notify_Ack(CephContext *_cct, C_notify_Finish *_onfinish,
168 C_aio_notify_Complete *_oncomplete)
169 : cct(_cct), onfinish(_onfinish), oncomplete(_oncomplete)
170 {
171 }
172
173 void finish(int r) override
174 {
175 ldout(cct, 10) << __func__ << " linger op " << oncomplete->linger_op << " "
176 << "acked (" << r << ")" << dendl;
177 oncomplete->handle_ack(r);
178 if (r < 0) {
179 // on failure, we won't expect to see a notify_finish callback
180 onfinish->complete(r);
181 }
182 }
183};
184
185struct C_aio_selfmanaged_snap_op_Complete : public Context {
186 librados::RadosClient *client;
187 librados::AioCompletionImpl *c;
188
189 C_aio_selfmanaged_snap_op_Complete(librados::RadosClient *client,
190 librados::AioCompletionImpl *c)
191 : client(client), c(c) {
192 c->get();
193 }
194
195 void finish(int r) override {
196 c->lock.Lock();
197 c->rval = r;
198 c->complete = true;
199 c->cond.Signal();
200
201 if (c->callback_complete || c->callback_safe) {
202 client->finisher.queue(new librados::C_AioComplete(c));
203 }
204 c->put_unlock();
205 }
206};
207
208struct C_aio_selfmanaged_snap_create_Complete : public C_aio_selfmanaged_snap_op_Complete {
209 snapid_t snapid;
210 uint64_t *dest_snapid;
211
212 C_aio_selfmanaged_snap_create_Complete(librados::RadosClient *client,
213 librados::AioCompletionImpl *c,
214 uint64_t *dest_snapid)
215 : C_aio_selfmanaged_snap_op_Complete(client, c),
216 dest_snapid(dest_snapid) {
217 }
218
219 void finish(int r) override {
220 if (r >= 0) {
221 *dest_snapid = snapid;
222 }
223 C_aio_selfmanaged_snap_op_Complete::finish(r);
224 }
225};
226
227} // anonymous namespace
228} // namespace librados
229
230librados::IoCtxImpl::IoCtxImpl() :
231 ref_cnt(0), client(NULL), poolid(0), assert_ver(0), last_objver(0),
232 notify_timeout(30), aio_write_list_lock("librados::IoCtxImpl::aio_write_list_lock"),
233 aio_write_seq(0), objecter(NULL)
234{
235}
236
237librados::IoCtxImpl::IoCtxImpl(RadosClient *c, Objecter *objecter,
238 int64_t poolid, snapid_t s)
239 : ref_cnt(0), client(c), poolid(poolid), snap_seq(s),
240 assert_ver(0), last_objver(0),
241 notify_timeout(c->cct->_conf->client_notify_timeout),
242 oloc(poolid), aio_write_list_lock("librados::IoCtxImpl::aio_write_list_lock"),
243 aio_write_seq(0), objecter(objecter)
244{
245}
246
247void librados::IoCtxImpl::set_snap_read(snapid_t s)
248{
249 if (!s)
250 s = CEPH_NOSNAP;
251 ldout(client->cct, 10) << "set snap read " << snap_seq << " -> " << s << dendl;
252 snap_seq = s;
253}
254
255int librados::IoCtxImpl::set_snap_write_context(snapid_t seq, vector<snapid_t>& snaps)
256{
257 ::SnapContext n;
258 ldout(client->cct, 10) << "set snap write context: seq = " << seq
259 << " and snaps = " << snaps << dendl;
260 n.seq = seq;
261 n.snaps = snaps;
262 if (!n.is_valid())
263 return -EINVAL;
264 snapc = n;
265 return 0;
266}
267
268int librados::IoCtxImpl::get_object_hash_position(
269 const std::string& oid, uint32_t *hash_position)
270{
271 int64_t r = objecter->get_object_hash_position(poolid, oid, oloc.nspace);
272 if (r < 0)
273 return r;
274 *hash_position = (uint32_t)r;
275 return 0;
276}
277
278int librados::IoCtxImpl::get_object_pg_hash_position(
279 const std::string& oid, uint32_t *pg_hash_position)
280{
281 int64_t r = objecter->get_object_pg_hash_position(poolid, oid, oloc.nspace);
282 if (r < 0)
283 return r;
284 *pg_hash_position = (uint32_t)r;
285 return 0;
286}
287
288void librados::IoCtxImpl::queue_aio_write(AioCompletionImpl *c)
289{
290 get();
291 aio_write_list_lock.Lock();
292 assert(c->io == this);
293 c->aio_write_seq = ++aio_write_seq;
294 ldout(client->cct, 20) << "queue_aio_write " << this << " completion " << c
295 << " write_seq " << aio_write_seq << dendl;
296 aio_write_list.push_back(&c->aio_write_list_item);
297 aio_write_list_lock.Unlock();
298}
299
300void librados::IoCtxImpl::complete_aio_write(AioCompletionImpl *c)
301{
302 ldout(client->cct, 20) << "complete_aio_write " << c << dendl;
303 aio_write_list_lock.Lock();
304 assert(c->io == this);
305 c->aio_write_list_item.remove_myself();
306
307 map<ceph_tid_t, std::list<AioCompletionImpl*> >::iterator waiters = aio_write_waiters.begin();
308 while (waiters != aio_write_waiters.end()) {
309 if (!aio_write_list.empty() &&
310 aio_write_list.front()->aio_write_seq <= waiters->first) {
311 ldout(client->cct, 20) << " next outstanding write is " << aio_write_list.front()->aio_write_seq
312 << " <= waiter " << waiters->first
313 << ", stopping" << dendl;
314 break;
315 }
316 ldout(client->cct, 20) << " waking waiters on seq " << waiters->first << dendl;
317 for (std::list<AioCompletionImpl*>::iterator it = waiters->second.begin();
318 it != waiters->second.end(); ++it) {
319 client->finisher.queue(new C_AioCompleteAndSafe(*it));
320 (*it)->put();
321 }
322 aio_write_waiters.erase(waiters++);
323 }
324
325 aio_write_cond.Signal();
326 aio_write_list_lock.Unlock();
327 put();
328}
329
330void librados::IoCtxImpl::flush_aio_writes_async(AioCompletionImpl *c)
331{
332 ldout(client->cct, 20) << "flush_aio_writes_async " << this
333 << " completion " << c << dendl;
334 Mutex::Locker l(aio_write_list_lock);
335 ceph_tid_t seq = aio_write_seq;
336 if (aio_write_list.empty()) {
337 ldout(client->cct, 20) << "flush_aio_writes_async no writes. (tid "
338 << seq << ")" << dendl;
339 client->finisher.queue(new C_AioCompleteAndSafe(c));
340 } else {
341 ldout(client->cct, 20) << "flush_aio_writes_async " << aio_write_list.size()
342 << " writes in flight; waiting on tid " << seq << dendl;
343 c->get();
344 aio_write_waiters[seq].push_back(c);
345 }
346}
347
348void librados::IoCtxImpl::flush_aio_writes()
349{
350 ldout(client->cct, 20) << "flush_aio_writes" << dendl;
351 aio_write_list_lock.Lock();
352 ceph_tid_t seq = aio_write_seq;
353 while (!aio_write_list.empty() &&
354 aio_write_list.front()->aio_write_seq <= seq)
355 aio_write_cond.Wait(aio_write_list_lock);
356 aio_write_list_lock.Unlock();
357}
358
359string librados::IoCtxImpl::get_cached_pool_name()
360{
361 std::string pn;
362 client->pool_get_name(get_id(), &pn);
363 return pn;
364}
365
366// SNAPS
367
368int librados::IoCtxImpl::snap_create(const char *snapName)
369{
370 int reply;
371 string sName(snapName);
372
373 Mutex mylock ("IoCtxImpl::snap_create::mylock");
374 Cond cond;
375 bool done;
376 Context *onfinish = new C_SafeCond(&mylock, &cond, &done, &reply);
377 reply = objecter->create_pool_snap(poolid, sName, onfinish);
378
379 if (reply < 0) {
380 delete onfinish;
381 } else {
382 mylock.Lock();
383 while (!done)
384 cond.Wait(mylock);
385 mylock.Unlock();
386 }
387 return reply;
388}
389
390int librados::IoCtxImpl::selfmanaged_snap_create(uint64_t *psnapid)
391{
392 int reply;
393
394 Mutex mylock("IoCtxImpl::selfmanaged_snap_create::mylock");
395 Cond cond;
396 bool done;
397 Context *onfinish = new C_SafeCond(&mylock, &cond, &done, &reply);
398 snapid_t snapid;
399 reply = objecter->allocate_selfmanaged_snap(poolid, &snapid, onfinish);
400
401 if (reply < 0) {
402 delete onfinish;
403 } else {
404 mylock.Lock();
405 while (!done)
406 cond.Wait(mylock);
407 mylock.Unlock();
408 if (reply == 0)
409 *psnapid = snapid;
410 }
411 return reply;
412}
413
414void librados::IoCtxImpl::aio_selfmanaged_snap_create(uint64_t *snapid,
415 AioCompletionImpl *c)
416{
417 C_aio_selfmanaged_snap_create_Complete *onfinish =
418 new C_aio_selfmanaged_snap_create_Complete(client, c, snapid);
419 int r = objecter->allocate_selfmanaged_snap(poolid, &onfinish->snapid,
420 onfinish);
421 if (r < 0) {
422 onfinish->complete(r);
423 }
424}
425
426int librados::IoCtxImpl::snap_remove(const char *snapName)
427{
428 int reply;
429 string sName(snapName);
430
431 Mutex mylock ("IoCtxImpl::snap_remove::mylock");
432 Cond cond;
433 bool done;
434 Context *onfinish = new C_SafeCond(&mylock, &cond, &done, &reply);
435 reply = objecter->delete_pool_snap(poolid, sName, onfinish);
436
437 if (reply < 0) {
438 delete onfinish;
439 } else {
440 mylock.Lock();
441 while(!done)
442 cond.Wait(mylock);
443 mylock.Unlock();
444 }
445 return reply;
446}
447
448int librados::IoCtxImpl::selfmanaged_snap_rollback_object(const object_t& oid,
449 ::SnapContext& snapc,
450 uint64_t snapid)
451{
452 int reply;
453
454 Mutex mylock("IoCtxImpl::snap_rollback::mylock");
455 Cond cond;
456 bool done;
457 Context *onack = new C_SafeCond(&mylock, &cond, &done, &reply);
458
459 ::ObjectOperation op;
460 prepare_assert_ops(&op);
461 op.rollback(snapid);
462 objecter->mutate(oid, oloc,
463 op, snapc, ceph::real_clock::now(), 0,
464 onack, NULL);
465
466 mylock.Lock();
467 while (!done) cond.Wait(mylock);
468 mylock.Unlock();
469 return reply;
470}
471
472int librados::IoCtxImpl::rollback(const object_t& oid, const char *snapName)
473{
474 snapid_t snap;
475
476 int r = objecter->pool_snap_by_name(poolid, snapName, &snap);
477 if (r < 0) {
478 return r;
479 }
480
481 return selfmanaged_snap_rollback_object(oid, snapc, snap);
482}
483
484int librados::IoCtxImpl::selfmanaged_snap_remove(uint64_t snapid)
485{
486 int reply;
487
488 Mutex mylock("IoCtxImpl::selfmanaged_snap_remove::mylock");
489 Cond cond;
490 bool done;
491 objecter->delete_selfmanaged_snap(poolid, snapid_t(snapid),
492 new C_SafeCond(&mylock, &cond, &done, &reply));
493
494 mylock.Lock();
495 while (!done) cond.Wait(mylock);
496 mylock.Unlock();
497 return (int)reply;
498}
499
500void librados::IoCtxImpl::aio_selfmanaged_snap_remove(uint64_t snapid,
501 AioCompletionImpl *c)
502{
503 Context *onfinish = new C_aio_selfmanaged_snap_op_Complete(client, c);
504 objecter->delete_selfmanaged_snap(poolid, snapid, onfinish);
505}
506
507int librados::IoCtxImpl::pool_change_auid(unsigned long long auid)
508{
509 int reply;
510
511 Mutex mylock("IoCtxImpl::pool_change_auid::mylock");
512 Cond cond;
513 bool done;
514 objecter->change_pool_auid(poolid,
515 new C_SafeCond(&mylock, &cond, &done, &reply),
516 auid);
517
518 mylock.Lock();
519 while (!done) cond.Wait(mylock);
520 mylock.Unlock();
521 return reply;
522}
523
524int librados::IoCtxImpl::pool_change_auid_async(unsigned long long auid,
525 PoolAsyncCompletionImpl *c)
526{
527 objecter->change_pool_auid(poolid,
528 new C_PoolAsync_Safe(c),
529 auid);
530 return 0;
531}
532
533int librados::IoCtxImpl::snap_list(vector<uint64_t> *snaps)
534{
535 return objecter->pool_snap_list(poolid, snaps);
536}
537
538int librados::IoCtxImpl::snap_lookup(const char *name, uint64_t *snapid)
539{
540 return objecter->pool_snap_by_name(poolid, name, (snapid_t *)snapid);
541}
542
543int librados::IoCtxImpl::snap_get_name(uint64_t snapid, std::string *s)
544{
545 pool_snap_info_t info;
546 int ret = objecter->pool_snap_get_info(poolid, snapid, &info);
547 if (ret < 0) {
548 return ret;
549 }
550 *s = info.name.c_str();
551 return 0;
552}
553
554int librados::IoCtxImpl::snap_get_stamp(uint64_t snapid, time_t *t)
555{
556 pool_snap_info_t info;
557 int ret = objecter->pool_snap_get_info(poolid, snapid, &info);
558 if (ret < 0) {
559 return ret;
560 }
561 *t = info.stamp.sec();
562 return 0;
563}
564
565
566// IO
567
568int librados::IoCtxImpl::nlist(Objecter::NListContext *context, int max_entries)
569{
570 Cond cond;
571 bool done;
572 int r = 0;
573 Mutex mylock("IoCtxImpl::nlist::mylock");
574
575 if (context->at_end())
576 return 0;
577
578 context->max_entries = max_entries;
579 context->nspace = oloc.nspace;
580
581 objecter->list_nobjects(context, new C_SafeCond(&mylock, &cond, &done, &r));
582
583 mylock.Lock();
584 while(!done)
585 cond.Wait(mylock);
586 mylock.Unlock();
587
588 return r;
589}
590
591uint32_t librados::IoCtxImpl::nlist_seek(Objecter::NListContext *context,
592 uint32_t pos)
593{
594 context->list.clear();
595 return objecter->list_nobjects_seek(context, pos);
596}
597
598uint32_t librados::IoCtxImpl::nlist_seek(Objecter::NListContext *context,
599 const rados_object_list_cursor& cursor)
600{
601 context->list.clear();
602 return objecter->list_nobjects_seek(context, *(const hobject_t *)cursor);
603}
604
605rados_object_list_cursor librados::IoCtxImpl::nlist_get_cursor(Objecter::NListContext *context)
606{
607 hobject_t *c = new hobject_t;
608
609 objecter->list_nobjects_get_cursor(context, c);
610 return (rados_object_list_cursor)c;
611}
612
613int librados::IoCtxImpl::create(const object_t& oid, bool exclusive)
614{
615 ::ObjectOperation op;
616 prepare_assert_ops(&op);
617 op.create(exclusive);
618 return operate(oid, &op, NULL);
619}
620
621/*
622 * add any version assert operations that are appropriate given the
623 * stat in the IoCtx, either the target version assert or any src
624 * object asserts. these affect a single ioctx operation, so clear
625 * the ioctx state when we're doing.
626 *
627 * return a pointer to the ObjectOperation if we added any events;
628 * this is convenient for passing the extra_ops argument into Objecter
629 * methods.
630 */
631::ObjectOperation *librados::IoCtxImpl::prepare_assert_ops(::ObjectOperation *op)
632{
633 ::ObjectOperation *pop = NULL;
634 if (assert_ver) {
635 op->assert_version(assert_ver);
636 assert_ver = 0;
637 pop = op;
638 }
639 return pop;
640}
641
642int librados::IoCtxImpl::write(const object_t& oid, bufferlist& bl,
643 size_t len, uint64_t off)
644{
645 if (len > UINT_MAX/2)
646 return -E2BIG;
647 ::ObjectOperation op;
648 prepare_assert_ops(&op);
649 bufferlist mybl;
650 mybl.substr_of(bl, 0, len);
651 op.write(off, mybl);
652 return operate(oid, &op, NULL);
653}
654
655int librados::IoCtxImpl::append(const object_t& oid, bufferlist& bl, size_t len)
656{
657 if (len > UINT_MAX/2)
658 return -E2BIG;
659 ::ObjectOperation op;
660 prepare_assert_ops(&op);
661 bufferlist mybl;
662 mybl.substr_of(bl, 0, len);
663 op.append(mybl);
664 return operate(oid, &op, NULL);
665}
666
667int librados::IoCtxImpl::write_full(const object_t& oid, bufferlist& bl)
668{
669 if (bl.length() > UINT_MAX/2)
670 return -E2BIG;
671 ::ObjectOperation op;
672 prepare_assert_ops(&op);
673 op.write_full(bl);
674 return operate(oid, &op, NULL);
675}
676
677int librados::IoCtxImpl::writesame(const object_t& oid, bufferlist& bl,
678 size_t write_len, uint64_t off)
679{
680 if ((bl.length() > UINT_MAX/2) || (write_len > UINT_MAX/2))
681 return -E2BIG;
682 if ((bl.length() == 0) || (write_len % bl.length()))
683 return -EINVAL;
684 ::ObjectOperation op;
685 prepare_assert_ops(&op);
686 bufferlist mybl;
687 mybl.substr_of(bl, 0, bl.length());
688 op.writesame(off, write_len, mybl);
689 return operate(oid, &op, NULL);
690}
691
692int librados::IoCtxImpl::operate(const object_t& oid, ::ObjectOperation *o,
693 ceph::real_time *pmtime, int flags)
694{
695 ceph::real_time ut = (pmtime ? *pmtime :
696 ceph::real_clock::now());
697
698 /* can't write to a snapshot */
699 if (snap_seq != CEPH_NOSNAP)
700 return -EROFS;
701
702 if (!o->size())
703 return 0;
704
705 Mutex mylock("IoCtxImpl::operate::mylock");
706 Cond cond;
707 bool done;
708 int r;
709 version_t ver;
710
711 Context *oncommit = new C_SafeCond(&mylock, &cond, &done, &r);
712
713 int op = o->ops[0].op.op;
714 ldout(client->cct, 10) << ceph_osd_op_name(op) << " oid=" << oid
715 << " nspace=" << oloc.nspace << dendl;
716 Objecter::Op *objecter_op = objecter->prepare_mutate_op(oid, oloc,
717 *o, snapc, ut, flags,
718 oncommit, &ver);
719 objecter->op_submit(objecter_op);
720
721 mylock.Lock();
722 while (!done)
723 cond.Wait(mylock);
724 mylock.Unlock();
725 ldout(client->cct, 10) << "Objecter returned from "
726 << ceph_osd_op_name(op) << " r=" << r << dendl;
727
728 set_sync_op_version(ver);
729
730 return r;
731}
732
733int librados::IoCtxImpl::operate_read(const object_t& oid,
734 ::ObjectOperation *o,
735 bufferlist *pbl,
736 int flags)
737{
738 if (!o->size())
739 return 0;
740
741 Mutex mylock("IoCtxImpl::operate_read::mylock");
742 Cond cond;
743 bool done;
744 int r;
745 version_t ver;
746
747 Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
748
749 int op = o->ops[0].op.op;
750 ldout(client->cct, 10) << ceph_osd_op_name(op) << " oid=" << oid << " nspace=" << oloc.nspace << dendl;
751 Objecter::Op *objecter_op = objecter->prepare_read_op(oid, oloc,
752 *o, snap_seq, pbl, flags,
753 onack, &ver);
754 objecter->op_submit(objecter_op);
755
756 mylock.Lock();
757 while (!done)
758 cond.Wait(mylock);
759 mylock.Unlock();
760 ldout(client->cct, 10) << "Objecter returned from "
761 << ceph_osd_op_name(op) << " r=" << r << dendl;
762
763 set_sync_op_version(ver);
764
765 return r;
766}
767
768int librados::IoCtxImpl::aio_operate_read(const object_t &oid,
769 ::ObjectOperation *o,
770 AioCompletionImpl *c,
771 int flags,
772 bufferlist *pbl,
773 const blkin_trace_info *trace_info)
774{
775 FUNCTRACE();
776 Context *oncomplete = new C_aio_Complete(c);
777
778#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
779 ((C_aio_Complete *) oncomplete)->oid = oid;
780#endif
781 c->is_read = true;
782 c->io = this;
783
784 ZTracer::Trace trace;
785 if (trace_info)
786 trace.init("rados operate read", &objecter->trace_endpoint, trace_info);
787
788 trace.event("init root span");
789 Objecter::Op *objecter_op = objecter->prepare_read_op(oid, oloc,
790 *o, snap_seq, pbl, flags,
791 oncomplete, &c->objver, nullptr, 0, &trace);
792 objecter->op_submit(objecter_op, &c->tid);
793 trace.event("rados operate read submitted");
794
795 return 0;
796}
797
798int librados::IoCtxImpl::aio_operate(const object_t& oid,
799 ::ObjectOperation *o, AioCompletionImpl *c,
800 const SnapContext& snap_context, int flags,
801 const blkin_trace_info *trace_info)
802{
803 FUNCTRACE();
804 OID_EVENT_TRACE(oid.name.c_str(), "RADOS_WRITE_OP_BEGIN");
805 auto ut = ceph::real_clock::now();
806 /* can't write to a snapshot */
807 if (snap_seq != CEPH_NOSNAP)
808 return -EROFS;
809
810 Context *oncomplete = new C_aio_Complete(c);
811#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
812 ((C_aio_Complete *) oncomplete)->oid = oid;
813#endif
814
815 c->io = this;
816 queue_aio_write(c);
817
818 ZTracer::Trace trace;
819 if (trace_info)
820 trace.init("rados operate", &objecter->trace_endpoint, trace_info);
821
822 trace.event("init root span");
823 Objecter::Op *op = objecter->prepare_mutate_op(
824 oid, oloc, *o, snap_context, ut, flags,
825 oncomplete, &c->objver, osd_reqid_t(), &trace);
826 objecter->op_submit(op, &c->tid);
827 trace.event("rados operate op submitted");
828
829 return 0;
830}
831
832int librados::IoCtxImpl::aio_read(const object_t oid, AioCompletionImpl *c,
833 bufferlist *pbl, size_t len, uint64_t off,
834 uint64_t snapid, const blkin_trace_info *info)
835{
836 FUNCTRACE();
837 if (len > (size_t) INT_MAX)
838 return -EDOM;
839
840 OID_EVENT_TRACE(oid.name.c_str(), "RADOS_READ_OP_BEGIN");
841 Context *oncomplete = new C_aio_Complete(c);
842
843#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
844 ((C_aio_Complete *) oncomplete)->oid = oid;
845#endif
846 c->is_read = true;
847 c->io = this;
848 c->blp = pbl;
849
850 ZTracer::Trace trace;
851 if (info)
852 trace.init("rados read", &objecter->trace_endpoint, info);
853
854 Objecter::Op *o = objecter->prepare_read_op(
855 oid, oloc,
856 off, len, snapid, pbl, 0,
857 oncomplete, &c->objver, nullptr, 0, &trace);
858 objecter->op_submit(o, &c->tid);
859 return 0;
860}
861
862int librados::IoCtxImpl::aio_read(const object_t oid, AioCompletionImpl *c,
863 char *buf, size_t len, uint64_t off,
864 uint64_t snapid, const blkin_trace_info *info)
865{
866 FUNCTRACE();
867 if (len > (size_t) INT_MAX)
868 return -EDOM;
869
870 OID_EVENT_TRACE(oid.name.c_str(), "RADOS_READ_OP_BEGIN");
871 Context *oncomplete = new C_aio_Complete(c);
872
873#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
874 ((C_aio_Complete *) oncomplete)->oid = oid;
875#endif
876 c->is_read = true;
877 c->io = this;
878 c->bl.clear();
879 c->bl.push_back(buffer::create_static(len, buf));
880 c->blp = &c->bl;
881 c->out_buf = buf;
882
883 ZTracer::Trace trace;
884 if (info)
885 trace.init("rados read", &objecter->trace_endpoint, info);
886
887 Objecter::Op *o = objecter->prepare_read_op(
888 oid, oloc,
889 off, len, snapid, &c->bl, 0,
890 oncomplete, &c->objver, nullptr, 0, &trace);
891 objecter->op_submit(o, &c->tid);
892 return 0;
893}
894
895class C_ObjectOperation : public Context {
896public:
897 ::ObjectOperation m_ops;
898 explicit C_ObjectOperation(Context *c) : m_ctx(c) {}
899 void finish(int r) override {
900 m_ctx->complete(r);
901 }
902private:
903 Context *m_ctx;
904};
905
906int librados::IoCtxImpl::aio_sparse_read(const object_t oid,
907 AioCompletionImpl *c,
908 std::map<uint64_t,uint64_t> *m,
909 bufferlist *data_bl, size_t len,
910 uint64_t off, uint64_t snapid)
911{
912 FUNCTRACE();
913 if (len > (size_t) INT_MAX)
914 return -EDOM;
915
916 Context *nested = new C_aio_Complete(c);
917 C_ObjectOperation *onack = new C_ObjectOperation(nested);
918
919#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
920 ((C_aio_Complete *) nested)->oid = oid;
921#endif
922 c->is_read = true;
923 c->io = this;
924
925 onack->m_ops.sparse_read(off, len, m, data_bl, NULL);
926
927 Objecter::Op *o = objecter->prepare_read_op(
928 oid, oloc,
929 onack->m_ops, snapid, NULL, 0,
930 onack, &c->objver);
931 objecter->op_submit(o, &c->tid);
932 return 0;
933}
934
935int librados::IoCtxImpl::aio_cmpext(const object_t& oid,
936 AioCompletionImpl *c,
937 uint64_t off,
938 bufferlist& cmp_bl)
939{
940 if (cmp_bl.length() > UINT_MAX/2)
941 return -E2BIG;
942
943 Context *onack = new C_aio_Complete(c);
944
945 c->is_read = true;
946 c->io = this;
947
948 Objecter::Op *o = objecter->prepare_cmpext_op(
949 oid, oloc, off, cmp_bl, snap_seq, 0,
950 onack, &c->objver);
951 objecter->op_submit(o, &c->tid);
952
953 return 0;
954}
955
956/* use m_ops.cmpext() + prepare_read_op() for non-bufferlist C API */
957int librados::IoCtxImpl::aio_cmpext(const object_t& oid,
958 AioCompletionImpl *c,
959 const char *cmp_buf,
960 size_t cmp_len,
961 uint64_t off)
962{
963 if (cmp_len > UINT_MAX/2)
964 return -E2BIG;
965
966 bufferlist cmp_bl;
967 cmp_bl.append(cmp_buf, cmp_len);
968
969 Context *nested = new C_aio_Complete(c);
970 C_ObjectOperation *onack = new C_ObjectOperation(nested);
971
972 c->is_read = true;
973 c->io = this;
974
975 onack->m_ops.cmpext(off, cmp_len, cmp_buf, NULL);
976
977 Objecter::Op *o = objecter->prepare_read_op(
978 oid, oloc, onack->m_ops, snap_seq, NULL, 0, onack, &c->objver);
979 objecter->op_submit(o, &c->tid);
980 return 0;
981}
982
983int librados::IoCtxImpl::aio_write(const object_t &oid, AioCompletionImpl *c,
984 const bufferlist& bl, size_t len,
985 uint64_t off, const blkin_trace_info *info)
986{
987 FUNCTRACE();
988 auto ut = ceph::real_clock::now();
989 ldout(client->cct, 20) << "aio_write " << oid << " " << off << "~" << len << " snapc=" << snapc << " snap_seq=" << snap_seq << dendl;
990 OID_EVENT_TRACE(oid.name.c_str(), "RADOS_WRITE_OP_BEGIN");
991
992 if (len > UINT_MAX/2)
993 return -E2BIG;
994 /* can't write to a snapshot */
995 if (snap_seq != CEPH_NOSNAP)
996 return -EROFS;
997
998 Context *oncomplete = new C_aio_Complete(c);
999
1000#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
1001 ((C_aio_Complete *) oncomplete)->oid = oid;
1002#endif
1003 ZTracer::Trace trace;
1004 if (info)
1005 trace.init("rados write", &objecter->trace_endpoint, info);
1006
1007 c->io = this;
1008 queue_aio_write(c);
1009
1010 Objecter::Op *o = objecter->prepare_write_op(
1011 oid, oloc,
1012 off, len, snapc, bl, ut, 0,
1013 oncomplete, &c->objver, nullptr, 0, &trace);
1014 objecter->op_submit(o, &c->tid);
1015
1016 return 0;
1017}
1018
1019int librados::IoCtxImpl::aio_append(const object_t &oid, AioCompletionImpl *c,
1020 const bufferlist& bl, size_t len)
1021{
1022 FUNCTRACE();
1023 auto ut = ceph::real_clock::now();
1024
1025 if (len > UINT_MAX/2)
1026 return -E2BIG;
1027 /* can't write to a snapshot */
1028 if (snap_seq != CEPH_NOSNAP)
1029 return -EROFS;
1030
1031 Context *oncomplete = new C_aio_Complete(c);
1032#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
1033 ((C_aio_Complete *) oncomplete)->oid = oid;
1034#endif
1035
1036 c->io = this;
1037 queue_aio_write(c);
1038
1039 Objecter::Op *o = objecter->prepare_append_op(
1040 oid, oloc,
1041 len, snapc, bl, ut, 0,
1042 oncomplete, &c->objver);
1043 objecter->op_submit(o, &c->tid);
1044
1045 return 0;
1046}
1047
1048int librados::IoCtxImpl::aio_write_full(const object_t &oid,
1049 AioCompletionImpl *c,
1050 const bufferlist& bl)
1051{
1052 FUNCTRACE();
1053 auto ut = ceph::real_clock::now();
1054
1055 if (bl.length() > UINT_MAX/2)
1056 return -E2BIG;
1057 /* can't write to a snapshot */
1058 if (snap_seq != CEPH_NOSNAP)
1059 return -EROFS;
1060
1061 Context *oncomplete = new C_aio_Complete(c);
1062#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
1063 ((C_aio_Complete *) oncomplete)->oid = oid;
1064#endif
1065
1066 c->io = this;
1067 queue_aio_write(c);
1068
1069 Objecter::Op *o = objecter->prepare_write_full_op(
1070 oid, oloc,
1071 snapc, bl, ut, 0,
1072 oncomplete, &c->objver);
1073 objecter->op_submit(o, &c->tid);
1074
1075 return 0;
1076}
1077
1078int librados::IoCtxImpl::aio_writesame(const object_t &oid,
1079 AioCompletionImpl *c,
1080 const bufferlist& bl,
1081 size_t write_len,
1082 uint64_t off)
1083{
1084 FUNCTRACE();
1085 auto ut = ceph::real_clock::now();
1086
1087 if ((bl.length() > UINT_MAX/2) || (write_len > UINT_MAX/2))
1088 return -E2BIG;
1089 if ((bl.length() == 0) || (write_len % bl.length()))
1090 return -EINVAL;
1091 /* can't write to a snapshot */
1092 if (snap_seq != CEPH_NOSNAP)
1093 return -EROFS;
1094
1095 Context *oncomplete = new C_aio_Complete(c);
1096
1097#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
1098 ((C_aio_Complete *) oncomplete)->oid = oid;
1099#endif
1100 c->io = this;
1101 queue_aio_write(c);
1102
1103 Objecter::Op *o = objecter->prepare_writesame_op(
1104 oid, oloc,
1105 write_len, off,
1106 snapc, bl, ut, 0,
1107 oncomplete, &c->objver);
1108 objecter->op_submit(o, &c->tid);
1109
1110 return 0;
1111}
1112
1113int librados::IoCtxImpl::aio_remove(const object_t &oid, AioCompletionImpl *c, int flags)
1114{
1115 FUNCTRACE();
1116 auto ut = ceph::real_clock::now();
1117
1118 /* can't write to a snapshot */
1119 if (snap_seq != CEPH_NOSNAP)
1120 return -EROFS;
1121
1122 Context *oncomplete = new C_aio_Complete(c);
1123
1124#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
1125 ((C_aio_Complete *) oncomplete)->oid = oid;
1126#endif
1127 c->io = this;
1128 queue_aio_write(c);
1129
1130 Objecter::Op *o = objecter->prepare_remove_op(
1131 oid, oloc,
1132 snapc, ut, flags,
1133 oncomplete, &c->objver);
1134 objecter->op_submit(o, &c->tid);
1135
1136 return 0;
1137}
1138
1139
1140int librados::IoCtxImpl::aio_stat(const object_t& oid, AioCompletionImpl *c,
1141 uint64_t *psize, time_t *pmtime)
1142{
1143 C_aio_stat_Ack *onack = new C_aio_stat_Ack(c, pmtime);
1144 c->is_read = true;
1145 c->io = this;
1146 Objecter::Op *o = objecter->prepare_stat_op(
1147 oid, oloc,
1148 snap_seq, psize, &onack->mtime, 0,
1149 onack, &c->objver);
1150 objecter->op_submit(o, &c->tid);
1151 return 0;
1152}
1153
1154int librados::IoCtxImpl::aio_stat2(const object_t& oid, AioCompletionImpl *c,
1155 uint64_t *psize, struct timespec *pts)
1156{
1157 C_aio_stat2_Ack *onack = new C_aio_stat2_Ack(c, pts);
1158 c->is_read = true;
1159 c->io = this;
1160 Objecter::Op *o = objecter->prepare_stat_op(
1161 oid, oloc,
1162 snap_seq, psize, &onack->mtime, 0,
1163 onack, &c->objver);
1164 objecter->op_submit(o, &c->tid);
1165 return 0;
1166}
1167
1168int librados::IoCtxImpl::aio_getxattr(const object_t& oid, AioCompletionImpl *c,
1169 const char *name, bufferlist& bl)
1170{
1171 ::ObjectOperation rd;
1172 prepare_assert_ops(&rd);
1173 rd.getxattr(name, &bl, NULL);
1174 int r = aio_operate_read(oid, &rd, c, 0, &bl);
1175 return r;
1176}
1177
1178int librados::IoCtxImpl::aio_rmxattr(const object_t& oid, AioCompletionImpl *c,
1179 const char *name)
1180{
1181 ::ObjectOperation op;
1182 prepare_assert_ops(&op);
1183 op.rmxattr(name);
1184 return aio_operate(oid, &op, c, snapc, 0);
1185}
1186
1187int librados::IoCtxImpl::aio_setxattr(const object_t& oid, AioCompletionImpl *c,
1188 const char *name, bufferlist& bl)
1189{
1190 ::ObjectOperation op;
1191 prepare_assert_ops(&op);
1192 op.setxattr(name, bl);
1193 return aio_operate(oid, &op, c, snapc, 0);
1194}
1195
1196namespace {
1197struct AioGetxattrsData {
1198 AioGetxattrsData(librados::AioCompletionImpl *c, map<string, bufferlist>* attrset,
1199 librados::RadosClient *_client) :
1200 user_completion(c), user_attrset(attrset), client(_client) {}
1201 struct librados::C_AioCompleteAndSafe user_completion;
1202 map<string, bufferlist> result_attrset;
1203 map<std::string, bufferlist>* user_attrset;
1204 librados::RadosClient *client;
1205};
1206}
1207
1208static void aio_getxattrs_complete(rados_completion_t c, void *arg) {
1209 AioGetxattrsData *cdata = reinterpret_cast<AioGetxattrsData*>(arg);
1210 int rc = rados_aio_get_return_value(c);
1211 cdata->user_attrset->clear();
1212 if (rc >= 0) {
1213 for (map<string,bufferlist>::iterator p = cdata->result_attrset.begin();
1214 p != cdata->result_attrset.end();
1215 ++p) {
1216 ldout(cdata->client->cct, 10) << "IoCtxImpl::getxattrs: xattr=" << p->first << dendl;
1217 (*cdata->user_attrset)[p->first] = p->second;
1218 }
1219 }
1220 cdata->user_completion.finish(rc);
1221 ((librados::AioCompletionImpl*)c)->put();
1222 delete cdata;
1223}
1224
1225int librados::IoCtxImpl::aio_getxattrs(const object_t& oid, AioCompletionImpl *c,
1226 map<std::string, bufferlist>& attrset)
1227{
1228 AioGetxattrsData *cdata = new AioGetxattrsData(c, &attrset, client);
1229 ::ObjectOperation rd;
1230 prepare_assert_ops(&rd);
1231 rd.getxattrs(&cdata->result_attrset, NULL);
1232 librados::AioCompletionImpl *comp = new librados::AioCompletionImpl;
1233 comp->set_complete_callback(cdata, aio_getxattrs_complete);
1234 return aio_operate_read(oid, &rd, comp, 0, NULL);
1235}
1236
1237int librados::IoCtxImpl::aio_cancel(AioCompletionImpl *c)
1238{
1239 return objecter->op_cancel(c->tid, -ECANCELED);
1240}
1241
1242
1243int librados::IoCtxImpl::hit_set_list(uint32_t hash, AioCompletionImpl *c,
1244 std::list< std::pair<time_t, time_t> > *pls)
1245{
1246 Context *oncomplete = new C_aio_Complete(c);
1247 c->is_read = true;
1248 c->io = this;
1249
1250 ::ObjectOperation rd;
1251 rd.hit_set_ls(pls, NULL);
1252 object_locator_t oloc(poolid);
1253 Objecter::Op *o = objecter->prepare_pg_read_op(
1254 hash, oloc, rd, NULL, 0, oncomplete, NULL, NULL);
1255 objecter->op_submit(o, &c->tid);
1256 return 0;
1257}
1258
1259int librados::IoCtxImpl::hit_set_get(uint32_t hash, AioCompletionImpl *c,
1260 time_t stamp,
1261 bufferlist *pbl)
1262{
1263 Context *oncomplete = new C_aio_Complete(c);
1264 c->is_read = true;
1265 c->io = this;
1266
1267 ::ObjectOperation rd;
1268 rd.hit_set_get(ceph::real_clock::from_time_t(stamp), pbl, 0);
1269 object_locator_t oloc(poolid);
1270 Objecter::Op *o = objecter->prepare_pg_read_op(
1271 hash, oloc, rd, NULL, 0, oncomplete, NULL, NULL);
1272 objecter->op_submit(o, &c->tid);
1273 return 0;
1274}
1275
1276int librados::IoCtxImpl::remove(const object_t& oid)
1277{
1278 ::ObjectOperation op;
1279 prepare_assert_ops(&op);
1280 op.remove();
1281 return operate(oid, &op, NULL);
1282}
1283
1284int librados::IoCtxImpl::remove(const object_t& oid, int flags)
1285{
1286 ::ObjectOperation op;
1287 prepare_assert_ops(&op);
1288 op.remove();
1289 return operate(oid, &op, NULL, flags);
1290}
1291
1292int librados::IoCtxImpl::trunc(const object_t& oid, uint64_t size)
1293{
1294 ::ObjectOperation op;
1295 prepare_assert_ops(&op);
1296 op.truncate(size);
1297 return operate(oid, &op, NULL);
1298}
1299
1300int librados::IoCtxImpl::get_inconsistent_objects(const pg_t& pg,
1301 const librados::object_id_t& start_after,
1302 uint64_t max_to_get,
1303 AioCompletionImpl *c,
1304 std::vector<inconsistent_obj_t>* objects,
1305 uint32_t* interval)
1306{
1307 Context *oncomplete = new C_aio_Complete(c);
1308 c->is_read = true;
1309 c->io = this;
1310
1311 ::ObjectOperation op;
1312 op.scrub_ls(start_after, max_to_get, objects, interval, nullptr);
1313 object_locator_t oloc{poolid, pg.ps()};
1314 Objecter::Op *o = objecter->prepare_pg_read_op(
1315 oloc.hash, oloc, op, nullptr, CEPH_OSD_FLAG_PGOP, oncomplete,
1316 nullptr, nullptr);
1317 objecter->op_submit(o, &c->tid);
1318 return 0;
1319}
1320
1321int librados::IoCtxImpl::get_inconsistent_snapsets(const pg_t& pg,
1322 const librados::object_id_t& start_after,
1323 uint64_t max_to_get,
1324 AioCompletionImpl *c,
1325 std::vector<inconsistent_snapset_t>* snapsets,
1326 uint32_t* interval)
1327{
1328 Context *oncomplete = new C_aio_Complete(c);
1329 c->is_read = true;
1330 c->io = this;
1331
1332 ::ObjectOperation op;
1333 op.scrub_ls(start_after, max_to_get, snapsets, interval, nullptr);
1334 object_locator_t oloc{poolid, pg.ps()};
1335 Objecter::Op *o = objecter->prepare_pg_read_op(
1336 oloc.hash, oloc, op, nullptr, CEPH_OSD_FLAG_PGOP, oncomplete,
1337 nullptr, nullptr);
1338 objecter->op_submit(o, &c->tid);
1339 return 0;
1340}
1341
1342int librados::IoCtxImpl::tmap_update(const object_t& oid, bufferlist& cmdbl)
1343{
1344 ::ObjectOperation wr;
1345 prepare_assert_ops(&wr);
1346 wr.tmap_update(cmdbl);
1347 return operate(oid, &wr, NULL);
1348}
1349
1350int librados::IoCtxImpl::tmap_put(const object_t& oid, bufferlist& bl)
1351{
1352 ::ObjectOperation wr;
1353 prepare_assert_ops(&wr);
1354 wr.tmap_put(bl);
1355 return operate(oid, &wr, NULL);
1356}
1357
1358int librados::IoCtxImpl::tmap_get(const object_t& oid, bufferlist& bl)
1359{
1360 ::ObjectOperation rd;
1361 prepare_assert_ops(&rd);
1362 rd.tmap_get(&bl, NULL);
1363 return operate_read(oid, &rd, NULL);
1364}
1365
1366int librados::IoCtxImpl::tmap_to_omap(const object_t& oid, bool nullok)
1367{
1368 ::ObjectOperation wr;
1369 prepare_assert_ops(&wr);
1370 wr.tmap_to_omap(nullok);
1371 return operate(oid, &wr, NULL);
1372}
1373
1374int librados::IoCtxImpl::exec(const object_t& oid,
1375 const char *cls, const char *method,
1376 bufferlist& inbl, bufferlist& outbl)
1377{
1378 ::ObjectOperation rd;
1379 prepare_assert_ops(&rd);
1380 rd.call(cls, method, inbl);
1381 return operate_read(oid, &rd, &outbl);
1382}
1383
1384int librados::IoCtxImpl::aio_exec(const object_t& oid, AioCompletionImpl *c,
1385 const char *cls, const char *method,
1386 bufferlist& inbl, bufferlist *outbl)
1387{
1388 FUNCTRACE();
1389 Context *oncomplete = new C_aio_Complete(c);
1390
1391#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
1392 ((C_aio_Complete *) oncomplete)->oid = oid;
1393#endif
1394 c->is_read = true;
1395 c->io = this;
1396
1397 ::ObjectOperation rd;
1398 prepare_assert_ops(&rd);
1399 rd.call(cls, method, inbl);
1400 Objecter::Op *o = objecter->prepare_read_op(
1401 oid, oloc, rd, snap_seq, outbl, 0, oncomplete, &c->objver);
1402 objecter->op_submit(o, &c->tid);
1403 return 0;
1404}
1405
1406int librados::IoCtxImpl::aio_exec(const object_t& oid, AioCompletionImpl *c,
1407 const char *cls, const char *method,
1408 bufferlist& inbl, char *buf, size_t out_len)
1409{
1410 FUNCTRACE();
1411 Context *oncomplete = new C_aio_Complete(c);
1412
1413#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
1414 ((C_aio_Complete *) oncomplete)->oid = oid;
1415#endif
1416 c->is_read = true;
1417 c->io = this;
1418 c->bl.clear();
1419 c->bl.push_back(buffer::create_static(out_len, buf));
1420 c->blp = &c->bl;
1421 c->out_buf = buf;
1422
1423 ::ObjectOperation rd;
1424 prepare_assert_ops(&rd);
1425 rd.call(cls, method, inbl);
1426 Objecter::Op *o = objecter->prepare_read_op(
1427 oid, oloc, rd, snap_seq, &c->bl, 0, oncomplete, &c->objver);
1428 objecter->op_submit(o, &c->tid);
1429 return 0;
1430}
1431
1432int librados::IoCtxImpl::read(const object_t& oid,
1433 bufferlist& bl, size_t len, uint64_t off)
1434{
1435 if (len > (size_t) INT_MAX)
1436 return -EDOM;
1437 OID_EVENT_TRACE(oid.name.c_str(), "RADOS_READ_OP_BEGIN");
1438
1439 ::ObjectOperation rd;
1440 prepare_assert_ops(&rd);
1441 rd.read(off, len, &bl, NULL, NULL);
1442 int r = operate_read(oid, &rd, &bl);
1443 if (r < 0)
1444 return r;
1445
1446 if (bl.length() < len) {
1447 ldout(client->cct, 10) << "Returned length " << bl.length()
1448 << " less than original length "<< len << dendl;
1449 }
1450
1451 return bl.length();
1452}
1453
1454int librados::IoCtxImpl::cmpext(const object_t& oid, uint64_t off,
1455 bufferlist& cmp_bl)
1456{
1457 if (cmp_bl.length() > UINT_MAX/2)
1458 return -E2BIG;
1459
1460 ::ObjectOperation op;
1461 prepare_assert_ops(&op);
1462 op.cmpext(off, cmp_bl, NULL);
1463 return operate_read(oid, &op, NULL);
1464}
1465
1466int librados::IoCtxImpl::mapext(const object_t& oid,
1467 uint64_t off, size_t len,
1468 std::map<uint64_t,uint64_t>& m)
1469{
1470 bufferlist bl;
1471
1472 Mutex mylock("IoCtxImpl::read::mylock");
1473 Cond cond;
1474 bool done;
1475 int r;
1476 Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
1477
1478 objecter->mapext(oid, oloc,
1479 off, len, snap_seq, &bl, 0,
1480 onack);
1481
1482 mylock.Lock();
1483 while (!done)
1484 cond.Wait(mylock);
1485 mylock.Unlock();
1486 ldout(client->cct, 10) << "Objecter returned from read r=" << r << dendl;
1487
1488 if (r < 0)
1489 return r;
1490
1491 bufferlist::iterator iter = bl.begin();
1492 ::decode(m, iter);
1493
1494 return m.size();
1495}
1496
1497int librados::IoCtxImpl::sparse_read(const object_t& oid,
1498 std::map<uint64_t,uint64_t>& m,
1499 bufferlist& data_bl, size_t len,
1500 uint64_t off)
1501{
1502 if (len > (size_t) INT_MAX)
1503 return -EDOM;
1504
1505 ::ObjectOperation rd;
1506 prepare_assert_ops(&rd);
1507 rd.sparse_read(off, len, &m, &data_bl, NULL);
1508
1509 int r = operate_read(oid, &rd, NULL);
1510 if (r < 0)
1511 return r;
1512
1513 return m.size();
1514}
1515
1516int librados::IoCtxImpl::checksum(const object_t& oid, uint8_t type,
1517 const bufferlist &init_value, size_t len,
1518 uint64_t off, size_t chunk_size,
1519 bufferlist *pbl)
1520{
1521 if (len > (size_t) INT_MAX) {
1522 return -EDOM;
1523 }
1524
1525 ::ObjectOperation rd;
1526 prepare_assert_ops(&rd);
1527 rd.checksum(type, init_value, off, len, chunk_size, pbl, nullptr, nullptr);
1528
1529 int r = operate_read(oid, &rd, nullptr);
1530 if (r < 0) {
1531 return r;
1532 }
1533
1534 return 0;
1535}
1536
1537int librados::IoCtxImpl::stat(const object_t& oid, uint64_t *psize, time_t *pmtime)
1538{
1539 uint64_t size;
1540 real_time mtime;
1541
1542 if (!psize)
1543 psize = &size;
1544
1545 ::ObjectOperation rd;
1546 prepare_assert_ops(&rd);
1547 rd.stat(psize, &mtime, NULL);
1548 int r = operate_read(oid, &rd, NULL);
1549
1550 if (r >= 0 && pmtime) {
1551 *pmtime = real_clock::to_time_t(mtime);
1552 }
1553
1554 return r;
1555}
1556
1557int librados::IoCtxImpl::stat2(const object_t& oid, uint64_t *psize, struct timespec *pts)
1558{
1559 uint64_t size;
1560 ceph::real_time mtime;
1561
1562 if (!psize)
1563 psize = &size;
1564
1565 ::ObjectOperation rd;
1566 prepare_assert_ops(&rd);
1567 rd.stat(psize, &mtime, NULL);
1568 int r = operate_read(oid, &rd, NULL);
1569 if (r < 0) {
1570 return r;
1571 }
1572
1573 if (pts) {
1574 *pts = ceph::real_clock::to_timespec(mtime);
1575 }
1576
1577 return 0;
1578}
1579
1580int librados::IoCtxImpl::getxattr(const object_t& oid,
1581 const char *name, bufferlist& bl)
1582{
1583 ::ObjectOperation rd;
1584 prepare_assert_ops(&rd);
1585 rd.getxattr(name, &bl, NULL);
1586 int r = operate_read(oid, &rd, &bl);
1587 if (r < 0)
1588 return r;
1589
1590 return bl.length();
1591}
1592
1593int librados::IoCtxImpl::rmxattr(const object_t& oid, const char *name)
1594{
1595 ::ObjectOperation op;
1596 prepare_assert_ops(&op);
1597 op.rmxattr(name);
1598 return operate(oid, &op, NULL);
1599}
1600
1601int librados::IoCtxImpl::setxattr(const object_t& oid,
1602 const char *name, bufferlist& bl)
1603{
1604 ::ObjectOperation op;
1605 prepare_assert_ops(&op);
1606 op.setxattr(name, bl);
1607 return operate(oid, &op, NULL);
1608}
1609
1610int librados::IoCtxImpl::getxattrs(const object_t& oid,
1611 map<std::string, bufferlist>& attrset)
1612{
1613 map<string, bufferlist> aset;
1614
1615 ::ObjectOperation rd;
1616 prepare_assert_ops(&rd);
1617 rd.getxattrs(&aset, NULL);
1618 int r = operate_read(oid, &rd, NULL);
1619
1620 attrset.clear();
1621 if (r >= 0) {
1622 for (map<string,bufferlist>::iterator p = aset.begin(); p != aset.end(); ++p) {
1623 ldout(client->cct, 10) << "IoCtxImpl::getxattrs: xattr=" << p->first << dendl;
1624 attrset[p->first.c_str()] = p->second;
1625 }
1626 }
1627
1628 return r;
1629}
1630
1631void librados::IoCtxImpl::set_sync_op_version(version_t ver)
1632{
1633 ANNOTATE_BENIGN_RACE_SIZED(&last_objver, sizeof(last_objver),
1634 "IoCtxImpl last_objver");
1635 last_objver = ver;
1636}
1637
1638struct WatchInfo : public Objecter::WatchContext {
1639 librados::IoCtxImpl *ioctx;
1640 object_t oid;
1641 librados::WatchCtx *ctx;
1642 librados::WatchCtx2 *ctx2;
1643 bool internal = false;
1644
1645 WatchInfo(librados::IoCtxImpl *io, object_t o,
1646 librados::WatchCtx *c, librados::WatchCtx2 *c2,
1647 bool inter)
1648 : ioctx(io), oid(o), ctx(c), ctx2(c2), internal(inter) {
1649 ioctx->get();
1650 }
1651 ~WatchInfo() override {
1652 ioctx->put();
1653 if (internal) {
1654 delete ctx;
1655 delete ctx2;
1656 }
1657 }
1658
1659 void handle_notify(uint64_t notify_id,
1660 uint64_t cookie,
1661 uint64_t notifier_id,
1662 bufferlist& bl) override {
1663 ldout(ioctx->client->cct, 10) << __func__ << " " << notify_id
1664 << " cookie " << cookie
1665 << " notifier_id " << notifier_id
1666 << " len " << bl.length()
1667 << dendl;
1668
1669 if (ctx2)
1670 ctx2->handle_notify(notify_id, cookie, notifier_id, bl);
1671 if (ctx) {
1672 ctx->notify(0, 0, bl);
1673
1674 // send ACK back to OSD if using legacy protocol
1675 bufferlist empty;
1676 ioctx->notify_ack(oid, notify_id, cookie, empty);
1677 }
1678 }
1679 void handle_error(uint64_t cookie, int err) override {
1680 ldout(ioctx->client->cct, 10) << __func__ << " cookie " << cookie
1681 << " err " << err
1682 << dendl;
1683 if (ctx2)
1684 ctx2->handle_error(cookie, err);
1685 }
1686};
1687
1688int librados::IoCtxImpl::watch(const object_t& oid, uint64_t *handle,
1689 librados::WatchCtx *ctx,
1690 librados::WatchCtx2 *ctx2,
1691 bool internal)
1692{
1693 return watch(oid, handle, ctx, ctx2, 0, internal);
1694}
1695
1696int librados::IoCtxImpl::watch(const object_t& oid, uint64_t *handle,
1697 librados::WatchCtx *ctx,
1698 librados::WatchCtx2 *ctx2,
1699 uint32_t timeout,
1700 bool internal)
1701{
1702 ::ObjectOperation wr;
1703 version_t objver;
1704 C_SaferCond onfinish;
1705
1706 Objecter::LingerOp *linger_op = objecter->linger_register(oid, oloc, 0);
1707 *handle = linger_op->get_cookie();
1708 linger_op->watch_context = new WatchInfo(this,
1709 oid, ctx, ctx2, internal);
1710
1711 prepare_assert_ops(&wr);
1712 wr.watch(*handle, CEPH_OSD_WATCH_OP_WATCH, timeout);
1713 bufferlist bl;
1714 objecter->linger_watch(linger_op, wr,
1715 snapc, ceph::real_clock::now(), bl,
1716 &onfinish,
1717 &objver);
1718
1719 int r = onfinish.wait();
1720
1721 set_sync_op_version(objver);
1722
1723 if (r < 0) {
1724 objecter->linger_cancel(linger_op);
1725 *handle = 0;
1726 }
1727
1728 return r;
1729}
1730
1731int librados::IoCtxImpl::aio_watch(const object_t& oid,
1732 AioCompletionImpl *c,
1733 uint64_t *handle,
1734 librados::WatchCtx *ctx,
1735 librados::WatchCtx2 *ctx2,
1736 bool internal) {
1737 return aio_watch(oid, c, handle, ctx, ctx2, 0, internal);
1738}
1739
1740int librados::IoCtxImpl::aio_watch(const object_t& oid,
1741 AioCompletionImpl *c,
1742 uint64_t *handle,
1743 librados::WatchCtx *ctx,
1744 librados::WatchCtx2 *ctx2,
1745 uint32_t timeout,
1746 bool internal)
1747{
1748 Objecter::LingerOp *linger_op = objecter->linger_register(oid, oloc, 0);
1749 c->io = this;
1750 Context *oncomplete = new C_aio_linger_Complete(c, linger_op, false);
1751
1752 ::ObjectOperation wr;
1753 *handle = linger_op->get_cookie();
1754 linger_op->watch_context = new WatchInfo(this, oid, ctx, ctx2, internal);
1755
1756 prepare_assert_ops(&wr);
1757 wr.watch(*handle, CEPH_OSD_WATCH_OP_WATCH, timeout);
1758 bufferlist bl;
1759 objecter->linger_watch(linger_op, wr,
1760 snapc, ceph::real_clock::now(), bl,
1761 oncomplete, &c->objver);
1762
1763 return 0;
1764}
1765
1766
1767int librados::IoCtxImpl::notify_ack(
1768 const object_t& oid,
1769 uint64_t notify_id,
1770 uint64_t cookie,
1771 bufferlist& bl)
1772{
1773 ::ObjectOperation rd;
1774 prepare_assert_ops(&rd);
1775 rd.notify_ack(notify_id, cookie, bl);
1776 objecter->read(oid, oloc, rd, snap_seq, (bufferlist*)NULL, 0, 0, 0);
1777 return 0;
1778}
1779
1780int librados::IoCtxImpl::watch_check(uint64_t cookie)
1781{
1782 Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
1783 return objecter->linger_check(linger_op);
1784}
1785
1786int librados::IoCtxImpl::unwatch(uint64_t cookie)
1787{
1788 Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
1789 C_SaferCond onfinish;
1790 version_t ver = 0;
1791
1792 ::ObjectOperation wr;
1793 prepare_assert_ops(&wr);
1794 wr.watch(cookie, CEPH_OSD_WATCH_OP_UNWATCH);
1795 objecter->mutate(linger_op->target.base_oid, oloc, wr,
1796 snapc, ceph::real_clock::now(), 0,
1797 &onfinish, &ver);
1798 objecter->linger_cancel(linger_op);
1799
1800 int r = onfinish.wait();
1801 set_sync_op_version(ver);
1802 return r;
1803}
1804
1805int librados::IoCtxImpl::aio_unwatch(uint64_t cookie, AioCompletionImpl *c)
1806{
1807 c->io = this;
1808 Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
1809 Context *oncomplete = new C_aio_linger_Complete(c, linger_op, true);
1810
1811 ::ObjectOperation wr;
1812 prepare_assert_ops(&wr);
1813 wr.watch(cookie, CEPH_OSD_WATCH_OP_UNWATCH);
1814 objecter->mutate(linger_op->target.base_oid, oloc, wr,
1815 snapc, ceph::real_clock::now(), 0,
1816 oncomplete, &c->objver);
1817 return 0;
1818}
1819
1820int librados::IoCtxImpl::notify(const object_t& oid, bufferlist& bl,
1821 uint64_t timeout_ms,
1822 bufferlist *preply_bl,
1823 char **preply_buf, size_t *preply_buf_len)
1824{
1825 Objecter::LingerOp *linger_op = objecter->linger_register(oid, oloc, 0);
1826
1827 C_SaferCond notify_finish_cond;
1828 Context *notify_finish = new C_notify_Finish(client->cct, &notify_finish_cond,
1829 objecter, linger_op, preply_bl,
1830 preply_buf, preply_buf_len);
1831
1832 uint32_t timeout = notify_timeout;
1833 if (timeout_ms)
1834 timeout = timeout_ms / 1000;
1835
1836 // Construct RADOS op
1837 ::ObjectOperation rd;
1838 prepare_assert_ops(&rd);
1839 bufferlist inbl;
1840 rd.notify(linger_op->get_cookie(), 1, timeout, bl, &inbl);
1841
1842 // Issue RADOS op
1843 C_SaferCond onack;
1844 version_t objver;
1845 objecter->linger_notify(linger_op,
1846 rd, snap_seq, inbl, NULL,
1847 &onack, &objver);
1848
1849 ldout(client->cct, 10) << __func__ << " issued linger op " << linger_op << dendl;
1850 int r = onack.wait();
1851 ldout(client->cct, 10) << __func__ << " linger op " << linger_op
1852 << " acked (" << r << ")" << dendl;
1853
1854 if (r == 0) {
1855 ldout(client->cct, 10) << __func__ << " waiting for watch_notify finish "
1856 << linger_op << dendl;
1857 r = notify_finish_cond.wait();
1858
1859 } else {
1860 ldout(client->cct, 10) << __func__ << " failed to initiate notify, r = "
1861 << r << dendl;
1862 notify_finish->complete(r);
1863 }
1864
1865 objecter->linger_cancel(linger_op);
1866
1867 set_sync_op_version(objver);
1868 return r;
1869}
1870
1871int librados::IoCtxImpl::aio_notify(const object_t& oid, AioCompletionImpl *c,
1872 bufferlist& bl, uint64_t timeout_ms,
1873 bufferlist *preply_bl, char **preply_buf,
1874 size_t *preply_buf_len)
1875{
1876 Objecter::LingerOp *linger_op = objecter->linger_register(oid, oloc, 0);
1877
1878 c->io = this;
1879
1880 C_aio_notify_Complete *oncomplete = new C_aio_notify_Complete(c, linger_op);
1881 C_notify_Finish *onnotify = new C_notify_Finish(client->cct, oncomplete,
1882 objecter, linger_op,
1883 preply_bl, preply_buf,
1884 preply_buf_len);
1885 Context *onack = new C_aio_notify_Ack(client->cct, onnotify, oncomplete);
1886
1887 uint32_t timeout = notify_timeout;
1888 if (timeout_ms)
1889 timeout = timeout_ms / 1000;
1890
1891 // Construct RADOS op
1892 ::ObjectOperation rd;
1893 prepare_assert_ops(&rd);
1894 bufferlist inbl;
1895 rd.notify(linger_op->get_cookie(), 1, timeout, bl, &inbl);
1896
1897 // Issue RADOS op
1898 objecter->linger_notify(linger_op,
1899 rd, snap_seq, inbl, NULL,
1900 onack, &c->objver);
1901 return 0;
1902}
1903
1904int librados::IoCtxImpl::set_alloc_hint(const object_t& oid,
1905 uint64_t expected_object_size,
1906 uint64_t expected_write_size,
1907 uint32_t flags)
1908{
1909 ::ObjectOperation wr;
1910 prepare_assert_ops(&wr);
1911 wr.set_alloc_hint(expected_object_size, expected_write_size, flags);
1912 return operate(oid, &wr, NULL);
1913}
1914
1915version_t librados::IoCtxImpl::last_version()
1916{
1917 return last_objver;
1918}
1919
1920void librados::IoCtxImpl::set_assert_version(uint64_t ver)
1921{
1922 assert_ver = ver;
1923}
1924
1925void librados::IoCtxImpl::set_notify_timeout(uint32_t timeout)
1926{
1927 notify_timeout = timeout;
1928}
1929
1930int librados::IoCtxImpl::cache_pin(const object_t& oid)
1931{
1932 ::ObjectOperation wr;
1933 prepare_assert_ops(&wr);
1934 wr.cache_pin();
1935 return operate(oid, &wr, NULL);
1936}
1937
1938int librados::IoCtxImpl::cache_unpin(const object_t& oid)
1939{
1940 ::ObjectOperation wr;
1941 prepare_assert_ops(&wr);
1942 wr.cache_unpin();
1943 return operate(oid, &wr, NULL);
1944}
1945
1946
1947///////////////////////////// C_aio_stat_Ack ////////////////////////////
1948
1949librados::IoCtxImpl::C_aio_stat_Ack::C_aio_stat_Ack(AioCompletionImpl *_c,
1950 time_t *pm)
1951 : c(_c), pmtime(pm)
1952{
1953 assert(!c->io);
1954 c->get();
1955}
1956
1957void librados::IoCtxImpl::C_aio_stat_Ack::finish(int r)
1958{
1959 c->lock.Lock();
1960 c->rval = r;
1961 c->complete = true;
1962 c->cond.Signal();
1963
1964 if (r >= 0 && pmtime) {
1965 *pmtime = real_clock::to_time_t(mtime);
1966 }
1967
1968 if (c->callback_complete) {
1969 c->io->client->finisher.queue(new C_AioComplete(c));
1970 }
1971
1972 c->put_unlock();
1973}
1974
1975///////////////////////////// C_aio_stat2_Ack ////////////////////////////
1976
1977librados::IoCtxImpl::C_aio_stat2_Ack::C_aio_stat2_Ack(AioCompletionImpl *_c,
1978 struct timespec *pt)
1979 : c(_c), pts(pt)
1980{
1981 assert(!c->io);
1982 c->get();
1983}
1984
1985void librados::IoCtxImpl::C_aio_stat2_Ack::finish(int r)
1986{
1987 c->lock.Lock();
1988 c->rval = r;
1989 c->complete = true;
1990 c->cond.Signal();
1991
1992 if (r >= 0 && pts) {
1993 *pts = real_clock::to_timespec(mtime);
1994 }
1995
1996 if (c->callback_complete) {
1997 c->io->client->finisher.queue(new C_AioComplete(c));
1998 }
1999
2000 c->put_unlock();
2001}
2002
2003//////////////////////////// C_aio_Complete ////////////////////////////////
2004
2005librados::IoCtxImpl::C_aio_Complete::C_aio_Complete(AioCompletionImpl *_c)
2006 : c(_c)
2007{
2008 c->get();
2009}
2010
2011void librados::IoCtxImpl::C_aio_Complete::finish(int r)
2012{
2013 c->lock.Lock();
2014 c->rval = r;
2015 c->complete = true;
2016 c->cond.Signal();
2017
2018 if (r == 0 && c->blp && c->blp->length() > 0) {
2019 if (c->out_buf && !c->blp->is_provided_buffer(c->out_buf))
2020 c->blp->copy(0, c->blp->length(), c->out_buf);
2021 c->rval = c->blp->length();
2022 }
2023
2024 if (c->callback_complete ||
2025 c->callback_safe) {
2026 c->io->client->finisher.queue(new C_AioComplete(c));
2027 }
2028
2029 if (c->aio_write_seq) {
2030 c->io->complete_aio_write(c);
2031 }
2032
2033#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
2034 OID_EVENT_TRACE(oid.name.c_str(), "RADOS_OP_COMPLETE");
2035#endif
2036 c->put_unlock();
2037}
2038
2039void librados::IoCtxImpl::object_list_slice(
2040 const hobject_t start,
2041 const hobject_t finish,
2042 const size_t n,
2043 const size_t m,
2044 hobject_t *split_start,
2045 hobject_t *split_finish)
2046{
2047 if (start.is_max()) {
2048 *split_start = hobject_t::get_max();
2049 *split_finish = hobject_t::get_max();
2050 return;
2051 }
2052
2053 uint64_t start_hash = hobject_t::_reverse_bits(start.get_hash());
2054 uint64_t finish_hash =
2055 finish.is_max() ? 0x100000000 :
2056 hobject_t::_reverse_bits(finish.get_hash());
2057
2058 uint64_t diff = finish_hash - start_hash;
2059 uint64_t rev_start = start_hash + (diff * n / m);
2060 uint64_t rev_finish = start_hash + (diff * (n + 1) / m);
2061 if (n == 0) {
2062 *split_start = start;
2063 } else {
2064 *split_start = hobject_t(
2065 object_t(), string(), CEPH_NOSNAP,
2066 hobject_t::_reverse_bits(rev_start), poolid, string());
2067 }
2068
2069 if (n == m - 1)
2070 *split_finish = finish;
2071 else if (rev_finish >= 0x100000000)
2072 *split_finish = hobject_t::get_max();
2073 else
2074 *split_finish = hobject_t(
2075 object_t(), string(), CEPH_NOSNAP,
2076 hobject_t::_reverse_bits(rev_finish), poolid, string());
2077}
2078