--- /dev/null
+#ifndef CEPH_RGW_COROUTINE_H
+#define CEPH_RGW_COROUTINE_H
+
+#ifdef _ASSERT_H
+#define NEED_ASSERT_H
+#pragma push_macro("_ASSERT_H")
+#endif
+
+#include <boost/asio.hpp>
+#include <boost/intrusive_ptr.hpp>
+
+#ifdef NEED_ASSERT_H
+#pragma pop_macro("_ASSERT_H")
+#endif
+
+#include "include/utime.h"
+#include "common/RefCountedObj.h"
+#include "common/debug.h"
+#include "common/Timer.h"
+#include "common/admin_socket.h"
+
+#include "rgw_common.h"
+#include <boost/asio/coroutine.hpp>
+
+#include <atomic>
+
+#define RGW_ASYNC_OPS_MGR_WINDOW 100
+
+class RGWCoroutinesStack;
+class RGWCoroutinesManager;
+class RGWAioCompletionNotifier;
+
+class RGWCompletionManager : public RefCountedObject {
+ CephContext *cct;
+ list<void *> complete_reqs;
+ using NotifierRef = boost::intrusive_ptr<RGWAioCompletionNotifier>;
+ set<NotifierRef> cns;
+
+ Mutex lock;
+ Cond cond;
+
+ SafeTimer timer;
+
+ std::atomic<bool> going_down = { false };
+
+ map<void *, void *> waiters;
+
+ class WaitContext;
+
+protected:
+ void _wakeup(void *opaque);
+ void _complete(RGWAioCompletionNotifier *cn, void *user_info);
+public:
+ RGWCompletionManager(CephContext *_cct);
+ ~RGWCompletionManager() override;
+
+ void complete(RGWAioCompletionNotifier *cn, void *user_info);
+ int get_next(void **user_info);
+ bool try_get_next(void **user_info);
+
+ void go_down();
+
+ /*
+ * wait for interval length to complete user_info
+ */
+ void wait_interval(void *opaque, const utime_t& interval, void *user_info);
+ void wakeup(void *opaque);
+
+ void register_completion_notifier(RGWAioCompletionNotifier *cn);
+ void unregister_completion_notifier(RGWAioCompletionNotifier *cn);
+};
+
+/* a single use librados aio completion notifier that hooks into the RGWCompletionManager */
+class RGWAioCompletionNotifier : public RefCountedObject {
+ librados::AioCompletion *c;
+ RGWCompletionManager *completion_mgr;
+ void *user_data;
+ Mutex lock;
+ bool registered;
+
+public:
+ RGWAioCompletionNotifier(RGWCompletionManager *_mgr, void *_user_data);
+ ~RGWAioCompletionNotifier() override {
+ c->release();
+ lock.Lock();
+ bool need_unregister = registered;
+ if (registered) {
+ completion_mgr->get();
+ }
+ registered = false;
+ lock.Unlock();
+ if (need_unregister) {
+ completion_mgr->unregister_completion_notifier(this);
+ completion_mgr->put();
+ }
+ }
+
+ librados::AioCompletion *completion() {
+ return c;
+ }
+
+ void unregister() {
+ Mutex::Locker l(lock);
+ if (!registered) {
+ return;
+ }
+ registered = false;
+ }
+
+ void cb() {
+ lock.Lock();
+ if (!registered) {
+ lock.Unlock();
+ put();
+ return;
+ }
+ completion_mgr->get();
+ registered = false;
+ lock.Unlock();
+ completion_mgr->complete(this, user_data);
+ completion_mgr->put();
+ put();
+ }
+};
+
+struct RGWCoroutinesEnv {
+ uint64_t run_context;
+ RGWCoroutinesManager *manager;
+ list<RGWCoroutinesStack *> *scheduled_stacks;
+ RGWCoroutinesStack *stack;
+
+ RGWCoroutinesEnv() : run_context(0), manager(NULL), scheduled_stacks(NULL), stack(NULL) {}
+};
+
+enum RGWCoroutineState {
+ RGWCoroutine_Error = -2,
+ RGWCoroutine_Done = -1,
+ RGWCoroutine_Run = 0,
+};
+
+struct rgw_spawned_stacks {
+ vector<RGWCoroutinesStack *> entries;
+
+ rgw_spawned_stacks() {}
+
+ void add_pending(RGWCoroutinesStack *s) {
+ entries.push_back(s);
+ }
+
+ void inherit(rgw_spawned_stacks *source) {
+ for (vector<RGWCoroutinesStack *>::iterator iter = source->entries.begin();
+ iter != source->entries.end(); ++iter) {
+ add_pending(*iter);
+ }
+ source->entries.clear();
+ }
+};
+
+
+
+class RGWCoroutine : public RefCountedObject, public boost::asio::coroutine {
+ friend class RGWCoroutinesStack;
+
+ struct StatusItem {
+ utime_t timestamp;
+ string status;
+
+ StatusItem(utime_t& t, const string& s) : timestamp(t), status(s) {}
+
+ void dump(Formatter *f) const;
+ };
+
+#define MAX_COROUTINE_HISTORY 10
+
+ struct Status {
+ CephContext *cct;
+ RWLock lock;
+ int max_history;
+
+ utime_t timestamp;
+ stringstream status;
+
+ Status(CephContext *_cct) : cct(_cct), lock("RGWCoroutine::Status::lock"), max_history(MAX_COROUTINE_HISTORY) {}
+
+ deque<StatusItem> history;
+
+ stringstream& set_status();
+ } status;
+
+ stringstream description;
+
+protected:
+ bool _yield_ret;
+ boost::asio::coroutine drain_cr;
+
+ CephContext *cct;
+
+ RGWCoroutinesStack *stack;
+ int retcode;
+ int state;
+
+ rgw_spawned_stacks spawned;
+
+ stringstream error_stream;
+
+ int set_state(int s, int ret = 0) {
+ state = s;
+ return ret;
+ }
+ int set_cr_error(int ret) {
+ state = RGWCoroutine_Error;
+ return ret;
+ }
+ int set_cr_done() {
+ state = RGWCoroutine_Done;
+ return 0;
+ }
+ void set_io_blocked(bool flag);
+ int io_block(int ret = 0);
+
+ void reset_description() {
+ description.str(string());
+ }
+
+ stringstream& set_description() {
+ return description;
+ }
+ stringstream& set_status() {
+ return status.set_status();
+ }
+
+ stringstream& set_status(const string& s) {
+ stringstream& status = set_status();
+ status << s;
+ return status;
+ }
+
+public:
+ RGWCoroutine(CephContext *_cct) : status(_cct), _yield_ret(false), cct(_cct), stack(NULL), retcode(0), state(RGWCoroutine_Run) {}
+ ~RGWCoroutine() override;
+
+ virtual int operate() = 0;
+
+ bool is_done() { return (state == RGWCoroutine_Done || state == RGWCoroutine_Error); }
+ bool is_error() { return (state == RGWCoroutine_Error); }
+
+ stringstream& log_error() { return error_stream; }
+ string error_str() {
+ return error_stream.str();
+ }
+
+ void set_retcode(int r) {
+ retcode = r;
+ }
+
+ int get_ret_status() {
+ return retcode;
+ }
+
+ void call(RGWCoroutine *op); /* call at the same stack we're in */
+ RGWCoroutinesStack *spawn(RGWCoroutine *op, bool wait); /* execute on a different stack */
+ bool collect(int *ret, RGWCoroutinesStack *skip_stack); /* returns true if needs to be called again */
+ bool collect_next(int *ret, RGWCoroutinesStack **collected_stack = NULL); /* returns true if found a stack to collect */
+
+ int wait(const utime_t& interval);
+ bool drain_children(int num_cr_left, RGWCoroutinesStack *skip_stack = NULL); /* returns true if needed to be called again */
+ void wakeup();
+ void set_sleeping(bool flag); /* put in sleep, or wakeup from sleep */
+
+ size_t num_spawned() {
+ return spawned.entries.size();
+ }
+
+ void wait_for_child();
+
+ virtual string to_str() const;
+
+ RGWCoroutinesStack *get_stack() const {
+ return stack;
+ }
+
+ void dump(Formatter *f) const;
+};
+
+ostream& operator<<(ostream& out, const RGWCoroutine& cr);
+
+#define yield_until_true(x) \
+do { \
+ do { \
+ yield _yield_ret = x; \
+ } while (!_yield_ret); \
+ _yield_ret = false; \
+} while (0)
+
+#define drain_all() \
+ drain_cr = boost::asio::coroutine(); \
+ yield_until_true(drain_children(0))
+
+#define drain_all_but(n) \
+ drain_cr = boost::asio::coroutine(); \
+ yield_until_true(drain_children(n))
+
+#define drain_all_but_stack(stack) \
+ drain_cr = boost::asio::coroutine(); \
+ yield_until_true(drain_children(1, stack))
+
+template <class T>
+class RGWConsumerCR : public RGWCoroutine {
+ list<T> product;
+
+public:
+ RGWConsumerCR(CephContext *_cct) : RGWCoroutine(_cct) {}
+
+ bool has_product() {
+ return !product.empty();
+ }
+
+ void wait_for_product() {
+ if (!has_product()) {
+ set_sleeping(true);
+ }
+ }
+
+ bool consume(T *p) {
+ if (product.empty()) {
+ return false;
+ }
+ *p = product.front();
+ product.pop_front();
+ return true;
+ }
+
+ void receive(const T& p, bool wakeup = true);
+ void receive(list<T>& l, bool wakeup = true);
+};
+
+class RGWCoroutinesStack : public RefCountedObject {
+ friend class RGWCoroutine;
+ friend class RGWCoroutinesManager;
+
+ CephContext *cct;
+
+ RGWCoroutinesManager *ops_mgr;
+
+ list<RGWCoroutine *> ops;
+ list<RGWCoroutine *>::iterator pos;
+
+ rgw_spawned_stacks spawned;
+
+ set<RGWCoroutinesStack *> blocked_by_stack;
+ set<RGWCoroutinesStack *> blocking_stacks;
+
+ bool done_flag;
+ bool error_flag;
+ bool blocked_flag;
+ bool sleep_flag;
+ bool interval_wait_flag;
+
+ bool is_scheduled;
+
+ bool is_waiting_for_child;
+
+ int retcode;
+
+ uint64_t run_count;
+
+protected:
+ RGWCoroutinesEnv *env;
+ RGWCoroutinesStack *parent;
+
+ RGWCoroutinesStack *spawn(RGWCoroutine *source_op, RGWCoroutine *next_op, bool wait);
+ bool collect(RGWCoroutine *op, int *ret, RGWCoroutinesStack *skip_stack); /* returns true if needs to be called again */
+ bool collect_next(RGWCoroutine *op, int *ret, RGWCoroutinesStack **collected_stack); /* returns true if found a stack to collect */
+public:
+ RGWCoroutinesStack(CephContext *_cct, RGWCoroutinesManager *_ops_mgr, RGWCoroutine *start = NULL);
+ ~RGWCoroutinesStack() override;
+
+ int operate(RGWCoroutinesEnv *env);
+
+ bool is_done() {
+ return done_flag;
+ }
+ bool is_error() {
+ return error_flag;
+ }
+ bool is_blocked_by_stack() {
+ return !blocked_by_stack.empty();
+ }
+ void set_io_blocked(bool flag) {
+ blocked_flag = flag;
+ }
+ bool is_io_blocked() {
+ return blocked_flag;
+ }
+ void set_interval_wait(bool flag) {
+ interval_wait_flag = flag;
+ }
+ bool is_interval_waiting() {
+ return interval_wait_flag;
+ }
+ void set_sleeping(bool flag) {
+ bool wakeup = sleep_flag & !flag;
+ sleep_flag = flag;
+ if (wakeup) {
+ schedule();
+ }
+ }
+ bool is_sleeping() {
+ return sleep_flag;
+ }
+ void set_is_scheduled(bool flag) {
+ is_scheduled = flag;
+ }
+
+ bool is_blocked() {
+ return is_blocked_by_stack() || is_sleeping() ||
+ is_io_blocked() || waiting_for_child() ;
+ }
+
+ void schedule(list<RGWCoroutinesStack *> *stacks = NULL) {
+ if (!stacks) {
+ stacks = env->scheduled_stacks;
+ }
+ if (!is_scheduled) {
+ stacks->push_back(this);
+ is_scheduled = true;
+ }
+ }
+
+ int get_ret_status() {
+ return retcode;
+ }
+
+ string error_str();
+
+ void call(RGWCoroutine *next_op);
+ RGWCoroutinesStack *spawn(RGWCoroutine *next_op, bool wait);
+ int unwind(int retcode);
+
+ int wait(const utime_t& interval);
+ void wakeup();
+
+ bool collect(int *ret, RGWCoroutinesStack *skip_stack); /* returns true if needs to be called again */
+
+ RGWAioCompletionNotifier *create_completion_notifier();
+ RGWCompletionManager *get_completion_mgr();
+
+ void set_blocked_by(RGWCoroutinesStack *s) {
+ blocked_by_stack.insert(s);
+ s->blocking_stacks.insert(this);
+ }
+
+ void set_wait_for_child(bool flag) {
+ is_waiting_for_child = flag;
+ }
+
+ bool waiting_for_child() {
+ return is_waiting_for_child;
+ }
+
+ bool unblock_stack(RGWCoroutinesStack **s);
+
+ RGWCoroutinesEnv *get_env() { return env; }
+
+ void dump(Formatter *f) const;
+};
+
+template <class T>
+void RGWConsumerCR<T>::receive(list<T>& l, bool wakeup)
+{
+ product.splice(product.end(), l);
+ if (wakeup) {
+ set_sleeping(false);
+ }
+}
+
+
+template <class T>
+void RGWConsumerCR<T>::receive(const T& p, bool wakeup)
+{
+ product.push_back(p);
+ if (wakeup) {
+ set_sleeping(false);
+ }
+}
+
+class RGWCoroutinesManagerRegistry : public RefCountedObject, public AdminSocketHook {
+ CephContext *cct;
+
+ set<RGWCoroutinesManager *> managers;
+ RWLock lock;
+
+ string admin_command;
+
+public:
+ RGWCoroutinesManagerRegistry(CephContext *_cct) : cct(_cct), lock("RGWCoroutinesRegistry::lock") {}
+ ~RGWCoroutinesManagerRegistry() override;
+
+ void add(RGWCoroutinesManager *mgr);
+ void remove(RGWCoroutinesManager *mgr);
+
+ int hook_to_admin_command(const string& command);
+ bool call(std::string command, cmdmap_t& cmdmap, std::string format,
+ bufferlist& out) override;
+
+ void dump(Formatter *f) const;
+};
+
+class RGWCoroutinesManager {
+ CephContext *cct;
+ std::atomic<bool> going_down = { false };
+
+ std::atomic<int64_t> run_context_count = { 0 };
+ map<uint64_t, set<RGWCoroutinesStack *> > run_contexts;
+
+ RWLock lock;
+
+ void handle_unblocked_stack(set<RGWCoroutinesStack *>& context_stacks, list<RGWCoroutinesStack *>& scheduled_stacks, RGWCoroutinesStack *stack, int *waiting_count);
+protected:
+ RGWCompletionManager *completion_mgr;
+ RGWCoroutinesManagerRegistry *cr_registry;
+
+ int ops_window;
+
+ string id;
+
+ void put_completion_notifier(RGWAioCompletionNotifier *cn);
+public:
+ RGWCoroutinesManager(CephContext *_cct, RGWCoroutinesManagerRegistry *_cr_registry) : cct(_cct), lock("RGWCoroutinesManager::lock"),
+ cr_registry(_cr_registry), ops_window(RGW_ASYNC_OPS_MGR_WINDOW) {
+ completion_mgr = new RGWCompletionManager(cct);
+ if (cr_registry) {
+ cr_registry->add(this);
+ }
+ }
+ virtual ~RGWCoroutinesManager() {
+ stop();
+ completion_mgr->put();
+ if (cr_registry) {
+ cr_registry->remove(this);
+ }
+ }
+
+ int run(list<RGWCoroutinesStack *>& ops);
+ int run(RGWCoroutine *op);
+ void stop() {
+ bool expected = false;
+ if (going_down.compare_exchange_strong(expected, true)) {
+ completion_mgr->go_down();
+ }
+ }
+
+ virtual void report_error(RGWCoroutinesStack *op);
+
+ RGWAioCompletionNotifier *create_completion_notifier(RGWCoroutinesStack *stack);
+ RGWCompletionManager *get_completion_mgr() { return completion_mgr; }
+
+ void schedule(RGWCoroutinesEnv *env, RGWCoroutinesStack *stack);
+ RGWCoroutinesStack *allocate_stack();
+
+ virtual string get_id();
+ void dump(Formatter *f) const;
+};
+
+class RGWSimpleCoroutine : public RGWCoroutine {
+ bool called_cleanup;
+
+ int operate() override;
+
+ int state_init();
+ int state_send_request();
+ int state_request_complete();
+ int state_all_complete();
+
+ void call_cleanup();
+
+public:
+ RGWSimpleCoroutine(CephContext *_cct) : RGWCoroutine(_cct), called_cleanup(false) {}
+ ~RGWSimpleCoroutine() override;
+
+ virtual int init() { return 0; }
+ virtual int send_request() = 0;
+ virtual int request_complete() = 0;
+ virtual int finish() { return 0; }
+ virtual void request_cleanup() {}
+};
+
+#endif
Code Review / stor4nfv.git / blobdiff