ceph/src/rgw/rgw_coroutine.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/Context.h"
   5 #include "common/ceph_json.h"
   6 #include "rgw_coroutine.h"
   7
   8 // re-include our assert to clobber the system one; fix dout:
   9 #include "include/ceph_assert.h"
  10
  11 #include <boost/asio/yield.hpp>
  12
  13 #define dout_subsys ceph_subsys_rgw
  14 #define dout_context g_ceph_context
  15
  16
  17 class RGWCompletionManager::WaitContext : public Context {
  18   RGWCompletionManager *manager;
  19   void *opaque;
  20 public:
  21   WaitContext(RGWCompletionManager *_cm, void *_opaque) : manager(_cm), opaque(_opaque) {}
  22   void finish(int r) override {
  23     manager->_wakeup(opaque);
  24   }
  25 };
  26
  27 RGWCompletionManager::RGWCompletionManager(CephContext *_cct) : cct(_cct),
  28                                             timer(cct, lock)
  29 {
  30   timer.init();
  31 }
  32
  33 RGWCompletionManager::~RGWCompletionManager()
  34 {
  35   std::lock_guard l{lock};
  36   timer.cancel_all_events();
  37   timer.shutdown();
  38 }
  39
  40 void RGWCompletionManager::complete(RGWAioCompletionNotifier *cn, const rgw_io_id& io_id, void *user_info)
  41 {
  42   std::lock_guard l{lock};
  43   _complete(cn, io_id, user_info);
  44 }
  45
  46 void RGWCompletionManager::register_completion_notifier(RGWAioCompletionNotifier *cn)
  47 {
  48   std::lock_guard l{lock};
  49   if (cn) {
  50     cns.insert(cn);
  51   }
  52 }
  53
  54 void RGWCompletionManager::unregister_completion_notifier(RGWAioCompletionNotifier *cn)
  55 {
  56   std::lock_guard l{lock};
  57   if (cn) {
  58     cns.erase(cn);
  59   }
  60 }
  61
  62 void RGWCompletionManager::_complete(RGWAioCompletionNotifier *cn, const rgw_io_id& io_id, void *user_info)
  63 {
  64   if (cn) {
  65     cns.erase(cn);
  66   }
  67
  68   if (complete_reqs_set.find(io_id) != complete_reqs_set.end()) {
  69     /* already have completion for this io_id, don't allow multiple completions for it */
  70     return;
  71   }
  72   complete_reqs.push_back(io_completion{io_id, user_info});
  73   cond.notify_all();
  74 }
  75
  76 int RGWCompletionManager::get_next(io_completion *io)
  77 {
  78   std::unique_lock l{lock};
  79   while (complete_reqs.empty()) {
  80     if (going_down) {
  81       return -ECANCELED;
  82     }
  83     cond.wait(l);
  84   }
  85   *io = complete_reqs.front();
  86   complete_reqs_set.erase(io->io_id);
  87   complete_reqs.pop_front();
  88   return 0;
  89 }
  90
  91 bool RGWCompletionManager::try_get_next(io_completion *io)
  92 {
  93   std::lock_guard l{lock};
  94   if (complete_reqs.empty()) {
  95     return false;
  96   }
  97   *io = complete_reqs.front();
  98   complete_reqs_set.erase(io->io_id);
  99   complete_reqs.pop_front();
 100   return true;
 101 }
 102
 103 void RGWCompletionManager::go_down()
 104 {
 105   std::lock_guard l{lock};
 106   for (auto cn : cns) {
 107     cn->unregister();
 108   }
 109   going_down = true;
 110   cond.notify_all();
 111 }
 112
 113 void RGWCompletionManager::wait_interval(void *opaque, const utime_t& interval, void *user_info)
 114 {
 115   std::lock_guard l{lock};
 116   ceph_assert(waiters.find(opaque) == waiters.end());
 117   waiters[opaque] = user_info;
 118   timer.add_event_after(interval, new WaitContext(this, opaque));
 119 }
 120
 121 void RGWCompletionManager::wakeup(void *opaque)
 122 {
 123   std::lock_guard l{lock};
 124   _wakeup(opaque);
 125 }
 126
 127 void RGWCompletionManager::_wakeup(void *opaque)
 128 {
 129   map<void *, void *>::iterator iter = waiters.find(opaque);
 130   if (iter != waiters.end()) {
 131     void *user_id = iter->second;
 132     waiters.erase(iter);
 133     _complete(NULL, rgw_io_id{0, -1} /* no IO id */, user_id);
 134   }
 135 }
 136
 137 RGWCoroutine::~RGWCoroutine() {
 138   for (auto stack : spawned.entries) {
 139     stack->put();
 140   }
 141 }
 142
 143 void RGWCoroutine::init_new_io(RGWIOProvider *io_provider)
 144 {
 145   stack->init_new_io(io_provider);
 146 }
 147
 148 void RGWCoroutine::set_io_blocked(bool flag) {
 149   stack->set_io_blocked(flag);
 150 }
 151
 152 void RGWCoroutine::set_sleeping(bool flag) {
 153   stack->set_sleeping(flag);
 154 }
 155
 156 int RGWCoroutine::io_block(int ret, int64_t io_id) {
 157   return io_block(ret, rgw_io_id{io_id, -1});
 158 }
 159
 160 int RGWCoroutine::io_block(int ret, const rgw_io_id& io_id) {
 161   if (stack->consume_io_finish(io_id)) {
 162     return 0;
 163   }
 164   set_io_blocked(true);
 165   stack->set_io_blocked_id(io_id);
 166   return ret;
 167 }
 168
 169 void RGWCoroutine::io_complete(const rgw_io_id& io_id) {
 170   stack->io_complete(io_id);
 171 }
 172
 173 void RGWCoroutine::StatusItem::dump(Formatter *f) const {
 174   ::encode_json("timestamp", timestamp, f);
 175   ::encode_json("status", status, f);
 176 }
 177
 178 stringstream& RGWCoroutine::Status::set_status()
 179 {
 180   std::unique_lock l{lock};
 181   string s = status.str();
 182   status.str(string());
 183   if (!timestamp.is_zero()) {
 184     history.push_back(StatusItem(timestamp, s));
 185   }
 186   if (history.size() > (size_t)max_history) {
 187     history.pop_front();
 188   }
 189   timestamp = ceph_clock_now();
 190
 191   return status;
 192 }
 193
 194 int64_t RGWCoroutinesManager::get_next_io_id()
 195 {
 196   return (int64_t)++max_io_id;
 197 }
 198
 199 RGWCoroutinesStack::RGWCoroutinesStack(CephContext *_cct, RGWCoroutinesManager *_ops_mgr, RGWCoroutine *start) : cct(_cct), ops_mgr(_ops_mgr),
 200                                                                                                          done_flag(false), error_flag(false), blocked_flag(false),
 201                                                                                                          sleep_flag(false), interval_wait_flag(false), is_scheduled(false), is_waiting_for_child(false),
 202                                                                                                          retcode(0), run_count(0),
 203                                                                                                          env(NULL), parent(NULL)
 204 {
 205   if (start) {
 206     ops.push_back(start);
 207   }
 208   pos = ops.begin();
 209 }
 210
 211 RGWCoroutinesStack::~RGWCoroutinesStack()
 212 {
 213   for (auto op : ops) {
 214     op->put();
 215   }
 216
 217   for (auto stack : spawned.entries) {
 218     stack->put();
 219   }
 220 }
 221
 222 int RGWCoroutinesStack::operate(RGWCoroutinesEnv *_env)
 223 {
 224   env = _env;
 225   RGWCoroutine *op = *pos;
 226   op->stack = this;
 227   ldout(cct, 20) << *op << ": operate()" << dendl;
 228   int r = op->operate_wrapper();
 229   if (r < 0) {
 230     ldout(cct, 20) << *op << ": operate() returned r=" << r << dendl;
 231   }
 232
 233   error_flag = op->is_error();
 234
 235   if (op->is_done()) {
 236     int op_retcode = r;
 237     r = unwind(op_retcode);
 238     op->put();
 239     done_flag = (pos == ops.end());
 240     blocked_flag &= !done_flag;
 241     if (done_flag) {
 242       retcode = op_retcode;
 243     }
 244     return r;
 245   }
 246
 247   /* should r ever be negative at this point? */
 248   ceph_assert(r >= 0);
 249
 250   return 0;
 251 }
 252
 253 string RGWCoroutinesStack::error_str()
 254 {
 255   if (pos != ops.end()) {
 256     return (*pos)->error_str();
 257   }
 258   return string();
 259 }
 260
 261 void RGWCoroutinesStack::call(RGWCoroutine *next_op) {
 262   if (!next_op) {
 263     return;
 264   }
 265   ops.push_back(next_op);
 266   if (pos != ops.end()) {
 267     ++pos;
 268   } else {
 269     pos = ops.begin();
 270   }
 271 }
 272
 273 void RGWCoroutinesStack::schedule()
 274 {
 275   env->manager->schedule(env, this);
 276 }
 277
 278 void RGWCoroutinesStack::_schedule()
 279 {
 280   env->manager->_schedule(env, this);
 281 }
 282
 283 RGWCoroutinesStack *RGWCoroutinesStack::spawn(RGWCoroutine *source_op, RGWCoroutine *op, bool wait)
 284 {
 285   if (!op) {
 286     return NULL;
 287   }
 288
 289   rgw_spawned_stacks *s = (source_op ? &source_op->spawned : &spawned);
 290
 291   RGWCoroutinesStack *stack = env->manager->allocate_stack();
 292   s->add_pending(stack);
 293   stack->parent = this;
 294
 295   stack->get(); /* we'll need to collect the stack */
 296   stack->call(op);
 297
 298   env->manager->schedule(env, stack);
 299
 300   if (wait) {
 301     set_blocked_by(stack);
 302   }
 303
 304   return stack;
 305 }
 306
 307 RGWCoroutinesStack *RGWCoroutinesStack::spawn(RGWCoroutine *op, bool wait)
 308 {
 309   return spawn(NULL, op, wait);
 310 }
 311
 312 int RGWCoroutinesStack::wait(const utime_t& interval)
 313 {
 314   RGWCompletionManager *completion_mgr = env->manager->get_completion_mgr();
 315   completion_mgr->wait_interval((void *)this, interval, (void *)this);
 316   set_io_blocked(true);
 317   set_interval_wait(true);
 318   return 0;
 319 }
 320
 321 void RGWCoroutinesStack::wakeup()
 322 {
 323   RGWCompletionManager *completion_mgr = env->manager->get_completion_mgr();
 324   completion_mgr->wakeup((void *)this);
 325 }
 326
 327 void RGWCoroutinesStack::io_complete(const rgw_io_id& io_id)
 328 {
 329   RGWCompletionManager *completion_mgr = env->manager->get_completion_mgr();
 330   completion_mgr->complete(nullptr, io_id, (void *)this);
 331 }
 332
 333 int RGWCoroutinesStack::unwind(int retcode)
 334 {
 335   rgw_spawned_stacks *src_spawned = &(*pos)->spawned;
 336
 337   if (pos == ops.begin()) {
 338     ldout(cct, 15) << "stack " << (void *)this << " end" << dendl;
 339     spawned.inherit(src_spawned);
 340     ops.clear();
 341     pos = ops.end();
 342     return retcode;
 343   }
 344
 345   --pos;
 346   ops.pop_back();
 347   RGWCoroutine *op = *pos;
 348   op->set_retcode(retcode);
 349   op->spawned.inherit(src_spawned);
 350   return 0;
 351 }
 352
 353 void RGWCoroutinesStack::cancel()
 354 {
 355   while (!ops.empty()) {
 356     RGWCoroutine *op = *pos;
 357     unwind(-ECANCELED);
 358     op->put();
 359   }
 360   put();
 361 }
 362
 363 bool RGWCoroutinesStack::collect(RGWCoroutine *op, int *ret, RGWCoroutinesStack *skip_stack) /* returns true if needs to be called again */
 364 {
 365   bool need_retry = false;
 366   rgw_spawned_stacks *s = (op ? &op->spawned : &spawned);
 367   *ret = 0;
 368   vector<RGWCoroutinesStack *> new_list;
 369
 370   for (vector<RGWCoroutinesStack *>::iterator iter = s->entries.begin(); iter != s->entries.end(); ++iter) {
 371     RGWCoroutinesStack *stack = *iter;
 372     if (stack == skip_stack || !stack->is_done()) {
 373       new_list.push_back(stack);
 374       if (!stack->is_done()) {
 375         ldout(cct, 20) << "collect(): s=" << (void *)this << " stack=" << (void *)stack << " is still running" << dendl;
 376       } else if (stack == skip_stack) {
 377         ldout(cct, 20) << "collect(): s=" << (void *)this << " stack=" << (void *)stack << " explicitly skipping stack" << dendl;
 378       }
 379       continue;
 380     }
 381     int r = stack->get_ret_status();
 382     stack->put();
 383     if (r < 0) {
 384       *ret = r;
 385       ldout(cct, 20) << "collect(): s=" << (void *)this << " stack=" << (void *)stack << " encountered error (r=" << r << "), skipping next stacks" << dendl;
 386       new_list.insert(new_list.end(), ++iter, s->entries.end());
 387       need_retry = (iter != s->entries.end());
 388       break;
 389     }
 390
 391     ldout(cct, 20) << "collect(): s=" << (void *)this << " stack=" << (void *)stack << " is complete" << dendl;
 392   }
 393
 394   s->entries.swap(new_list);
 395   return need_retry;
 396 }
 397
 398 bool RGWCoroutinesStack::collect_next(RGWCoroutine *op, int *ret, RGWCoroutinesStack **collected_stack) /* returns true if found a stack to collect */
 399 {
 400   rgw_spawned_stacks *s = (op ? &op->spawned : &spawned);
 401   *ret = 0;
 402
 403   if (collected_stack) {
 404     *collected_stack = NULL;
 405   }
 406
 407   for (vector<RGWCoroutinesStack *>::iterator iter = s->entries.begin(); iter != s->entries.end(); ++iter) {
 408     RGWCoroutinesStack *stack = *iter;
 409     if (!stack->is_done()) {
 410       continue;
 411     }
 412     int r = stack->get_ret_status();
 413     if (r < 0) {
 414       *ret = r;
 415     }
 416
 417     if (collected_stack) {
 418       *collected_stack = stack;
 419     }
 420     stack->put();
 421
 422     s->entries.erase(iter);
 423     return true;
 424   }
 425
 426   return false;
 427 }
 428
 429 bool RGWCoroutinesStack::collect(int *ret, RGWCoroutinesStack *skip_stack) /* returns true if needs to be called again */
 430 {
 431   return collect(NULL, ret, skip_stack);
 432 }
 433
 434 static void _aio_completion_notifier_cb(librados::completion_t cb, void *arg)
 435 {
 436   (static_cast<RGWAioCompletionNotifier *>(arg))->cb();
 437 }
 438
 439 RGWAioCompletionNotifier::RGWAioCompletionNotifier(RGWCompletionManager *_mgr, const rgw_io_id& _io_id, void *_user_data) : completion_mgr(_mgr),
 440                                                                          io_id(_io_id),
 441                                                                          user_data(_user_data), registered(true) {
 442   c = librados::Rados::aio_create_completion(this, _aio_completion_notifier_cb);
 443 }
 444
 445 RGWAioCompletionNotifier *RGWCoroutinesStack::create_completion_notifier()
 446 {
 447   return ops_mgr->create_completion_notifier(this);
 448 }
 449
 450 RGWCompletionManager *RGWCoroutinesStack::get_completion_mgr()
 451 {
 452   return ops_mgr->get_completion_mgr();
 453 }
 454
 455 bool RGWCoroutinesStack::unblock_stack(RGWCoroutinesStack **s)
 456 {
 457   if (blocking_stacks.empty()) {
 458     return false;
 459   }
 460
 461   set<RGWCoroutinesStack *>::iterator iter = blocking_stacks.begin();
 462   *s = *iter;
 463   blocking_stacks.erase(iter);
 464   (*s)->blocked_by_stack.erase(this);
 465
 466   return true;
 467 }
 468
 469 void RGWCoroutinesManager::report_error(RGWCoroutinesStack *op)
 470 {
 471   if (!op) {
 472     return;
 473   }
 474   string err = op->error_str();
 475   if (err.empty()) {
 476     return;
 477   }
 478   lderr(cct) << "ERROR: failed operation: " << op->error_str() << dendl;
 479 }
 480
 481 void RGWCoroutinesStack::dump(Formatter *f) const {
 482   stringstream ss;
 483   ss << (void *)this;
 484   ::encode_json("stack", ss.str(), f);
 485   ::encode_json("run_count", run_count, f);
 486   f->open_array_section("ops");
 487   for (auto& i : ops) {
 488     encode_json("op", *i, f);
 489   }
 490   f->close_section();
 491 }
 492
 493 void RGWCoroutinesStack::init_new_io(RGWIOProvider *io_provider)
 494 {
 495   io_provider->set_io_user_info((void *)this);
 496   io_provider->assign_io(env->manager->get_io_id_provider());
 497 }
 498
 499 bool RGWCoroutinesStack::try_io_unblock(const rgw_io_id& io_id)
 500 {
 501   if (!can_io_unblock(io_id)) {
 502     auto p = io_finish_ids.emplace(io_id.id, io_id);
 503     auto& iter = p.first;
 504     bool inserted = p.second;
 505     if (!inserted) { /* could not insert, entry already existed, add channel to completion mask */
 506       iter->second.channels |= io_id.channels;
 507     }
 508     return false;
 509   }
 510
 511   return true;
 512 }
 513
 514 bool RGWCoroutinesStack::consume_io_finish(const rgw_io_id& io_id)
 515 {
 516   auto iter = io_finish_ids.find(io_id.id);
 517   if (iter == io_finish_ids.end()) {
 518     return false;
 519   }
 520   int finish_mask = iter->second.channels;
 521   bool found = (finish_mask & io_id.channels) != 0;
 522
 523   finish_mask &= ~(finish_mask & io_id.channels);
 524
 525   if (finish_mask == 0) {
 526     io_finish_ids.erase(iter);
 527   }
 528   return found;
 529 }
 530
 531
 532 void RGWCoroutinesManager::handle_unblocked_stack(set<RGWCoroutinesStack *>& context_stacks, list<RGWCoroutinesStack *>& scheduled_stacks,
 533                                                   RGWCompletionManager::io_completion& io, int *blocked_count)
 534 {
 535   ceph_assert(ceph_mutex_is_wlocked(lock));
 536   RGWCoroutinesStack *stack = static_cast<RGWCoroutinesStack *>(io.user_info);
 537   if (context_stacks.find(stack) == context_stacks.end()) {
 538     return;
 539   }
 540   if (!stack->try_io_unblock(io.io_id)) {
 541     return;
 542   }
 543   if (stack->is_io_blocked()) {
 544     --(*blocked_count);
 545     stack->set_io_blocked(false);
 546   }
 547   stack->set_interval_wait(false);
 548   if (!stack->is_done()) {
 549     if (!stack->is_scheduled) {
 550       scheduled_stacks.push_back(stack);
 551       stack->set_is_scheduled(true);
 552     }
 553   } else {
 554     context_stacks.erase(stack);
 555     stack->put();
 556   }
 557 }
 558
 559 void RGWCoroutinesManager::schedule(RGWCoroutinesEnv *env, RGWCoroutinesStack *stack)
 560 {
 561   std::unique_lock wl{lock};
 562   _schedule(env, stack);
 563 }
 564
 565 void RGWCoroutinesManager::_schedule(RGWCoroutinesEnv *env, RGWCoroutinesStack *stack)
 566 {
 567   ceph_assert(ceph_mutex_is_wlocked(lock));
 568   if (!stack->is_scheduled) {
 569     env->scheduled_stacks->push_back(stack);
 570     stack->set_is_scheduled(true);
 571   }
 572   set<RGWCoroutinesStack *>& context_stacks = run_contexts[env->run_context];
 573   context_stacks.insert(stack);
 574 }
 575
 576 void RGWCoroutinesManager::set_sleeping(RGWCoroutine *cr, bool flag)
 577 {
 578   cr->set_sleeping(flag);
 579 }
 580
 581 void RGWCoroutinesManager::io_complete(RGWCoroutine *cr, const rgw_io_id& io_id)
 582 {
 583   cr->io_complete(io_id);
 584 }
 585
 586 int RGWCoroutinesManager::run(list<RGWCoroutinesStack *>& stacks)
 587 {
 588   int ret = 0;
 589   int blocked_count = 0;
 590   int interval_wait_count = 0;
 591   bool canceled = false; // set on going_down
 592   RGWCoroutinesEnv env;
 593   bool op_not_blocked;
 594
 595   uint64_t run_context = ++run_context_count;
 596
 597   lock.lock();
 598   set<RGWCoroutinesStack *>& context_stacks = run_contexts[run_context];
 599   list<RGWCoroutinesStack *> scheduled_stacks;
 600   for (auto& st : stacks) {
 601     context_stacks.insert(st);
 602     scheduled_stacks.push_back(st);
 603     st->set_is_scheduled(true);
 604   }
 605   env.run_context = run_context;
 606   env.manager = this;
 607   env.scheduled_stacks = &scheduled_stacks;
 608
 609   for (list<RGWCoroutinesStack *>::iterator iter = scheduled_stacks.begin(); iter != scheduled_stacks.end() && !going_down;) {
 610     RGWCompletionManager::io_completion io;
 611     RGWCoroutinesStack *stack = *iter;
 612     ++iter;
 613     scheduled_stacks.pop_front();
 614
 615     if (context_stacks.find(stack) == context_stacks.end()) {
 616       /* stack was probably schedule more than once due to IO, but was since complete */
 617       goto next;
 618     }
 619     env.stack = stack;
 620
 621     lock.unlock();
 622
 623     ret = stack->operate(&env);
 624
 625     lock.lock();
 626
 627     stack->set_is_scheduled(false);
 628     if (ret < 0) {
 629       ldout(cct, 20) << "stack->operate() returned ret=" << ret << dendl;
 630     }
 631
 632     if (stack->is_error()) {
 633       report_error(stack);
 634     }
 635
 636     op_not_blocked = false;
 637
 638     if (stack->is_io_blocked()) {
 639       ldout(cct, 20) << __func__ << ":" << " stack=" << (void *)stack << " is io blocked" << dendl;
 640       if (stack->is_interval_waiting()) {
 641         interval_wait_count++;
 642       }
 643       blocked_count++;
 644     } else if (stack->is_blocked()) {
 645       /* do nothing, we'll re-add the stack when the blocking stack is done,
 646        * or when we're awaken
 647        */
 648       ldout(cct, 20) << __func__ << ":" << " stack=" << (void *)stack << " is_blocked_by_stack()=" << stack->is_blocked_by_stack()
 649                      << " is_sleeping=" << stack->is_sleeping() << " waiting_for_child()=" << stack->waiting_for_child() << dendl;
 650     } else if (stack->is_done()) {
 651       ldout(cct, 20) << __func__ << ":" << " stack=" << (void *)stack << " is done" << dendl;
 652       RGWCoroutinesStack *s;
 653       while (stack->unblock_stack(&s)) {
 654         if (!s->is_blocked_by_stack() && !s->is_done()) {
 655           if (s->is_io_blocked()) {
 656             if (stack->is_interval_waiting()) {
 657               interval_wait_count++;
 658             }
 659             blocked_count++;
 660           } else {
 661             s->_schedule();
 662           }
 663         }
 664       }
 665       if (stack->parent && stack->parent->waiting_for_child()) {
 666         stack->parent->set_wait_for_child(false);
 667         stack->parent->_schedule();
 668       }
 669       context_stacks.erase(stack);
 670       stack->put();
 671       stack = NULL;
 672     } else {
 673       op_not_blocked = true;
 674       stack->run_count++;
 675       stack->_schedule();
 676     }
 677
 678     if (!op_not_blocked && stack) {
 679       stack->run_count = 0;
 680     }
 681
 682     while (completion_mgr->try_get_next(&io)) {
 683       handle_unblocked_stack(context_stacks, scheduled_stacks, io, &blocked_count);
 684     }
 685
 686     /*
 687      * only account blocked operations that are not in interval_wait, these are stacks that
 688      * were put on a wait without any real IO operations. While we mark these as io_blocked,
 689      * these aren't really waiting for IOs
 690      */
 691     while (blocked_count - interval_wait_count >= ops_window) {
 692       lock.unlock();
 693       ret = completion_mgr->get_next(&io);
 694       lock.lock();
 695       if (ret < 0) {
 696        ldout(cct, 5) << "completion_mgr.get_next() returned ret=" << ret << dendl;
 697       }
 698       handle_unblocked_stack(context_stacks, scheduled_stacks, io, &blocked_count);
 699     }
 700
 701 next:
 702     while (scheduled_stacks.empty() && blocked_count > 0) {
 703       lock.unlock();
 704       ret = completion_mgr->get_next(&io);
 705       lock.lock();
 706       if (ret < 0) {
 707         ldout(cct, 5) << "completion_mgr.get_next() returned ret=" << ret << dendl;
 708       }
 709       if (going_down) {
 710         ldout(cct, 5) << __func__ << "(): was stopped, exiting" << dendl;
 711         ret = -ECANCELED;
 712         canceled = true;
 713         break;
 714       }
 715       handle_unblocked_stack(context_stacks, scheduled_stacks, io, &blocked_count);
 716       iter = scheduled_stacks.begin();
 717     }
 718     if (canceled) {
 719       break;
 720     }
 721
 722     if (iter == scheduled_stacks.end()) {
 723       iter = scheduled_stacks.begin();
 724     }
 725   }
 726
 727   if (!context_stacks.empty() && !going_down) {
 728     JSONFormatter formatter(true);
 729     formatter.open_array_section("context_stacks");
 730     for (auto& s : context_stacks) {
 731       ::encode_json("entry", *s, &formatter);
 732     }
 733     formatter.close_section();
 734     lderr(cct) << __func__ << "(): ERROR: deadlock detected, dumping remaining coroutines:\n";
 735     formatter.flush(*_dout);
 736     *_dout << dendl;
 737     ceph_assert(context_stacks.empty() || going_down); // assert on deadlock
 738   }
 739
 740   for (auto stack : context_stacks) {
 741     ldout(cct, 20) << "clearing stack on run() exit: stack=" << (void *)stack << " nref=" << stack->get_nref() << dendl;
 742     stack->cancel();
 743   }
 744   run_contexts.erase(run_context);
 745   lock.unlock();
 746
 747   return ret;
 748 }
 749
 750 int RGWCoroutinesManager::run(RGWCoroutine *op)
 751 {
 752   if (!op) {
 753     return 0;
 754   }
 755   list<RGWCoroutinesStack *> stacks;
 756   RGWCoroutinesStack *stack = allocate_stack();
 757   op->get();
 758   stack->call(op);
 759
 760   stacks.push_back(stack);
 761
 762   int r = run(stacks);
 763   if (r < 0) {
 764     ldout(cct, 20) << "run(stacks) returned r=" << r << dendl;
 765   } else {
 766     r = op->get_ret_status();
 767   }
 768   op->put();
 769
 770   return r;
 771 }
 772
 773 RGWAioCompletionNotifier *RGWCoroutinesManager::create_completion_notifier(RGWCoroutinesStack *stack)
 774 {
 775   rgw_io_id io_id{get_next_io_id(), -1};
 776   RGWAioCompletionNotifier *cn = new RGWAioCompletionNotifier(completion_mgr, io_id, (void *)stack);
 777   completion_mgr->register_completion_notifier(cn);
 778   return cn;
 779 }
 780
 781 void RGWCoroutinesManager::dump(Formatter *f) const {
 782   std::shared_lock rl{lock};
 783
 784   f->open_array_section("run_contexts");
 785   for (auto& i : run_contexts) {
 786     f->open_object_section("context");
 787     ::encode_json("id", i.first, f);
 788     f->open_array_section("entries");
 789     for (auto& s : i.second) {
 790       ::encode_json("entry", *s, f);
 791     }
 792     f->close_section();
 793     f->close_section();
 794   }
 795   f->close_section();
 796 }
 797
 798 RGWCoroutinesStack *RGWCoroutinesManager::allocate_stack() {
 799   return new RGWCoroutinesStack(cct, this);
 800 }
 801
 802 string RGWCoroutinesManager::get_id()
 803 {
 804   if (!id.empty()) {
 805     return id;
 806   }
 807   stringstream ss;
 808   ss << (void *)this;
 809   return ss.str();
 810 }
 811
 812 void RGWCoroutinesManagerRegistry::add(RGWCoroutinesManager *mgr)
 813 {
 814   std::unique_lock wl{lock};
 815   if (managers.find(mgr) == managers.end()) {
 816     managers.insert(mgr);
 817     get();
 818   }
 819 }
 820
 821 void RGWCoroutinesManagerRegistry::remove(RGWCoroutinesManager *mgr)
 822 {
 823   std::unique_lock wl{lock};
 824   if (managers.find(mgr) != managers.end()) {
 825     managers.erase(mgr);
 826     put();
 827   }
 828 }
 829
 830 RGWCoroutinesManagerRegistry::~RGWCoroutinesManagerRegistry()
 831 {
 832   AdminSocket *admin_socket = cct->get_admin_socket();
 833   if (!admin_command.empty()) {
 834     admin_socket->unregister_commands(this);
 835   }
 836 }
 837
 838 int RGWCoroutinesManagerRegistry::hook_to_admin_command(const string& command)
 839 {
 840   AdminSocket *admin_socket = cct->get_admin_socket();
 841   if (!admin_command.empty()) {
 842     admin_socket->unregister_commands(this);
 843   }
 844   admin_command = command;
 845   int r = admin_socket->register_command(admin_command, this,
 846                                      "dump current coroutines stack state");
 847   if (r < 0) {
 848     lderr(cct) << "ERROR: fail to register admin socket command (r=" << r << ")" << dendl;
 849     return r;
 850   }
 851   return 0;
 852 }
 853
 854 int RGWCoroutinesManagerRegistry::call(std::string_view command,
 855                                        const cmdmap_t& cmdmap,
 856                                        Formatter *f,
 857                                        std::ostream& ss,
 858                                        bufferlist& out) {
 859   std::shared_lock rl{lock};
 860   ::encode_json("cr_managers", *this, f);
 861   return 0;
 862 }
 863
 864 void RGWCoroutinesManagerRegistry::dump(Formatter *f) const {
 865   f->open_array_section("coroutine_managers");
 866   for (auto m : managers) {
 867     ::encode_json("entry", *m, f);
 868   }
 869   f->close_section();
 870 }
 871
 872 void RGWCoroutine::call(RGWCoroutine *op)
 873 {
 874   if (op) {
 875     stack->call(op);
 876   } else {
 877     // the call()er expects this to set a retcode
 878     retcode = 0;
 879   }
 880 }
 881
 882 RGWCoroutinesStack *RGWCoroutine::spawn(RGWCoroutine *op, bool wait)
 883 {
 884   return stack->spawn(this, op, wait);
 885 }
 886
 887 bool RGWCoroutine::collect(int *ret, RGWCoroutinesStack *skip_stack) /* returns true if needs to be called again */
 888 {
 889   return stack->collect(this, ret, skip_stack);
 890 }
 891
 892 bool RGWCoroutine::collect_next(int *ret, RGWCoroutinesStack **collected_stack) /* returns true if found a stack to collect */
 893 {
 894   return stack->collect_next(this, ret, collected_stack);
 895 }
 896
 897 int RGWCoroutine::wait(const utime_t& interval)
 898 {
 899   return stack->wait(interval);
 900 }
 901
 902 void RGWCoroutine::wait_for_child()
 903 {
 904   /* should only wait for child if there is a child that is not done yet, and no complete children */
 905   if (spawned.entries.empty()) {
 906     return;
 907   }
 908   for (vector<RGWCoroutinesStack *>::iterator iter = spawned.entries.begin(); iter != spawned.entries.end(); ++iter) {
 909     if ((*iter)->is_done()) {
 910       return;
 911     }
 912   }
 913   stack->set_wait_for_child(true);
 914 }
 915
 916 string RGWCoroutine::to_str() const
 917 {
 918   return typeid(*this).name();
 919 }
 920
 921 ostream& operator<<(ostream& out, const RGWCoroutine& cr)
 922 {
 923   out << "cr:s=" << (void *)cr.get_stack() << ":op=" << (void *)&cr << ":" << typeid(cr).name();
 924   return out;
 925 }
 926
 927 bool RGWCoroutine::drain_children(int num_cr_left, RGWCoroutinesStack *skip_stack)
 928 {
 929   bool done = false;
 930   ceph_assert(num_cr_left >= 0);
 931   if (num_cr_left == 0 && skip_stack) {
 932     num_cr_left = 1;
 933   }
 934   reenter(&drain_cr) {
 935     while (num_spawned() > (size_t)num_cr_left) {
 936       yield wait_for_child();
 937       int ret;
 938       while (collect(&ret, skip_stack)) {
 939         if (ret < 0) {
 940           ldout(cct, 10) << "collect() returned ret=" << ret << dendl;
 941           /* we should have reported this error */
 942           log_error() << "ERROR: collect() returned error (ret=" << ret << ")";
 943         }
 944       }
 945     }
 946     done = true;
 947   }
 948   return done;
 949 }
 950
 951 void RGWCoroutine::wakeup()
 952 {
 953   stack->wakeup();
 954 }
 955
 956 RGWCoroutinesEnv *RGWCoroutine::get_env() const
 957 {
 958   return stack->get_env();
 959 }
 960
 961 void RGWCoroutine::dump(Formatter *f) const {
 962   if (!description.str().empty()) {
 963     encode_json("description", description.str(), f);
 964   }
 965   encode_json("type", to_str(), f);
 966   if (!spawned.entries.empty()) {
 967     f->open_array_section("spawned");
 968     for (auto& i : spawned.entries) {
 969       char buf[32];
 970       snprintf(buf, sizeof(buf), "%p", (void *)i);
 971       encode_json("stack", string(buf), f);
 972     }
 973     f->close_section();
 974   }
 975   if (!status.history.empty()) {
 976     encode_json("history", status.history, f);
 977   }
 978
 979   if (!status.status.str().empty()) {
 980     f->open_object_section("status");
 981     encode_json("status", status.status.str(), f);
 982     encode_json("timestamp", status.timestamp, f);
 983     f->close_section();
 984   }
 985 }
 986
 987 RGWSimpleCoroutine::~RGWSimpleCoroutine()
 988 {
 989   if (!called_cleanup) {
 990     request_cleanup();
 991   }
 992 }
 993
 994 void RGWSimpleCoroutine::call_cleanup()
 995 {
 996   called_cleanup = true;
 997   request_cleanup();
 998 }
 999
