+++ /dev/null
-
-#include "common/ceph_json.h"
-#include "rgw_coroutine.h"
-
-// re-include our assert to clobber the system one; fix dout:
-#include "include/assert.h"
-
-#include <boost/asio/yield.hpp>
-
-#define dout_subsys ceph_subsys_rgw
-
-
-class RGWCompletionManager::WaitContext : public Context {
- RGWCompletionManager *manager;
- void *opaque;
-public:
- WaitContext(RGWCompletionManager *_cm, void *_opaque) : manager(_cm), opaque(_opaque) {}
- void finish(int r) override {
- manager->_wakeup(opaque);
- }
-};
-
-RGWCompletionManager::RGWCompletionManager(CephContext *_cct) : cct(_cct), lock("RGWCompletionManager::lock"),
- timer(cct, lock)
-{
- timer.init();
-}
-
-RGWCompletionManager::~RGWCompletionManager()
-{
- Mutex::Locker l(lock);
- timer.cancel_all_events();
- timer.shutdown();
-}
-
-void RGWCompletionManager::complete(RGWAioCompletionNotifier *cn, void *user_info)
-{
- Mutex::Locker l(lock);
- _complete(cn, user_info);
-}
-
-void RGWCompletionManager::register_completion_notifier(RGWAioCompletionNotifier *cn)
-{
- Mutex::Locker l(lock);
- if (cn) {
- cns.insert(cn);
- }
-}
-
-void RGWCompletionManager::unregister_completion_notifier(RGWAioCompletionNotifier *cn)
-{
- Mutex::Locker l(lock);
- if (cn) {
- cns.erase(cn);
- }
-}
-
-void RGWCompletionManager::_complete(RGWAioCompletionNotifier *cn, void *user_info)
-{
- if (cn) {
- cns.erase(cn);
- }
- complete_reqs.push_back(user_info);
- cond.Signal();
-}
-
-int RGWCompletionManager::get_next(void **user_info)
-{
- Mutex::Locker l(lock);
- while (complete_reqs.empty()) {
- cond.Wait(lock);
- if (going_down) {
- return -ECANCELED;
- }
- }
- *user_info = complete_reqs.front();
- complete_reqs.pop_front();
- return 0;
-}
-
-bool RGWCompletionManager::try_get_next(void **user_info)
-{
- Mutex::Locker l(lock);
- if (complete_reqs.empty()) {
- return false;
- }
- *user_info = complete_reqs.front();
- complete_reqs.pop_front();
- return true;
-}
-
-void RGWCompletionManager::go_down()
-{
- Mutex::Locker l(lock);
- for (auto cn : cns) {
- cn->unregister();
- }
- going_down = true;
- cond.Signal();
-}
-
-void RGWCompletionManager::wait_interval(void *opaque, const utime_t& interval, void *user_info)
-{
- Mutex::Locker l(lock);
- assert(waiters.find(opaque) == waiters.end());
- waiters[opaque] = user_info;
- timer.add_event_after(interval, new WaitContext(this, opaque));
-}
-
-void RGWCompletionManager::wakeup(void *opaque)
-{
- Mutex::Locker l(lock);
- _wakeup(opaque);
-}
-
-void RGWCompletionManager::_wakeup(void *opaque)
-{
- map<void *, void *>::iterator iter = waiters.find(opaque);
- if (iter != waiters.end()) {
- void *user_id = iter->second;
- waiters.erase(iter);
- _complete(NULL, user_id);
- }
-}
-
-RGWCoroutine::~RGWCoroutine() {
- for (auto stack : spawned.entries) {
- stack->put();
- }
-}
-
-void RGWCoroutine::set_io_blocked(bool flag) {
- stack->set_io_blocked(flag);
-}
-
-void RGWCoroutine::set_sleeping(bool flag) {
- stack->set_sleeping(flag);
-}
-
-int RGWCoroutine::io_block(int ret) {
- set_io_blocked(true);
- return ret;
-}
-
-void RGWCoroutine::StatusItem::dump(Formatter *f) const {
- ::encode_json("timestamp", timestamp, f);
- ::encode_json("status", status, f);
-}
-
-stringstream& RGWCoroutine::Status::set_status()
-{
- RWLock::WLocker l(lock);
- string s = status.str();
- status.str(string());
- if (!timestamp.is_zero()) {
- history.push_back(StatusItem(timestamp, s));
- }
- if (history.size() > (size_t)max_history) {
- history.pop_front();
- }
- timestamp = ceph_clock_now();
-
- return status;
-}
-
-RGWCoroutinesStack::RGWCoroutinesStack(CephContext *_cct, RGWCoroutinesManager *_ops_mgr, RGWCoroutine *start) : cct(_cct), ops_mgr(_ops_mgr),
- done_flag(false), error_flag(false), blocked_flag(false),
- sleep_flag(false), interval_wait_flag(false), is_scheduled(false), is_waiting_for_child(false),
- retcode(0), run_count(0),
- env(NULL), parent(NULL)
-{
- if (start) {
- ops.push_back(start);
- }
- pos = ops.begin();
-}
-
-RGWCoroutinesStack::~RGWCoroutinesStack()
-{
- for (auto op : ops) {
- op->put();
- }
-
- for (auto stack : spawned.entries) {
- stack->put();
- }
-}
-
-int RGWCoroutinesStack::operate(RGWCoroutinesEnv *_env)
-{
- env = _env;
- RGWCoroutine *op = *pos;
- op->stack = this;
- ldout(cct, 20) << *op << ": operate()" << dendl;
- int r = op->operate();
- if (r < 0) {
- ldout(cct, 20) << *op << ": operate() returned r=" << r << dendl;
- }
-
- error_flag = op->is_error();
-
- if (op->is_done()) {
- int op_retcode = r;
- r = unwind(op_retcode);
- op->put();
- done_flag = (pos == ops.end());
- if (done_flag) {
- retcode = op_retcode;
- }
- return r;
- }
-
- /* should r ever be negative at this point? */
- assert(r >= 0);
-
- return 0;
-}
-
-string RGWCoroutinesStack::error_str()
-{
- if (pos != ops.end()) {
- return (*pos)->error_str();
- }
- return string();
-}
-
-void RGWCoroutinesStack::call(RGWCoroutine *next_op) {
- if (!next_op) {
- return;
- }
- ops.push_back(next_op);
- if (pos != ops.end()) {
- ++pos;
- } else {
- pos = ops.begin();
- }
-}
-
-RGWCoroutinesStack *RGWCoroutinesStack::spawn(RGWCoroutine *source_op, RGWCoroutine *op, bool wait)
-{
- if (!op) {
- return NULL;
- }
-
- rgw_spawned_stacks *s = (source_op ? &source_op->spawned : &spawned);
-
- RGWCoroutinesStack *stack = env->manager->allocate_stack();
- s->add_pending(stack);
- stack->parent = this;
-
- stack->get(); /* we'll need to collect the stack */
- stack->call(op);
-
- env->manager->schedule(env, stack);
-
- if (wait) {
- set_blocked_by(stack);
- }
-
- return stack;
-}
-
-RGWCoroutinesStack *RGWCoroutinesStack::spawn(RGWCoroutine *op, bool wait)
-{
- return spawn(NULL, op, wait);
-}
-
-int RGWCoroutinesStack::wait(const utime_t& interval)
-{
- RGWCompletionManager *completion_mgr = env->manager->get_completion_mgr();
- completion_mgr->wait_interval((void *)this, interval, (void *)this);
- set_io_blocked(true);
- set_interval_wait(true);
- return 0;
-}
-
-void RGWCoroutinesStack::wakeup()
-{
- RGWCompletionManager *completion_mgr = env->manager->get_completion_mgr();
- completion_mgr->wakeup((void *)this);
-}
-
-int RGWCoroutinesStack::unwind(int retcode)
-{
- rgw_spawned_stacks *src_spawned = &(*pos)->spawned;
-
- if (pos == ops.begin()) {
- spawned.inherit(src_spawned);
- ops.clear();
- pos = ops.end();
- return retcode;
- }
-
- --pos;
- ops.pop_back();
- RGWCoroutine *op = *pos;
- op->set_retcode(retcode);
- op->spawned.inherit(src_spawned);
- return 0;
-}
-
-
-bool RGWCoroutinesStack::collect(RGWCoroutine *op, int *ret, RGWCoroutinesStack *skip_stack) /* returns true if needs to be called again */
-{
- bool done = true;
- rgw_spawned_stacks *s = (op ? &op->spawned : &spawned);
- *ret = 0;
- vector<RGWCoroutinesStack *> new_list;
-
- for (vector<RGWCoroutinesStack *>::iterator iter = s->entries.begin(); iter != s->entries.end(); ++iter) {
- RGWCoroutinesStack *stack = *iter;
- if (stack == skip_stack || !stack->is_done()) {
- new_list.push_back(stack);
- if (!stack->is_done()) {
- ldout(cct, 20) << "collect(): s=" << (void *)this << " stack=" << (void *)stack << " is still running" << dendl;
- } else if (stack == skip_stack) {
- ldout(cct, 20) << "collect(): s=" << (void *)this << " stack=" << (void *)stack << " explicitly skipping stack" << dendl;
- }
- continue;
- }
- int r = stack->get_ret_status();
- stack->put();
- if (r < 0) {
- *ret = r;
- ldout(cct, 20) << "collect(): s=" << (void *)this << " stack=" << (void *)stack << " encountered error (r=" << r << "), skipping next stacks" << dendl;
- new_list.insert(new_list.end(), ++iter, s->entries.end());
- done &= (iter != s->entries.end());
- break;
- }
-
- ldout(cct, 20) << "collect(): s=" << (void *)this << " stack=" << (void *)stack << " is complete" << dendl;
- }
-
- s->entries.swap(new_list);
- return (!done);
-}
-
-bool RGWCoroutinesStack::collect_next(RGWCoroutine *op, int *ret, RGWCoroutinesStack **collected_stack) /* returns true if found a stack to collect */
-{
- rgw_spawned_stacks *s = (op ? &op->spawned : &spawned);
- *ret = 0;
-
- if (collected_stack) {
- *collected_stack = NULL;
- }
-
- for (vector<RGWCoroutinesStack *>::iterator iter = s->entries.begin(); iter != s->entries.end(); ++iter) {
- RGWCoroutinesStack *stack = *iter;
- if (!stack->is_done()) {
- continue;
- }
- int r = stack->get_ret_status();
- if (r < 0) {
- *ret = r;
- }
-
- if (collected_stack) {
- *collected_stack = stack;
- }
- stack->put();
-
- s->entries.erase(iter);
- return true;
- }
-
- return false;
-}
-
-bool RGWCoroutinesStack::collect(int *ret, RGWCoroutinesStack *skip_stack) /* returns true if needs to be called again */
-{
- return collect(NULL, ret, skip_stack);
-}
-
-static void _aio_completion_notifier_cb(librados::completion_t cb, void *arg)
-{
- ((RGWAioCompletionNotifier *)arg)->cb();
-}
-
-RGWAioCompletionNotifier::RGWAioCompletionNotifier(RGWCompletionManager *_mgr, void *_user_data) : completion_mgr(_mgr),
- user_data(_user_data), lock("RGWAioCompletionNotifier"), registered(true) {
- c = librados::Rados::aio_create_completion((void *)this, NULL,
- _aio_completion_notifier_cb);
-}
-
-RGWAioCompletionNotifier *RGWCoroutinesStack::create_completion_notifier()
-{
- return ops_mgr->create_completion_notifier(this);
-}
-
-RGWCompletionManager *RGWCoroutinesStack::get_completion_mgr()
-{
- return ops_mgr->get_completion_mgr();
-}
-
-bool RGWCoroutinesStack::unblock_stack(RGWCoroutinesStack **s)
-{
- if (blocking_stacks.empty()) {
- return false;
- }
-
- set<RGWCoroutinesStack *>::iterator iter = blocking_stacks.begin();
- *s = *iter;
- blocking_stacks.erase(iter);
- (*s)->blocked_by_stack.erase(this);
-
- return true;
-}
-
-void RGWCoroutinesManager::report_error(RGWCoroutinesStack *op)
-{
- if (!op) {
- return;
- }
- string err = op->error_str();
- if (err.empty()) {
- return;
- }
- lderr(cct) << "ERROR: failed operation: " << op->error_str() << dendl;
-}
-
-void RGWCoroutinesStack::dump(Formatter *f) const {
- stringstream ss;
- ss << (void *)this;
- ::encode_json("stack", ss.str(), f);
- ::encode_json("run_count", run_count, f);
- f->open_array_section("ops");
- for (auto& i : ops) {
- encode_json("op", *i, f);
- }
- f->close_section();
-}
-
-void RGWCoroutinesManager::handle_unblocked_stack(set<RGWCoroutinesStack *>& context_stacks, list<RGWCoroutinesStack *>& scheduled_stacks, RGWCoroutinesStack *stack, int *blocked_count)
-{
- RWLock::WLocker wl(lock);
- --(*blocked_count);
- stack->set_io_blocked(false);
- stack->set_interval_wait(false);
- if (!stack->is_done()) {
- scheduled_stacks.push_back(stack);
- } else {
- RWLock::WLocker wl(lock);
- context_stacks.erase(stack);
- stack->put();
- }
-}
-
-void RGWCoroutinesManager::schedule(RGWCoroutinesEnv *env, RGWCoroutinesStack *stack)
-{
- assert(lock.is_wlocked());
- env->scheduled_stacks->push_back(stack);
- set<RGWCoroutinesStack *>& context_stacks = run_contexts[env->run_context];
- context_stacks.insert(stack);
-}
-
-int RGWCoroutinesManager::run(list<RGWCoroutinesStack *>& stacks)
-{
- int ret = 0;
- int blocked_count = 0;
- int interval_wait_count = 0;
- bool canceled = false; // set on going_down
- RGWCoroutinesEnv env;
-
- uint64_t run_context = ++run_context_count;
-
- lock.get_write();
- set<RGWCoroutinesStack *>& context_stacks = run_contexts[run_context];
- list<RGWCoroutinesStack *> scheduled_stacks;
- for (auto& st : stacks) {
- context_stacks.insert(st);
- scheduled_stacks.push_back(st);
- }
- lock.unlock();
-
- env.run_context = run_context;
- env.manager = this;
- env.scheduled_stacks = &scheduled_stacks;
-
- for (list<RGWCoroutinesStack *>::iterator iter = scheduled_stacks.begin(); iter != scheduled_stacks.end() && !going_down;) {
- lock.get_write();
-
- RGWCoroutinesStack *stack = *iter;
- env.stack = stack;
-
- ret = stack->operate(&env);
- stack->set_is_scheduled(false);
- if (ret < 0) {
- ldout(cct, 20) << "stack->operate() returned ret=" << ret << dendl;
- }
-
- if (stack->is_error()) {
- report_error(stack);
- }
-
- bool op_not_blocked = false;
-
- if (stack->is_io_blocked()) {
- ldout(cct, 20) << __func__ << ":" << " stack=" << (void *)stack << " is io blocked" << dendl;
- if (stack->is_interval_waiting()) {
- interval_wait_count++;
- }
- blocked_count++;
- } else if (stack->is_blocked()) {
- /* do nothing, we'll re-add the stack when the blocking stack is done,
- * or when we're awaken
- */
- ldout(cct, 20) << __func__ << ":" << " stack=" << (void *)stack << " is_blocked_by_stack()=" << stack->is_blocked_by_stack()
- << " is_sleeping=" << stack->is_sleeping() << " waiting_for_child()=" << stack->waiting_for_child() << dendl;
- } else if (stack->is_done()) {
- ldout(cct, 20) << __func__ << ":" << " stack=" << (void *)stack << " is done" << dendl;
- RGWCoroutinesStack *s;
- while (stack->unblock_stack(&s)) {
- if (!s->is_blocked_by_stack() && !s->is_done()) {
- if (s->is_io_blocked()) {
- if (stack->is_interval_waiting()) {
- interval_wait_count++;
- }
- blocked_count++;
- } else {
- s->schedule();
- }
- }
- }
- if (stack->parent && stack->parent->waiting_for_child()) {
- stack->parent->set_wait_for_child(false);
- stack->parent->schedule();
- }
- context_stacks.erase(stack);
- stack->put();
- stack = NULL;
- } else {
- op_not_blocked = true;
- stack->run_count++;
- stack->schedule();
- }
-
- if (!op_not_blocked && stack) {
- stack->run_count = 0;
- }
-
- lock.unlock();
-
- RGWCoroutinesStack *blocked_stack;
- while (completion_mgr->try_get_next((void **)&blocked_stack)) {
- handle_unblocked_stack(context_stacks, scheduled_stacks, blocked_stack, &blocked_count);
- }
-
- /*
- * only account blocked operations that are not in interval_wait, these are stacks that
- * were put on a wait without any real IO operations. While we mark these as io_blocked,
- * these aren't really waiting for IOs
- */
- while (blocked_count - interval_wait_count >= ops_window) {
- ret = completion_mgr->get_next((void **)&blocked_stack);
- if (ret < 0) {
- ldout(cct, 0) << "ERROR: failed to clone shard, completion_mgr.get_next() returned ret=" << ret << dendl;
- }
- handle_unblocked_stack(context_stacks, scheduled_stacks, blocked_stack, &blocked_count);
- }
-
- ++iter;
- scheduled_stacks.pop_front();
-
-
- while (scheduled_stacks.empty() && blocked_count > 0) {
- ret = completion_mgr->get_next((void **)&blocked_stack);
- if (ret < 0) {
- ldout(cct, 0) << "ERROR: failed to clone shard, completion_mgr.get_next() returned ret=" << ret << dendl;
- }
- if (going_down) {
- ldout(cct, 5) << __func__ << "(): was stopped, exiting" << dendl;
- ret = -ECANCELED;
- canceled = true;
- break;
- }
- handle_unblocked_stack(context_stacks, scheduled_stacks, blocked_stack, &blocked_count);
- iter = scheduled_stacks.begin();
- }
- if (canceled) {
- break;
- }
-
- if (iter == scheduled_stacks.end()) {
- iter = scheduled_stacks.begin();
- }
- }
-
- lock.get_write();
- if (!context_stacks.empty() && !going_down) {
- JSONFormatter formatter(true);
- formatter.open_array_section("context_stacks");
- for (auto& s : context_stacks) {
- ::encode_json("entry", *s, &formatter);
- }
- formatter.close_section();
- lderr(cct) << __func__ << "(): ERROR: deadlock detected, dumping remaining coroutines:\n";
- formatter.flush(*_dout);
- *_dout << dendl;
- assert(context_stacks.empty() || going_down); // assert on deadlock
- }
-
- for (auto stack : context_stacks) {
- ldout(cct, 20) << "clearing stack on run() exit: stack=" << (void *)stack << " nref=" << stack->get_nref() << dendl;
- stack->put();
- }
- run_contexts.erase(run_context);
- lock.unlock();
-
- return ret;
-}
-
-int RGWCoroutinesManager::run(RGWCoroutine *op)
-{
- if (!op) {
- return 0;
- }
- list<RGWCoroutinesStack *> stacks;
- RGWCoroutinesStack *stack = allocate_stack();
- op->get();
- stack->call(op);
-
- stack->schedule(&stacks);
-
- int r = run(stacks);
- if (r < 0) {
- ldout(cct, 20) << "run(stacks) returned r=" << r << dendl;
- } else {
- r = op->get_ret_status();
- }
- op->put();
-
- return r;
-}
-
-RGWAioCompletionNotifier *RGWCoroutinesManager::create_completion_notifier(RGWCoroutinesStack *stack)
-{
- RGWAioCompletionNotifier *cn = new RGWAioCompletionNotifier(completion_mgr, (void *)stack);
- completion_mgr->register_completion_notifier(cn);
- return cn;
-}
-
-void RGWCoroutinesManager::dump(Formatter *f) const {
- RWLock::RLocker rl(lock);
-
- f->open_array_section("run_contexts");
- for (auto& i : run_contexts) {
- f->open_object_section("context");
- ::encode_json("id", i.first, f);
- f->open_array_section("entries");
- for (auto& s : i.second) {
- ::encode_json("entry", *s, f);
- }
- f->close_section();
- f->close_section();
- }
- f->close_section();
-}
-
-RGWCoroutinesStack *RGWCoroutinesManager::allocate_stack() {
- return new RGWCoroutinesStack(cct, this);
-}
-
-string RGWCoroutinesManager::get_id()
-{
- if (!id.empty()) {
- return id;
- }
- stringstream ss;
- ss << (void *)this;
- return ss.str();
-}
-
-void RGWCoroutinesManagerRegistry::add(RGWCoroutinesManager *mgr)
-{
- RWLock::WLocker wl(lock);
- if (managers.find(mgr) == managers.end()) {
- managers.insert(mgr);
- get();
- }
-}
-
-void RGWCoroutinesManagerRegistry::remove(RGWCoroutinesManager *mgr)
-{
- RWLock::WLocker wl(lock);
- if (managers.find(mgr) != managers.end()) {
- managers.erase(mgr);
- put();
- }
-}
-
-RGWCoroutinesManagerRegistry::~RGWCoroutinesManagerRegistry()
-{
- AdminSocket *admin_socket = cct->get_admin_socket();
- if (!admin_command.empty()) {
- admin_socket->unregister_command(admin_command);
- }
-}
-
-int RGWCoroutinesManagerRegistry::hook_to_admin_command(const string& command)
-{
- AdminSocket *admin_socket = cct->get_admin_socket();
- if (!admin_command.empty()) {
- admin_socket->unregister_command(admin_command);
- }
- admin_command = command;
- int r = admin_socket->register_command(admin_command, admin_command, this,
- "dump current coroutines stack state");
- if (r < 0) {
- lderr(cct) << "ERROR: fail to register admin socket command (r=" << r << ")" << dendl;
- return r;
- }
- return 0;
-}
-
-bool RGWCoroutinesManagerRegistry::call(std::string command, cmdmap_t& cmdmap, std::string format,
- bufferlist& out) {
- RWLock::RLocker rl(lock);
- stringstream ss;
- JSONFormatter f;
- ::encode_json("cr_managers", *this, &f);
- f.flush(ss);
- out.append(ss);
- return true;
-}
-
-void RGWCoroutinesManagerRegistry::dump(Formatter *f) const {
- f->open_array_section("coroutine_managers");
- for (auto m : managers) {
- ::encode_json("entry", *m, f);
- }
- f->close_section();
-}
-
-void RGWCoroutine::call(RGWCoroutine *op)
-{
- stack->call(op);
-}
-
-RGWCoroutinesStack *RGWCoroutine::spawn(RGWCoroutine *op, bool wait)
-{
- return stack->spawn(this, op, wait);
-}
-
-bool RGWCoroutine::collect(int *ret, RGWCoroutinesStack *skip_stack) /* returns true if needs to be called again */
-{
- return stack->collect(this, ret, skip_stack);
-}
-
-bool RGWCoroutine::collect_next(int *ret, RGWCoroutinesStack **collected_stack) /* returns true if found a stack to collect */
-{
- return stack->collect_next(this, ret, collected_stack);
-}
-
-int RGWCoroutine::wait(const utime_t& interval)
-{
- return stack->wait(interval);
-}
-
-void RGWCoroutine::wait_for_child()
-{
- /* should only wait for child if there is a child that is not done yet, and no complete children */
- if (spawned.entries.empty()) {
- return;
- }
- for (vector<RGWCoroutinesStack *>::iterator iter = spawned.entries.begin(); iter != spawned.entries.end(); ++iter) {
- if ((*iter)->is_done()) {
- return;
- }
- }
- stack->set_wait_for_child(true);
-}
-
-string RGWCoroutine::to_str() const
-{
- return typeid(*this).name();
-}
-
-ostream& operator<<(ostream& out, const RGWCoroutine& cr)
-{
- out << "cr:s=" << (void *)cr.get_stack() << ":op=" << (void *)&cr << ":" << typeid(cr).name();
- return out;
-}
-
-bool RGWCoroutine::drain_children(int num_cr_left, RGWCoroutinesStack *skip_stack)
-{
- bool done = false;
- assert(num_cr_left >= 0);
- if (num_cr_left == 0 && skip_stack) {
- num_cr_left = 1;
- }
- reenter(&drain_cr) {
- while (num_spawned() > (size_t)num_cr_left) {
- yield wait_for_child();
- int ret;
- while (collect(&ret, skip_stack)) {
- if (ret < 0) {
- ldout(cct, 10) << "collect() returned ret=" << ret << dendl;
- /* we should have reported this error */
- log_error() << "ERROR: collect() returned error (ret=" << ret << ")";
- }
- }
- }
- done = true;
- }
- return done;
-}
-
-void RGWCoroutine::wakeup()
-{
- stack->wakeup();
-}
-
-void RGWCoroutine::dump(Formatter *f) const {
- if (!description.str().empty()) {
- encode_json("description", description.str(), f);
- }
- encode_json("type", to_str(), f);
- if (!spawned.entries.empty()) {
- f->open_array_section("spawned");
- for (auto& i : spawned.entries) {
- char buf[32];
- snprintf(buf, sizeof(buf), "%p", (void *)i);
- encode_json("stack", string(buf), f);
- }
- f->close_section();
- }
- if (!status.history.empty()) {
- encode_json("history", status.history, f);
- }
-
- if (!status.status.str().empty()) {
- f->open_object_section("status");
- encode_json("status", status.status.str(), f);
- encode_json("timestamp", status.timestamp, f);
- f->close_section();
- }
-}
-
-RGWSimpleCoroutine::~RGWSimpleCoroutine()
-{
- if (!called_cleanup) {
- request_cleanup();
- }
-}
-
-void RGWSimpleCoroutine::call_cleanup()
-{
- called_cleanup = true;
- request_cleanup();
-}
-
-int RGWSimpleCoroutine::operate()
-{
- int ret = 0;
- reenter(this) {
- yield return state_init();
- yield return state_send_request();
- yield return state_request_complete();
- yield return state_all_complete();
- drain_all();
- call_cleanup();
- return set_state(RGWCoroutine_Done, ret);
- }
- return 0;
-}
-
-int RGWSimpleCoroutine::state_init()
-{
- int ret = init();
- if (ret < 0) {
- call_cleanup();
- return set_state(RGWCoroutine_Error, ret);
- }
- return 0;
-}
-
-int RGWSimpleCoroutine::state_send_request()
-{
- int ret = send_request();
- if (ret < 0) {
- call_cleanup();
- return set_state(RGWCoroutine_Error, ret);
- }
- return io_block(0);
-}
-
-int RGWSimpleCoroutine::state_request_complete()
-{
- int ret = request_complete();
- if (ret < 0) {
- call_cleanup();
- return set_state(RGWCoroutine_Error, ret);
- }
- return 0;
-}
-
-int RGWSimpleCoroutine::state_all_complete()
-{
- int ret = finish();
- if (ret < 0) {
- call_cleanup();
- return set_state(RGWCoroutine_Error, ret);
- }
- return 0;
-}
-
-
Code Review / stor4nfv.git / blobdiff