ceph/src/rgw/rgw_cr_rados.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #include "include/compat.h"
   5 #include "rgw_sal.h"
   6 #include "rgw_zone.h"
   7 #include "rgw_coroutine.h"
   8 #include "rgw_cr_rados.h"
   9 #include "rgw_sync_counters.h"
  10 #include "rgw_bucket.h"
  11
  12 #include "services/svc_zone.h"
  13 #include "services/svc_zone_utils.h"
  14 #include "services/svc_sys_obj.h"
  15 #include "services/svc_cls.h"
  16
  17 #include "cls/lock/cls_lock_client.h"
  18 #include "cls/rgw/cls_rgw_client.h"
  19
  20 #include <boost/asio/yield.hpp>
  21
  22 #define dout_context g_ceph_context
  23 #define dout_subsys ceph_subsys_rgw
  24
  25 bool RGWAsyncRadosProcessor::RGWWQ::_enqueue(RGWAsyncRadosRequest *req) {
  26   if (processor->is_going_down()) {
  27     return false;
  28   }
  29   req->get();
  30   processor->m_req_queue.push_back(req);
  31   dout(20) << "enqueued request req=" << hex << req << dec << dendl;
  32   _dump_queue();
  33   return true;
  34 }
  35
  36 bool RGWAsyncRadosProcessor::RGWWQ::_empty() {
  37   return processor->m_req_queue.empty();
  38 }
  39
  40 RGWAsyncRadosRequest *RGWAsyncRadosProcessor::RGWWQ::_dequeue() {
  41   if (processor->m_req_queue.empty())
  42     return NULL;
  43   RGWAsyncRadosRequest *req = processor->m_req_queue.front();
  44   processor->m_req_queue.pop_front();
  45   dout(20) << "dequeued request req=" << hex << req << dec << dendl;
  46   _dump_queue();
  47   return req;
  48 }
  49
  50 void RGWAsyncRadosProcessor::RGWWQ::_process(RGWAsyncRadosRequest *req, ThreadPool::TPHandle& handle) {
  51   processor->handle_request(req);
  52   processor->req_throttle.put(1);
  53 }
  54
  55 void RGWAsyncRadosProcessor::RGWWQ::_dump_queue() {
  56   if (!g_conf()->subsys.should_gather<ceph_subsys_rgw, 20>()) {
  57     return;
  58   }
  59   deque<RGWAsyncRadosRequest *>::iterator iter;
  60   if (processor->m_req_queue.empty()) {
  61     dout(20) << "RGWWQ: empty" << dendl;
  62     return;
  63   }
  64   dout(20) << "RGWWQ:" << dendl;
  65   for (iter = processor->m_req_queue.begin(); iter != processor->m_req_queue.end(); ++iter) {
  66     dout(20) << "req: " << hex << *iter << dec << dendl;
  67   }
  68 }
  69
  70 RGWAsyncRadosProcessor::RGWAsyncRadosProcessor(CephContext *_cct, int num_threads)
  71   : cct(_cct), m_tp(cct, "RGWAsyncRadosProcessor::m_tp", "rados_async", num_threads),
  72     req_throttle(_cct, "rgw_async_rados_ops", num_threads * 2),
  73     req_wq(this, g_conf()->rgw_op_thread_timeout,
  74     g_conf()->rgw_op_thread_suicide_timeout, &m_tp) {
  75 }
  76
  77 void RGWAsyncRadosProcessor::start() {
  78   m_tp.start();
  79 }
  80
  81 void RGWAsyncRadosProcessor::stop() {
  82   going_down = true;
  83   m_tp.drain(&req_wq);
  84   m_tp.stop();
  85   for (auto iter = m_req_queue.begin(); iter != m_req_queue.end(); ++iter) {
  86     (*iter)->put();
  87   }
  88 }
  89
  90 void RGWAsyncRadosProcessor::handle_request(RGWAsyncRadosRequest *req) {
  91   req->send_request();
  92   req->put();
  93 }
  94
  95 void RGWAsyncRadosProcessor::queue(RGWAsyncRadosRequest *req) {
  96   req_throttle.get(1);
  97   req_wq.queue(req);
  98 }
  99
 100 int RGWAsyncGetSystemObj::_send_request()
 101 {
 102   map<string, bufferlist> *pattrs = want_attrs ? &attrs : nullptr;
 103
 104   auto sysobj = obj_ctx.get_obj(obj);
 105   return sysobj.rop()
 106                .set_objv_tracker(&objv_tracker)
 107                .set_attrs(pattrs)
 108                .set_raw_attrs(raw_attrs)
 109                .read(&bl, null_yield);
 110 }
 111
 112 RGWAsyncGetSystemObj::RGWAsyncGetSystemObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWSI_SysObj *_svc,
 113                        RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
 114                        bool want_attrs, bool raw_attrs)
 115   : RGWAsyncRadosRequest(caller, cn), obj_ctx(_svc),
 116     obj(_obj), want_attrs(want_attrs), raw_attrs(raw_attrs)
 117 {
 118   if (_objv_tracker) {
 119     objv_tracker = *_objv_tracker;
 120   }
 121 }
 122
 123 int RGWSimpleRadosReadAttrsCR::send_request()
 124 {
 125   req = new RGWAsyncGetSystemObj(this, stack->create_completion_notifier(),
 126                                  svc, nullptr, obj, true, raw_attrs);
 127   async_rados->queue(req);
 128   return 0;
 129 }
 130
 131 int RGWSimpleRadosReadAttrsCR::request_complete()
 132 {
 133   if (pattrs) {
 134     *pattrs = std::move(req->attrs);
 135   }
 136   return req->get_ret_status();
 137 }
 138
 139 int RGWAsyncPutSystemObj::_send_request()
 140 {
 141   auto obj_ctx = svc->init_obj_ctx();
 142   auto sysobj = obj_ctx.get_obj(obj);
 143   return sysobj.wop()
 144                .set_objv_tracker(&objv_tracker)
 145                .set_exclusive(exclusive)
 146                .write_data(bl, null_yield);
 147 }
 148
 149 RGWAsyncPutSystemObj::RGWAsyncPutSystemObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn,
 150                      RGWSI_SysObj *_svc,
 151                      RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
 152                      bool _exclusive, bufferlist _bl)
 153   : RGWAsyncRadosRequest(caller, cn), svc(_svc),
 154     obj(_obj), exclusive(_exclusive), bl(std::move(_bl))
 155 {
 156   if (_objv_tracker) {
 157     objv_tracker = *_objv_tracker;
 158   }
 159 }
 160
 161 int RGWAsyncPutSystemObjAttrs::_send_request()
 162 {
 163   auto obj_ctx = svc->init_obj_ctx();
 164   auto sysobj = obj_ctx.get_obj(obj);
 165   return sysobj.wop()
 166                .set_objv_tracker(&objv_tracker)
 167                .set_exclusive(false)
 168                .set_attrs(attrs)
 169                .write_attrs(null_yield);
 170 }
 171
 172 RGWAsyncPutSystemObjAttrs::RGWAsyncPutSystemObjAttrs(RGWCoroutine *caller, RGWAioCompletionNotifier *cn,
 173                      RGWSI_SysObj *_svc,
 174                      RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
 175                      map<string, bufferlist> _attrs)
 176   : RGWAsyncRadosRequest(caller, cn), svc(_svc),
 177     obj(_obj), attrs(std::move(_attrs))
 178 {
 179   if (_objv_tracker) {
 180     objv_tracker = *_objv_tracker;
 181   }
 182 }
 183
 184
 185 RGWOmapAppend::RGWOmapAppend(RGWAsyncRadosProcessor *_async_rados, rgw::sal::RGWRadosStore *_store, const rgw_raw_obj& _obj,
 186                              uint64_t _window_size)
 187                       : RGWConsumerCR<string>(_store->ctx()), async_rados(_async_rados),
 188                         store(_store), obj(_obj), going_down(false), num_pending_entries(0), window_size(_window_size), total_entries(0)
 189 {
 190 }
 191
 192 int RGWAsyncLockSystemObj::_send_request()
 193 {
 194   rgw_rados_ref ref;
 195   int r = store->getRados()->get_raw_obj_ref(obj, &ref);
 196   if (r < 0) {
 197     lderr(store->ctx()) << "ERROR: failed to get ref for (" << obj << ") ret=" << r << dendl;
 198     return r;
 199   }
 200
 201   rados::cls::lock::Lock l(lock_name);
 202   utime_t duration(duration_secs, 0);
 203   l.set_duration(duration);
 204   l.set_cookie(cookie);
 205   l.set_may_renew(true);
 206
 207   return l.lock_exclusive(&ref.pool.ioctx(), ref.obj.oid);
 208 }
 209
 210 RGWAsyncLockSystemObj::RGWAsyncLockSystemObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, rgw::sal::RGWRadosStore *_store,
 211                       RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
 212                        const string& _name, const string& _cookie, uint32_t _duration_secs) : RGWAsyncRadosRequest(caller, cn), store(_store),
 213                                                               obj(_obj),
 214                                                               lock_name(_name),
 215                                                               cookie(_cookie),
 216                                                               duration_secs(_duration_secs)
 217 {
 218 }
 219
 220 int RGWAsyncUnlockSystemObj::_send_request()
 221 {
 222   rgw_rados_ref ref;
 223   int r = store->getRados()->get_raw_obj_ref(obj, &ref);
 224   if (r < 0) {
 225     lderr(store->ctx()) << "ERROR: failed to get ref for (" << obj << ") ret=" << r << dendl;
 226     return r;
 227   }
 228
 229   rados::cls::lock::Lock l(lock_name);
 230
 231   l.set_cookie(cookie);
 232
 233   return l.unlock(&ref.pool.ioctx(), ref.obj.oid);
 234 }
 235
 236 RGWAsyncUnlockSystemObj::RGWAsyncUnlockSystemObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, rgw::sal::RGWRadosStore *_store,
 237                                                  RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
 238                                                  const string& _name, const string& _cookie) : RGWAsyncRadosRequest(caller, cn), store(_store),
 239   obj(_obj),
 240   lock_name(_name), cookie(_cookie)
 241 {
 242 }
 243
 244 RGWRadosSetOmapKeysCR::RGWRadosSetOmapKeysCR(rgw::sal::RGWRadosStore *_store,
 245                       const rgw_raw_obj& _obj,
 246                       map<string, bufferlist>& _entries) : RGWSimpleCoroutine(_store->ctx()),
 247                                                 store(_store),
 248                                                 entries(_entries),
 249                                                 obj(_obj), cn(NULL)
 250 {
 251   stringstream& s = set_description();
 252   s << "set omap keys dest=" << obj << " keys=[" << s.str() << "]";
 253   for (auto i = entries.begin(); i != entries.end(); ++i) {
 254     if (i != entries.begin()) {
 255       s << ", ";
 256     }
 257     s << i->first;
 258   }
 259   s << "]";
 260 }
 261
 262 int RGWRadosSetOmapKeysCR::send_request()
 263 {
 264   int r = store->getRados()->get_raw_obj_ref(obj, &ref);
 265   if (r < 0) {
 266     lderr(store->ctx()) << "ERROR: failed to get ref for (" << obj << ") ret=" << r << dendl;
 267     return r;
 268   }
 269
 270   set_status() << "sending request";
 271
 272   librados::ObjectWriteOperation op;
 273   op.omap_set(entries);
 274
 275   cn = stack->create_completion_notifier();
 276   return ref.pool.ioctx().aio_operate(ref.obj.oid, cn->completion(), &op);
 277 }
 278
 279 int RGWRadosSetOmapKeysCR::request_complete()
 280 {
 281   int r = cn->completion()->get_return_value();
 282
 283   set_status() << "request complete; ret=" << r;
 284
 285   return r;
 286 }
 287
 288 RGWRadosGetOmapKeysCR::RGWRadosGetOmapKeysCR(rgw::sal::RGWRadosStore *_store,
 289                       const rgw_raw_obj& _obj,
 290                       const string& _marker,
 291                       int _max_entries,
 292                       ResultPtr _result)
 293   : RGWSimpleCoroutine(_store->ctx()), store(_store), obj(_obj),
 294     marker(_marker), max_entries(_max_entries),
 295     result(std::move(_result))
 296 {
 297   ceph_assert(result); // must be allocated
 298   set_description() << "get omap keys dest=" << obj << " marker=" << marker;
 299 }
 300
 301 int RGWRadosGetOmapKeysCR::send_request() {
 302   int r = store->getRados()->get_raw_obj_ref(obj, &result->ref);
 303   if (r < 0) {
 304     lderr(store->ctx()) << "ERROR: failed to get ref for (" << obj << ") ret=" << r << dendl;
 305     return r;
 306   }
 307
 308   set_status() << "send request";
 309
 310   librados::ObjectReadOperation op;
 311   op.omap_get_keys2(marker, max_entries, &result->entries, &result->more, nullptr);
 312
 313   cn = stack->create_completion_notifier(result);
 314   return result->ref.pool.ioctx().aio_operate(result->ref.obj.oid, cn->completion(), &op, NULL);
 315 }
 316
 317 int RGWRadosGetOmapKeysCR::request_complete()
 318 {
 319   int r = cn->completion()->get_return_value();
 320
 321   set_status() << "request complete; ret=" << r;
 322
 323   return r;
 324 }
 325
 326 RGWRadosRemoveOmapKeysCR::RGWRadosRemoveOmapKeysCR(rgw::sal::RGWRadosStore *_store,
 327                       const rgw_raw_obj& _obj,
 328                       const set<string>& _keys) : RGWSimpleCoroutine(_store->ctx()),
 329                                                 store(_store),
 330                                                 keys(_keys),
 331                                                 obj(_obj), cn(NULL)
 332 {
 333   set_description() << "remove omap keys dest=" << obj << " keys=" << keys;
 334 }
 335
 336 int RGWRadosRemoveOmapKeysCR::send_request() {
 337   int r = store->getRados()->get_raw_obj_ref(obj, &ref);
 338   if (r < 0) {
 339     lderr(store->ctx()) << "ERROR: failed to get ref for (" << obj << ") ret=" << r << dendl;
 340     return r;
 341   }
 342
 343   set_status() << "send request";
 344
 345   librados::ObjectWriteOperation op;
 346   op.omap_rm_keys(keys);
 347
 348   cn = stack->create_completion_notifier();
 349   return ref.pool.ioctx().aio_operate(ref.obj.oid, cn->completion(), &op);
 350 }
 351
 352 int RGWRadosRemoveOmapKeysCR::request_complete()
 353 {
 354   int r = cn->completion()->get_return_value();
 355
 356   set_status() << "request complete; ret=" << r;
 357
 358   return r;
 359 }
 360
 361 RGWRadosRemoveCR::RGWRadosRemoveCR(rgw::sal::RGWRadosStore *store, const rgw_raw_obj& obj)
 362   : RGWSimpleCoroutine(store->ctx()), store(store), obj(obj)
 363 {
 364   set_description() << "remove dest=" << obj;
 365 }
 366
 367 int RGWRadosRemoveCR::send_request()
 368 {
 369   auto rados = store->getRados()->get_rados_handle();
 370   int r = rados->ioctx_create(obj.pool.name.c_str(), ioctx);
 371   if (r < 0) {
 372     lderr(cct) << "ERROR: failed to open pool (" << obj.pool.name << ") ret=" << r << dendl;
 373     return r;
 374   }
 375   ioctx.locator_set_key(obj.loc);
 376
 377   set_status() << "send request";
 378
 379   librados::ObjectWriteOperation op;
 380   op.remove();
 381
 382   cn = stack->create_completion_notifier();
 383   return ioctx.aio_operate(obj.oid, cn->completion(), &op);
 384 }
 385
 386 int RGWRadosRemoveCR::request_complete()
 387 {
 388   int r = cn->completion()->get_return_value();
 389
 390   set_status() << "request complete; ret=" << r;
 391
 392   return r;
 393 }
 394
 395 RGWSimpleRadosLockCR::RGWSimpleRadosLockCR(RGWAsyncRadosProcessor *_async_rados, rgw::sal::RGWRadosStore *_store,
 396                       const rgw_raw_obj& _obj,
 397                       const string& _lock_name,
 398                       const string& _cookie,
 399                       uint32_t _duration) : RGWSimpleCoroutine(_store->ctx()),
 400                                                 async_rados(_async_rados),
 401                                                 store(_store),
 402                                                 lock_name(_lock_name),
 403                                                 cookie(_cookie),
 404                                                 duration(_duration),
 405                                                 obj(_obj),
 406                                                 req(NULL)
 407 {
 408   set_description() << "rados lock dest=" << obj << " lock=" << lock_name << " cookie=" << cookie << " duration=" << duration;
 409 }
 410
 411 void RGWSimpleRadosLockCR::request_cleanup()
 412 {
 413   if (req) {
 414     req->finish();
 415     req = NULL;
 416   }
 417 }
 418
 419 int RGWSimpleRadosLockCR::send_request()
 420 {
 421   set_status() << "sending request";
 422   req = new RGWAsyncLockSystemObj(this, stack->create_completion_notifier(),
 423                                  store, NULL, obj, lock_name, cookie, duration);
 424   async_rados->queue(req);
 425   return 0;
 426 }
 427
 428 int RGWSimpleRadosLockCR::request_complete()
 429 {
 430   set_status() << "request complete; ret=" << req->get_ret_status();
 431   return req->get_ret_status();
 432 }
 433
 434 RGWSimpleRadosUnlockCR::RGWSimpleRadosUnlockCR(RGWAsyncRadosProcessor *_async_rados, rgw::sal::RGWRadosStore *_store,
 435                       const rgw_raw_obj& _obj,
 436                       const string& _lock_name,
 437                       const string& _cookie) : RGWSimpleCoroutine(_store->ctx()),
 438                                                 async_rados(_async_rados),
 439                                                 store(_store),
 440                                                 lock_name(_lock_name),
 441                                                 cookie(_cookie),
 442                                                 obj(_obj),
 443                                                 req(NULL)
 444 {
 445   set_description() << "rados unlock dest=" << obj << " lock=" << lock_name << " cookie=" << cookie;
 446 }
 447
 448 void RGWSimpleRadosUnlockCR::request_cleanup()
 449 {
 450   if (req) {
 451     req->finish();
 452     req = NULL;
 453   }
 454 }
 455
 456 int RGWSimpleRadosUnlockCR::send_request()
 457 {
 458   set_status() << "sending request";
 459
 460   req = new RGWAsyncUnlockSystemObj(this, stack->create_completion_notifier(),
 461                                  store, NULL, obj, lock_name, cookie);
 462   async_rados->queue(req);
 463   return 0;
 464 }
 465
 466 int RGWSimpleRadosUnlockCR::request_complete()
 467 {
 468   set_status() << "request complete; ret=" << req->get_ret_status();
 469   return req->get_ret_status();
 470 }
 471
 472 int RGWOmapAppend::operate() {
 473   reenter(this) {
 474     for (;;) {
 475       if (!has_product() && going_down) {
 476         set_status() << "going down";
 477         break;
 478       }
 479       set_status() << "waiting for product";
 480       yield wait_for_product();
 481       yield {
 482         string entry;
 483         while (consume(&entry)) {
 484           set_status() << "adding entry: " << entry;
 485           entries[entry] = bufferlist();
 486           if (entries.size() >= window_size) {
 487             break;
 488           }
 489         }
 490         if (entries.size() >= window_size || going_down) {
 491           set_status() << "flushing to omap";
 492           call(new RGWRadosSetOmapKeysCR(store, obj, entries));
 493           entries.clear();
 494         }
 495       }
 496       if (get_ret_status() < 0) {
 497         ldout(cct, 0) << "ERROR: failed to store entries in omap" << dendl;
 498         return set_state(RGWCoroutine_Error);
 499       }
 500     }
 501     /* done with coroutine */
 502     return set_state(RGWCoroutine_Done);
 503   }
 504   return 0;
 505 }
 506
 507 void RGWOmapAppend::flush_pending() {
 508   receive(pending_entries);
 509   num_pending_entries = 0;
 510 }
 511
 512 bool RGWOmapAppend::append(const string& s) {
 513   if (is_done()) {
 514     return false;
 515   }
 516   ++total_entries;
 517   pending_entries.push_back(s);
 518   if (++num_pending_entries >= (int)window_size) {
 519     flush_pending();
 520   }
 521   return true;
 522 }
 523
 524 bool RGWOmapAppend::finish() {
 525   going_down = true;
 526   flush_pending();
 527   set_sleeping(false);
 528   return (!is_done());
 529 }
 530
 531 int RGWAsyncGetBucketInstanceInfo::_send_request()
 532 {
 533   int r;
 534   if (!bucket.bucket_id.empty()) {
 535     RGWSysObjectCtx obj_ctx = store->svc()->sysobj->init_obj_ctx();
 536     r = store->getRados()->get_bucket_instance_info(obj_ctx, bucket, bucket_info, nullptr, &attrs, null_yield);
 537   } else {
 538     r = store->ctl()->bucket->read_bucket_info(bucket, &bucket_info, null_yield,
 539                                                RGWBucketCtl::BucketInstance::GetParams().set_attrs(&attrs));
 540   }
 541   if (r < 0) {
 542     ldout(store->ctx(), 0) << "ERROR: failed to get bucket instance info for "
 543         << bucket << dendl;
 544     return r;
 545   }
 546
 547   return 0;
 548 }
 549
 550 RGWRadosBILogTrimCR::RGWRadosBILogTrimCR(rgw::sal::RGWRadosStore *store,
 551                                          const RGWBucketInfo& bucket_info,
 552                                          int shard_id,
 553                                          const std::string& start_marker,
 554                                          const std::string& end_marker)
 555   : RGWSimpleCoroutine(store->ctx()), bs(store->getRados()),
 556     start_marker(BucketIndexShardsManager::get_shard_marker(start_marker)),
 557     end_marker(BucketIndexShardsManager::get_shard_marker(end_marker))
 558 {
 559   bs.init(bucket_info, shard_id);
 560 }
 561
 562 int RGWRadosBILogTrimCR::send_request()
 563 {
 564   bufferlist in;
 565   cls_rgw_bi_log_trim_op call;
 566   call.start_marker = std::move(start_marker);
 567   call.end_marker = std::move(end_marker);
 568   encode(call, in);
 569
 570   librados::ObjectWriteOperation op;
 571   op.exec(RGW_CLASS, RGW_BI_LOG_TRIM, in);
 572
 573   cn = stack->create_completion_notifier();
 574   return bs.bucket_obj.aio_operate(cn->completion(), &op);
 575 }
 576
 577 int RGWRadosBILogTrimCR::request_complete()
 578 {
 579   int r = cn->completion()->get_return_value();
 580   set_status() << "request complete; ret=" << r;
 581   return r;
 582 }
 583
 584 int RGWAsyncFetchRemoteObj::_send_request()
 585 {
 586   RGWObjectCtx obj_ctx(store);
 587
 588   char buf[16];
 589   snprintf(buf, sizeof(buf), ".%lld", (long long)store->getRados()->instance_id());
 590   map<string, bufferlist> attrs;
 591
 592   rgw_obj src_obj(src_bucket, key);
 593
 594   rgw_obj dest_obj(dest_bucket_info.bucket, dest_key.value_or(key));
 595
 596   std::optional<uint64_t> bytes_transferred;
 597   int r = store->getRados()->fetch_remote_obj(obj_ctx,
 598                        user_id.value_or(rgw_user()),
 599                        NULL, /* req_info */
 600                        source_zone,
 601                        dest_obj,
 602                        src_obj,
 603                        dest_bucket_info, /* dest */
 604                        nullptr, /* source */
 605                        dest_placement_rule,
 606                        NULL, /* real_time* src_mtime, */
 607                        NULL, /* real_time* mtime, */
 608                        NULL, /* const real_time* mod_ptr, */
 609                        NULL, /* const real_time* unmod_ptr, */
 610                        false, /* high precision time */
 611                        NULL, /* const char *if_match, */
 612                        NULL, /* const char *if_nomatch, */
 613                        RGWRados::ATTRSMOD_NONE,
 614                        copy_if_newer,
 615                        attrs,
 616                        RGWObjCategory::Main,
 617                        versioned_epoch,
 618                        real_time(), /* delete_at */
 619                        NULL, /* string *ptag, */
 620                        NULL, /* string *petag, */
 621                        NULL, /* void (*progress_cb)(off_t, void *), */
 622                        NULL, /* void *progress_data*); */
 623                        dpp,
 624                        filter.get(),
 625                        &zones_trace,
 626                        &bytes_transferred);
 627
 628   if (r < 0) {
 629     ldout(store->ctx(), 0) << "store->fetch_remote_obj() returned r=" << r << dendl;
 630     if (counters) {
 631       counters->inc(sync_counters::l_fetch_err, 1);
 632     }
 633   } else if (counters) {
 634     if (bytes_transferred) {
 635       counters->inc(sync_counters::l_fetch, *bytes_transferred);
 636     } else {
 637       counters->inc(sync_counters::l_fetch_not_modified);
 638     }
 639   }
 640   return r;
 641 }
 642
 643 int RGWAsyncStatRemoteObj::_send_request()
 644 {
 645   RGWObjectCtx obj_ctx(store);
 646
 647   string user_id;
 648   char buf[16];
 649   snprintf(buf, sizeof(buf), ".%lld", (long long)store->getRados()->instance_id());
 650
 651   rgw_obj src_obj(src_bucket, key);
 652
 653   int r = store->getRados()->stat_remote_obj(obj_ctx,
 654                        rgw_user(user_id),
 655                        nullptr, /* req_info */
 656                        source_zone,
 657                        src_obj,
 658                        nullptr, /* source */
 659                        pmtime, /* real_time* src_mtime, */
 660                        psize, /* uint64_t * */
 661                        nullptr, /* const real_time* mod_ptr, */
 662                        nullptr, /* const real_time* unmod_ptr, */
 663                        true, /* high precision time */
 664                        nullptr, /* const char *if_match, */
 665                        nullptr, /* const char *if_nomatch, */
 666                        pattrs,
 667                        pheaders,
 668                        nullptr,
 669                        nullptr, /* string *ptag, */
 670                        petag); /* string *petag, */
 671
 672   if (r < 0) {
 673     ldout(store->ctx(), 0) << "store->fetch_remote_obj() returned r=" << r << dendl;
 674   }
 675   return r;
 676 }
 677
 678
 679 int RGWAsyncRemoveObj::_send_request()
 680 {
 681   RGWObjectCtx obj_ctx(store);
 682
 683   rgw_obj obj(bucket_info.bucket, key);
 684
 685   ldout(store->ctx(), 0) << __func__ << "(): deleting obj=" << obj << dendl;
 686
 687   obj_ctx.set_atomic(obj);
 688
 689   RGWObjState *state;
 690
 691   int ret = store->getRados()->get_obj_state(&obj_ctx, bucket_info, obj, &state, null_yield);
 692   if (ret < 0) {
 693     ldout(store->ctx(), 20) << __func__ << "(): get_obj_state() obj=" << obj << " returned ret=" << ret << dendl;
 694     return ret;
 695   }
 696
 697   /* has there been any racing object write? */
 698   if (del_if_older && (state->mtime > timestamp)) {
 699     ldout(store->ctx(), 20) << __func__ << "(): skipping object removal obj=" << obj << " (obj mtime=" << state->mtime << ", request timestamp=" << timestamp << ")" << dendl;
 700     return 0;
 701   }
 702
 703   RGWAccessControlPolicy policy;
 704
 705   /* decode policy */
 706   map<string, bufferlist>::iterator iter = state->attrset.find(RGW_ATTR_ACL);
 707   if (iter != state->attrset.end()) {
 708     auto bliter = iter->second.cbegin();
 709     try {
 710       policy.decode(bliter);
 711     } catch (buffer::error& err) {
 712       ldout(store->ctx(), 0) << "ERROR: could not decode policy, caught buffer::error" << dendl;
 713       return -EIO;
 714     }
 715   }
 716
 717   RGWRados::Object del_target(store->getRados(), bucket_info, obj_ctx, obj);
 718   RGWRados::Object::Delete del_op(&del_target);
 719
 720   del_op.params.bucket_owner = bucket_info.owner;
 721   del_op.params.obj_owner = policy.get_owner();
 722   if (del_if_older) {
 723     del_op.params.unmod_since = timestamp;
 724   }
 725   if (versioned) {
 726     del_op.params.versioning_status = BUCKET_VERSIONED;
 727   }
 728   del_op.params.olh_epoch = versioned_epoch;
 729   del_op.params.marker_version_id = marker_version_id;
 730   del_op.params.obj_owner.set_id(rgw_user(owner));
 731   del_op.params.obj_owner.set_name(owner_display_name);
 732   del_op.params.mtime = timestamp;
 733   del_op.params.high_precision_time = true;
 734   del_op.params.zones_trace = &zones_trace;
 735
 736   ret = del_op.delete_obj(null_yield);
 737   if (ret < 0) {
 738     ldout(store->ctx(), 20) << __func__ << "(): delete_obj() obj=" << obj << " returned ret=" << ret << dendl;
 739   }
 740   return ret;
 741 }
 742
 743 int RGWContinuousLeaseCR::operate()
 744 {
 745   if (aborted) {
 746     caller->set_sleeping(false);
 747     return set_cr_done();
 748   }
 749   reenter(this) {
 750     while (!going_down) {
 751       yield call(new RGWSimpleRadosLockCR(async_rados, store, obj, lock_name, cookie, interval));
 752
 753       caller->set_sleeping(false); /* will only be relevant when we return, that's why we can do it early */
 754       if (retcode < 0) {
 755         set_locked(false);
 756         ldout(store->ctx(), 20) << *this << ": couldn't lock " << obj << ":" << lock_name << ": retcode=" << retcode << dendl;
 757         return set_state(RGWCoroutine_Error, retcode);
 758       }
 759       set_locked(true);
 760       yield wait(utime_t(interval / 2, 0));
 761     }
 762     set_locked(false); /* moot at this point anyway */
 763     yield call(new RGWSimpleRadosUnlockCR(async_rados, store, obj, lock_name, cookie));
 764     return set_state(RGWCoroutine_Done);
 765   }
 766   return 0;
 767 }
 768
 769 RGWRadosTimelogAddCR::RGWRadosTimelogAddCR(rgw::sal::RGWRadosStore *_store, const string& _oid,
 770                       const cls_log_entry& entry) : RGWSimpleCoroutine(_store->ctx()),
 771                                                 store(_store),
 772                                                 oid(_oid), cn(NULL)
 773 {
 774   stringstream& s = set_description();
 775   s << "timelog add entry oid=" <<  oid << "entry={id=" << entry.id << ", section=" << entry.section << ", name=" << entry.name << "}";
 776   entries.push_back(entry);
 777 }
 778
 779 int RGWRadosTimelogAddCR::send_request()
 780 {
 781   set_status() << "sending request";
 782
 783   cn = stack->create_completion_notifier();
 784   return store->svc()->cls->timelog.add(oid, entries, cn->completion(), true, null_yield);
 785 }
 786
 787 int RGWRadosTimelogAddCR::request_complete()
 788 {
 789   int r = cn->completion()->get_return_value();
 790
 791   set_status() << "request complete; ret=" << r;
 792
 793   return r;
 794 }
 795
 796 RGWRadosTimelogTrimCR::RGWRadosTimelogTrimCR(rgw::sal::RGWRadosStore *store,
 797                                              const std::string& oid,
 798                                              const real_time& start_time,
 799                                              const real_time& end_time,
 800                                              const std::string& from_marker,
 801                                              const std::string& to_marker)
 802   : RGWSimpleCoroutine(store->ctx()), store(store), oid(oid),
 803     start_time(start_time), end_time(end_time),
 804     from_marker(from_marker), to_marker(to_marker)
 805 {
 806   set_description() << "timelog trim oid=" <<  oid
 807       << " start_time=" << start_time << " end_time=" << end_time
 808       << " from_marker=" << from_marker << " to_marker=" << to_marker;
 809 }
 810
 811 int RGWRadosTimelogTrimCR::send_request()
 812 {
 813   set_status() << "sending request";
 814
 815   cn = stack->create_completion_notifier();
 816   return store->svc()->cls->timelog.trim(oid, start_time, end_time, from_marker,
 817                                       to_marker, cn->completion(),
 818                                       null_yield);
 819 }
 820
 821 int RGWRadosTimelogTrimCR::request_complete()
 822 {
 823   int r = cn->completion()->get_return_value();
 824
 825   set_status() << "request complete; ret=" << r;
 826
 827   return r;
 828 }
 829
 830
 831 RGWSyncLogTrimCR::RGWSyncLogTrimCR(rgw::sal::RGWRadosStore *store, const std::string& oid,
 832                                    const std::string& to_marker,
 833                                    std::string *last_trim_marker)
 834   : RGWRadosTimelogTrimCR(store, oid, real_time{}, real_time{},
 835                           std::string{}, to_marker),
 836     cct(store->ctx()), last_trim_marker(last_trim_marker)
 837 {
 838 }
 839
 840 int RGWSyncLogTrimCR::request_complete()
 841 {
 842   int r = RGWRadosTimelogTrimCR::request_complete();
 843   if (r != -ENODATA) {
 844     return r;
 845   }
 846   // nothing left to trim, update last_trim_marker
 847   if (*last_trim_marker < to_marker && to_marker != max_marker) {
 848     *last_trim_marker = to_marker;
 849   }
 850   return 0;
 851 }
 852
 853
 854 int RGWAsyncStatObj::_send_request()
 855 {
 856   rgw_raw_obj raw_obj;
 857   store->getRados()->obj_to_raw(bucket_info.placement_rule, obj, &raw_obj);
 858   return store->getRados()->raw_obj_stat(raw_obj, psize, pmtime, pepoch,
 859                              nullptr, nullptr, objv_tracker, null_yield);
 860 }
 861
 862 RGWStatObjCR::RGWStatObjCR(RGWAsyncRadosProcessor *async_rados, rgw::sal::RGWRadosStore *store,
 863                            const RGWBucketInfo& _bucket_info, const rgw_obj& obj, uint64_t *psize,
 864                            real_time* pmtime, uint64_t *pepoch,
 865                            RGWObjVersionTracker *objv_tracker)
 866   : RGWSimpleCoroutine(store->ctx()), store(store), async_rados(async_rados),
 867     bucket_info(_bucket_info), obj(obj), psize(psize), pmtime(pmtime), pepoch(pepoch),
 868     objv_tracker(objv_tracker)
 869 {
 870 }
 871
 872 void RGWStatObjCR::request_cleanup()
 873 {
 874   if (req) {
 875     req->finish();
 876     req = NULL;
 877   }
 878 }
 879
 880 int RGWStatObjCR::send_request()
 881 {
 882   req = new RGWAsyncStatObj(this, stack->create_completion_notifier(),
 883                             store, bucket_info, obj, psize, pmtime, pepoch, objv_tracker);
 884   async_rados->queue(req);
 885   return 0;
 886 }
 887
 888 int RGWStatObjCR::request_complete()
 889 {
 890   return req->get_ret_status();
 891 }
 892
 893 RGWRadosNotifyCR::RGWRadosNotifyCR(rgw::sal::RGWRadosStore *store, const rgw_raw_obj& obj,
 894                                    bufferlist& request, uint64_t timeout_ms,
 895                                    bufferlist *response)
 896   : RGWSimpleCoroutine(store->ctx()), store(store), obj(obj),
 897     request(request), timeout_ms(timeout_ms), response(response)
 898 {
 899   set_description() << "notify dest=" << obj;
 900 }
 901
 902 int RGWRadosNotifyCR::send_request()
 903 {
 904   int r = store->getRados()->get_raw_obj_ref(obj, &ref);
 905   if (r < 0) {
 906     lderr(store->ctx()) << "ERROR: failed to get ref for (" << obj << ") ret=" << r << dendl;
 907     return r;
 908   }
 909
 910   set_status() << "sending request";
 911
 912   cn = stack->create_completion_notifier();
 913   return ref.pool.ioctx().aio_notify(ref.obj.oid, cn->completion(), request,
 914                               timeout_ms, response);
 915 }
 916
 917 int RGWRadosNotifyCR::request_complete()
 918 {
 919   int r = cn->completion()->get_return_value();
 920
 921   set_status() << "request complete; ret=" << r;
 922
 923   return r;
 924 }