1000 int RGWSimpleCoroutine::operate()
1001 {
1002   int ret = 0;
1003   reenter(this) {
1004     yield return state_init();
1005     yield return state_send_request();
1006     yield return state_request_complete();
1007     yield return state_all_complete();
1008     drain_all();
1009     call_cleanup();
1010     return set_state(RGWCoroutine_Done, ret);
1011   }
1012   return 0;
1013 }
1014
1015 int RGWSimpleCoroutine::state_init()
1016 {
1017   int ret = init();
1018   if (ret < 0) {
1019     call_cleanup();
1020     return set_state(RGWCoroutine_Error, ret);
1021   }
1022   return 0;
1023 }
1024
1025 int RGWSimpleCoroutine::state_send_request()
1026 {
1027   int ret = send_request();
1028   if (ret < 0) {
1029     call_cleanup();
1030     return set_state(RGWCoroutine_Error, ret);
1031   }
1032   return io_block(0);
1033 }
1034
1035 int RGWSimpleCoroutine::state_request_complete()
1036 {
1037   int ret = request_complete();
1038   if (ret < 0) {
1039     call_cleanup();
1040     return set_state(RGWCoroutine_Error, ret);
1041   }
1042   return 0;
1043 }
1044
1045 int RGWSimpleCoroutine::state_all_complete()
1046 {
1047   int ret = finish();
1048   if (ret < 0) {
1049     call_cleanup();
1050     return set_state(RGWCoroutine_Error, ret);
1051   }
1052   return 0;
1053 }
1054
1055