ceph/src/rgw/rgw_coroutine.h

   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 #ifndef CEPH_RGW_COROUTINE_H
   5 #define CEPH_RGW_COROUTINE_H
   6
   7 #ifdef _ASSERT_H
   8 #define NEED_ASSERT_H
   9 #pragma push_macro("_ASSERT_H")
  10 #endif
  11
  12 #include <boost/asio.hpp>
  13 #include <boost/intrusive_ptr.hpp>
  14
  15 #ifdef NEED_ASSERT_H
  16 #pragma pop_macro("_ASSERT_H")
  17 #endif
  18
  19 #include "include/utime.h"
  20 #include "common/RefCountedObj.h"
  21 #include "common/debug.h"
  22 #include "common/Timer.h"
  23 #include "common/admin_socket.h"
  24 #include "common/RWLock.h"
  25
  26 #include "rgw_common.h"
  27 #include "rgw_http_client_types.h"
  28
  29 #include <boost/asio/coroutine.hpp>
  30
  31 #include <atomic>
  32
  33 #define RGW_ASYNC_OPS_MGR_WINDOW 100
  34
  35 class RGWCoroutinesStack;
  36 class RGWCoroutinesManager;
  37 class RGWAioCompletionNotifier;
  38
  39 class RGWCompletionManager : public RefCountedObject {
  40   friend class RGWCoroutinesManager;
  41
  42   CephContext *cct;
  43
  44   struct io_completion {
  45     rgw_io_id io_id;
  46     void *user_info;
  47   };
  48   list<io_completion> complete_reqs;
  49   set<rgw_io_id> complete_reqs_set;
  50   using NotifierRef = boost::intrusive_ptr<RGWAioCompletionNotifier>;
  51   set<NotifierRef> cns;
  52
  53   ceph::mutex lock = ceph::make_mutex("RGWCompletionManager::lock");
  54   ceph::condition_variable cond;
  55
  56   SafeTimer timer;
  57
  58   std::atomic<bool> going_down = { false };
  59
  60   map<void *, void *> waiters;
  61
  62   class WaitContext;
  63
  64 protected:
  65   void _wakeup(void *opaque);
  66   void _complete(RGWAioCompletionNotifier *cn, const rgw_io_id& io_id, void *user_info);
  67 public:
  68   explicit RGWCompletionManager(CephContext *_cct);
  69   ~RGWCompletionManager() override;
  70
  71   void complete(RGWAioCompletionNotifier *cn, const rgw_io_id& io_id, void *user_info);
  72   int get_next(io_completion *io);
  73   bool try_get_next(io_completion *io);
  74
  75   void go_down();
  76
  77   /*
  78    * wait for interval length to complete user_info
  79    */
  80   void wait_interval(void *opaque, const utime_t& interval, void *user_info);
  81   void wakeup(void *opaque);
  82
  83   void register_completion_notifier(RGWAioCompletionNotifier *cn);
  84   void unregister_completion_notifier(RGWAioCompletionNotifier *cn);
  85 };
  86
  87 /* a single use librados aio completion notifier that hooks into the RGWCompletionManager */
  88 class RGWAioCompletionNotifier : public RefCountedObject {
  89   librados::AioCompletion *c;
  90   RGWCompletionManager *completion_mgr;
  91   rgw_io_id io_id;
  92   void *user_data;
  93   ceph::mutex lock = ceph::make_mutex("RGWAioCompletionNotifier");
  94   bool registered;
  95
  96 public:
  97   RGWAioCompletionNotifier(RGWCompletionManager *_mgr, const rgw_io_id& _io_id, void *_user_data);
  98   ~RGWAioCompletionNotifier() override {
  99     c->release();
 100     lock.lock();
 101     bool need_unregister = registered;
 102     if (registered) {
 103       completion_mgr->get();
 104     }
 105     registered = false;
 106     lock.unlock();
 107     if (need_unregister) {
 108       completion_mgr->unregister_completion_notifier(this);
 109       completion_mgr->put();
 110     }
 111   }
 112
 113   librados::AioCompletion *completion() {
 114     return c;
 115   }
 116
 117   void unregister() {
 118     std::lock_guard l{lock};
 119     if (!registered) {
 120       return;
 121     }
 122     registered = false;
 123   }
 124
 125   void cb() {
 126     lock.lock();
 127     if (!registered) {
 128       lock.unlock();
 129       put();
 130       return;
 131     }
 132     completion_mgr->get();
 133     registered = false;
 134     lock.unlock();
 135     completion_mgr->complete(this, io_id, user_data);
 136     completion_mgr->put();
 137     put();
 138   }
 139 };
 140
 141 // completion notifier with opaque payload (ie a reference-counted pointer)
 142 template <typename T>
 143 class RGWAioCompletionNotifierWith : public RGWAioCompletionNotifier {
 144   T value;
 145 public:
 146   RGWAioCompletionNotifierWith(RGWCompletionManager *mgr,
 147                                const rgw_io_id& io_id, void *user_data,
 148                                T value)
 149     : RGWAioCompletionNotifier(mgr, io_id, user_data), value(std::move(value))
 150   {}
 151 };
 152
 153 struct RGWCoroutinesEnv {
 154   uint64_t run_context;
 155   RGWCoroutinesManager *manager;
 156   list<RGWCoroutinesStack *> *scheduled_stacks;
 157   RGWCoroutinesStack *stack;
 158
 159   RGWCoroutinesEnv() : run_context(0), manager(NULL), scheduled_stacks(NULL), stack(NULL) {}
 160 };
 161
 162 enum RGWCoroutineState {
 163   RGWCoroutine_Error = -2,
 164   RGWCoroutine_Done  = -1,
 165   RGWCoroutine_Run   =  0,
 166 };
 167
 168 struct rgw_spawned_stacks {
 169   vector<RGWCoroutinesStack *> entries;
 170
 171   rgw_spawned_stacks() {}
 172
 173   void add_pending(RGWCoroutinesStack *s) {
 174     entries.push_back(s);
 175   }
 176
 177   void inherit(rgw_spawned_stacks *source) {
 178     for (vector<RGWCoroutinesStack *>::iterator iter = source->entries.begin();
 179          iter != source->entries.end(); ++iter) {
 180       add_pending(*iter);
 181     }
 182     source->entries.clear();
 183   }
 184 };
 185
 186
 187
 188 class RGWCoroutine : public RefCountedObject, public boost::asio::coroutine {
 189   friend class RGWCoroutinesStack;
 190
 191   struct StatusItem {
 192     utime_t timestamp;
 193     string status;
 194
 195     StatusItem(utime_t& t, const string& s) : timestamp(t), status(s) {}
 196
 197     void dump(Formatter *f) const;
 198   };
 199
 200 #define MAX_COROUTINE_HISTORY 10
 201
 202   struct Status {
 203     CephContext *cct;
 204     ceph::shared_mutex lock =
 205       ceph::make_shared_mutex("RGWCoroutine::Status::lock");
 206     int max_history;
 207
 208     utime_t timestamp;
 209     stringstream status;
 210
 211     explicit Status(CephContext *_cct) : cct(_cct), max_history(MAX_COROUTINE_HISTORY) {}
 212
 213     deque<StatusItem> history;
 214
 215     stringstream& set_status();
 216   } status;
 217
 218   stringstream description;
 219
 220 protected:
 221   bool _yield_ret;
 222   boost::asio::coroutine drain_cr;
 223
 224   CephContext *cct;
 225
 226   RGWCoroutinesStack *stack;
 227   int retcode;
 228   int state;
 229
 230   rgw_spawned_stacks spawned;
 231
 232   stringstream error_stream;
 233
 234   int set_state(int s, int ret = 0) {
 235     retcode = ret;
 236     state = s;
 237     return ret;
 238   }
 239   int set_cr_error(int ret) {
 240     return set_state(RGWCoroutine_Error, ret);
 241   }
 242   int set_cr_done() {
 243     return set_state(RGWCoroutine_Done, 0);
 244   }
 245   void set_io_blocked(bool flag);
 246
 247   void reset_description() {
 248     description.str(string());
 249   }
 250
 251   stringstream& set_description() {
 252     return description;
 253   }
 254   stringstream& set_status() {
 255     return status.set_status();
 256   }
 257
 258   stringstream& set_status(const string& s) {
 259     stringstream& status = set_status();
 260     status << s;
 261     return status;
 262   }
 263
 264   virtual int operate_wrapper() {
 265     return operate();
 266   }
 267 public:
 268   RGWCoroutine(CephContext *_cct) : status(_cct), _yield_ret(false), cct(_cct), stack(NULL), retcode(0), state(RGWCoroutine_Run) {}
 269   ~RGWCoroutine() override;
 270
 271   virtual int operate() = 0;
 272
 273   bool is_done() { return (state == RGWCoroutine_Done || state == RGWCoroutine_Error); }
 274   bool is_error() { return (state == RGWCoroutine_Error); }
 275
 276   stringstream& log_error() { return error_stream; }
 277   string error_str() {
 278     return error_stream.str();
 279   }
 280
 281   void set_retcode(int r) {
 282     retcode = r;
 283   }
 284
 285   int get_ret_status() {
 286     return retcode;
 287   }
 288
 289   void call(RGWCoroutine *op); /* call at the same stack we're in */
 290   RGWCoroutinesStack *spawn(RGWCoroutine *op, bool wait); /* execute on a different stack */
 291   bool collect(int *ret, RGWCoroutinesStack *skip_stack); /* returns true if needs to be called again */
 292   bool collect_next(int *ret, RGWCoroutinesStack **collected_stack = NULL); /* returns true if found a stack to collect */
 293
 294   int wait(const utime_t& interval);
 295   bool drain_children(int num_cr_left, RGWCoroutinesStack *skip_stack = NULL); /* returns true if needed to be called again */
 296   void wakeup();
 297   void set_sleeping(bool flag); /* put in sleep, or wakeup from sleep */
 298
 299   size_t num_spawned() {
 300     return spawned.entries.size();
 301   }
 302
 303   void wait_for_child();
 304
 305   virtual string to_str() const;
 306
 307   RGWCoroutinesStack *get_stack() const {
 308     return stack;
 309   }
 310
 311   RGWCoroutinesEnv *get_env() const;
 312
 313   void dump(Formatter *f) const;
 314
 315   void init_new_io(RGWIOProvider *io_provider); /* only links the default io id */
 316
 317   int io_block(int ret = 0) {
 318     return io_block(ret, -1);
 319   }
 320   int io_block(int ret, int64_t io_id);
 321   int io_block(int ret, const rgw_io_id& io_id);
 322   void io_complete() {
 323     io_complete(rgw_io_id{});
 324   }
 325   void io_complete(const rgw_io_id& io_id);
 326 };
 327
 328 ostream& operator<<(ostream& out, const RGWCoroutine& cr);
 329
 330 #define yield_until_true(x)     \
 331 do {                            \
 332   do {                          \
 333     yield _yield_ret = x;       \
 334   } while (!_yield_ret);        \
 335   _yield_ret = false;           \
 336 } while (0)
 337
 338 #define drain_all() \
 339   drain_cr = boost::asio::coroutine(); \
 340   yield_until_true(drain_children(0))
 341
 342 #define drain_all_but(n) \
 343   drain_cr = boost::asio::coroutine(); \
 344   yield_until_true(drain_children(n))
 345
 346 #define drain_all_but_stack(stack) \
 347   drain_cr = boost::asio::coroutine(); \
 348   yield_until_true(drain_children(1, stack))
 349
 350 template <class T>
 351 class RGWConsumerCR : public RGWCoroutine {
 352   list<T> product;
 353
 354 public:
 355   explicit RGWConsumerCR(CephContext *_cct) : RGWCoroutine(_cct) {}
 356
 357   bool has_product() {
 358     return !product.empty();
 359   }
 360
 361   void wait_for_product() {
 362     if (!has_product()) {
 363       set_sleeping(true);
 364     }
 365   }
 366
 367   bool consume(T *p) {
 368     if (product.empty()) {
 369       return false;
 370     }
 371     *p = product.front();
 372     product.pop_front();
 373     return true;
 374   }
 375
 376   void receive(const T& p, bool wakeup = true);
 377   void receive(list<T>& l, bool wakeup = true);
 378 };
 379
 380 class RGWCoroutinesStack : public RefCountedObject {
 381   friend class RGWCoroutine;
 382   friend class RGWCoroutinesManager;
 383
 384   CephContext *cct;
 385
 386   RGWCoroutinesManager *ops_mgr;
 387
 388   list<RGWCoroutine *> ops;
 389   list<RGWCoroutine *>::iterator pos;
 390
 391   rgw_spawned_stacks spawned;
 392
 393   set<RGWCoroutinesStack *> blocked_by_stack;
 394   set<RGWCoroutinesStack *> blocking_stacks;
 395
 396   map<int64_t, rgw_io_id> io_finish_ids;
 397   rgw_io_id io_blocked_id;
 398
 399   bool done_flag;
 400   bool error_flag;
 401   bool blocked_flag;
 402   bool sleep_flag;
 403   bool interval_wait_flag;
 404
 405   bool is_scheduled;
 406
 407   bool is_waiting_for_child;
 408
 409   int retcode;
 410
 411   uint64_t run_count;
 412
 413 protected:
 414   RGWCoroutinesEnv *env;
 415   RGWCoroutinesStack *parent;
 416
 417   RGWCoroutinesStack *spawn(RGWCoroutine *source_op, RGWCoroutine *next_op, bool wait);
 418   bool collect(RGWCoroutine *op, int *ret, RGWCoroutinesStack *skip_stack); /* returns true if needs to be called again */
 419   bool collect_next(RGWCoroutine *op, int *ret, RGWCoroutinesStack **collected_stack); /* returns true if found a stack to collect */
 420 public:
 421   RGWCoroutinesStack(CephContext *_cct, RGWCoroutinesManager *_ops_mgr, RGWCoroutine *start = NULL);
 422   ~RGWCoroutinesStack() override;
 423
 424   int operate(RGWCoroutinesEnv *env);
 425
 426   bool is_done() {
 427     return done_flag;
 428   }
 429   bool is_error() {
 430     return error_flag;
 431   }
 432   bool is_blocked_by_stack() {
 433     return !blocked_by_stack.empty();
 434   }
 435   void set_io_blocked(bool flag) {
 436     blocked_flag = flag;
 437   }
 438   void set_io_blocked_id(const rgw_io_id& io_id) {
 439     io_blocked_id = io_id;
 440   }
 441   bool is_io_blocked() {
 442     return blocked_flag && !done_flag;
 443   }
 444   bool can_io_unblock(const rgw_io_id& io_id) {
 445     return ((io_blocked_id.id < 0) ||
 446             io_blocked_id.intersects(io_id));
 447   }
 448   bool try_io_unblock(const rgw_io_id& io_id);
 449   bool consume_io_finish(const rgw_io_id& io_id);
 450   void set_interval_wait(bool flag) {
 451     interval_wait_flag = flag;
 452   }
 453   bool is_interval_waiting() {
 454     return interval_wait_flag;
 455   }
 456   void set_sleeping(bool flag) {
 457     bool wakeup = sleep_flag & !flag;
 458     sleep_flag = flag;
 459     if (wakeup) {
 460       schedule();
 461     }
 462   }
 463   bool is_sleeping() {
 464     return sleep_flag;
 465   }
 466   void set_is_scheduled(bool flag) {
 467     is_scheduled = flag;
 468   }
 469
 470   bool is_blocked() {
 471     return is_blocked_by_stack() || is_sleeping() ||
 472           is_io_blocked() || waiting_for_child() ;
 473   }
 474
 475   void schedule();
 476   void _schedule();
 477
 478   int get_ret_status() {
 479     return retcode;
 480   }
 481
 482   string error_str();
 483
 484   void call(RGWCoroutine *next_op);
 485   RGWCoroutinesStack *spawn(RGWCoroutine *next_op, bool wait);
 486   int unwind(int retcode);
 487
 488   int wait(const utime_t& interval);
 489   void wakeup();
 490   void io_complete() {
 491     io_complete(rgw_io_id{});
 492   }
 493   void io_complete(const rgw_io_id& io_id);
 494
 495   bool collect(int *ret, RGWCoroutinesStack *skip_stack); /* returns true if needs to be called again */
 496
 497   void cancel();
 498
 499   RGWAioCompletionNotifier *create_completion_notifier();
 500   template <typename T>
 501   RGWAioCompletionNotifier *create_completion_notifier(T value);
 502   RGWCompletionManager *get_completion_mgr();
 503
 504   void set_blocked_by(RGWCoroutinesStack *s) {
 505     blocked_by_stack.insert(s);
 506     s->blocking_stacks.insert(this);
 507   }
 508
 509   void set_wait_for_child(bool flag) {
 510     is_waiting_for_child = flag;
 511   }
 512
 513   bool waiting_for_child() {
 514     return is_waiting_for_child;
 515   }
 516
 517   bool unblock_stack(RGWCoroutinesStack **s);
 518
 519   RGWCoroutinesEnv *get_env() const { return env; }
 520
 521   void dump(Formatter *f) const;
 522
 523   void init_new_io(RGWIOProvider *io_provider);
 524 };
 525
 526 template <class T>
 527 void RGWConsumerCR<T>::receive(list<T>& l, bool wakeup)
 528 {
 529   product.splice(product.end(), l);
 530   if (wakeup) {
 531     set_sleeping(false);
 532   }
 533 }
 534
 535
 536 template <class T>
 537 void RGWConsumerCR<T>::receive(const T& p, bool wakeup)
 538 {
 539   product.push_back(p);
 540   if (wakeup) {
 541     set_sleeping(false);
 542   }
 543 }
 544
 545 class RGWCoroutinesManagerRegistry : public RefCountedObject, public AdminSocketHook {
 546   CephContext *cct;
 547
 548   set<RGWCoroutinesManager *> managers;
 549   ceph::shared_mutex lock =
 550     ceph::make_shared_mutex("RGWCoroutinesRegistry::lock");
 551
 552   string admin_command;
 553
 554 public:
 555   explicit RGWCoroutinesManagerRegistry(CephContext *_cct) : cct(_cct) {}
 556   ~RGWCoroutinesManagerRegistry() override;
 557
 558   void add(RGWCoroutinesManager *mgr);
 559   void remove(RGWCoroutinesManager *mgr);
 560
 561   int hook_to_admin_command(const string& command);
 562   int call(std::string_view command, const cmdmap_t& cmdmap,
 563            Formatter *f,
 564            std::ostream& ss,
 565            bufferlist& out) override;
 566
 567   void dump(Formatter *f) const;
 568 };
 569
 570 class RGWCoroutinesManager {
 571   CephContext *cct;
 572   std::atomic<bool> going_down = { false };
 573
 574   std::atomic<int64_t> run_context_count = { 0 };
 575   map<uint64_t, set<RGWCoroutinesStack *> > run_contexts;
 576
 577   std::atomic<int64_t> max_io_id = { 0 };
 578
 579   mutable ceph::shared_mutex lock =
 580     ceph::make_shared_mutex("RGWCoroutinesManager::lock");
 581
 582   RGWIOIDProvider io_id_provider;
 583
 584   void handle_unblocked_stack(set<RGWCoroutinesStack *>& context_stacks, list<RGWCoroutinesStack *>& scheduled_stacks,
 585                               RGWCompletionManager::io_completion& io, int *waiting_count);
 586 protected:
 587   RGWCompletionManager *completion_mgr;
 588   RGWCoroutinesManagerRegistry *cr_registry;
 589
 590   int ops_window;
 591
 592   string id;
 593
 594   void put_completion_notifier(RGWAioCompletionNotifier *cn);
 595 public:
 596   RGWCoroutinesManager(CephContext *_cct, RGWCoroutinesManagerRegistry *_cr_registry) : cct(_cct),
 597                                                                                         cr_registry(_cr_registry), ops_window(RGW_ASYNC_OPS_MGR_WINDOW) {
 598     completion_mgr = new RGWCompletionManager(cct);
 599     if (cr_registry) {
 600       cr_registry->add(this);
 601     }
 602   }
 603   virtual ~RGWCoroutinesManager() {
 604     stop();
 605     completion_mgr->put();
 606     if (cr_registry) {
 607       cr_registry->remove(this);
 608     }
 609   }
 610
 611   int run(list<RGWCoroutinesStack *>& ops);
 612   int run(RGWCoroutine *op);
 613   void stop() {
 614     bool expected = false;
 615     if (going_down.compare_exchange_strong(expected, true)) {
 616       completion_mgr->go_down();
 617     }
 618   }
 619
 620   virtual void report_error(RGWCoroutinesStack *op);
 621
 622   RGWAioCompletionNotifier *create_completion_notifier(RGWCoroutinesStack *stack);
 623   template <typename T>
 624   RGWAioCompletionNotifier *create_completion_notifier(RGWCoroutinesStack *stack, T value);
 625   RGWCompletionManager *get_completion_mgr() { return completion_mgr; }
 626
 627   void schedule(RGWCoroutinesEnv *env, RGWCoroutinesStack *stack);
 628   void _schedule(RGWCoroutinesEnv *env, RGWCoroutinesStack *stack);
 629   RGWCoroutinesStack *allocate_stack();
 630
 631   int64_t get_next_io_id();
 632
 633   void set_sleeping(RGWCoroutine *cr, bool flag);
 634   void io_complete(RGWCoroutine *cr, const rgw_io_id& io_id);
 635
 636   virtual string get_id();
 637   void dump(Formatter *f) const;
 638
 639   RGWIOIDProvider& get_io_id_provider() {
 640     return io_id_provider;
 641   }
 642 };
 643
 644 template <typename T>
 645 RGWAioCompletionNotifier *RGWCoroutinesManager::create_completion_notifier(RGWCoroutinesStack *stack, T value)
 646 {
 647   rgw_io_id io_id{get_next_io_id(), -1};
 648   RGWAioCompletionNotifier *cn = new RGWAioCompletionNotifierWith<T>(completion_mgr, io_id, (void *)stack, std::move(value));
 649   completion_mgr->register_completion_notifier(cn);
 650   return cn;
 651 }
 652
 653 template <typename T>
 654 RGWAioCompletionNotifier *RGWCoroutinesStack::create_completion_notifier(T value)
 655 {
 656   return ops_mgr->create_completion_notifier(this, std::move(value));
 657 }
 658
 659 class RGWSimpleCoroutine : public RGWCoroutine {
 660   bool called_cleanup;
 661
 662   int operate() override;
 663
 664   int state_init();
 665   int state_send_request();
 666   int state_request_complete();
 667   int state_all_complete();
 668
 669   void call_cleanup();
 670
 671 public:
 672   RGWSimpleCoroutine(CephContext *_cct) : RGWCoroutine(_cct), called_cleanup(false) {}
 673   ~RGWSimpleCoroutine() override;
 674
 675   virtual int init() { return 0; }
 676   virtual int send_request() = 0;
 677   virtual int request_complete() = 0;
 678   virtual int finish() { return 0; }
 679   virtual void request_cleanup() {}
 680 };
 681
 682 #endif