#include <spawn/spawn.hpp>
-// use explicit executor types instead of the type-erased boost::asio::executor.
-// coroutines wrap the default io_context executor with a strand executor
-using yield_context = spawn::basic_yield_context<
- boost::asio::executor_binder<void(*)(),
- boost::asio::strand<boost::asio::io_context::executor_type>>>;
-
/// optional-like wrapper for a spawn::yield_context and its associated
/// boost::asio::io_context. operations that take an optional_yield argument
/// will, when passed a non-empty yield context, suspend this coroutine instead
/// of the blocking the thread of execution
class optional_yield {
boost::asio::io_context *c = nullptr;
- yield_context *y = nullptr;
+ spawn::yield_context *y = nullptr;
public:
/// construct with a valid io and yield_context
explicit optional_yield(boost::asio::io_context& c,
- yield_context& y) noexcept
+ spawn::yield_context& y) noexcept
: c(&c), y(&y) {}
/// type tag to construct an empty object
boost::asio::io_context& get_io_context() const noexcept { return *c; }
/// return a reference to the yield_context. only valid if non-empty
- yield_context& get_yield_context() const noexcept { return *y; }
+ spawn::yield_context& get_yield_context() const noexcept { return *y; }
};
// type tag object to construct an empty optional_yield
const int max_aio = std::max(1, op_state.get_max_aio());
for (int i=0; i<max_aio; i++) {
- spawn::spawn(context, [&](yield_context yield) {
+ spawn::spawn(context, [&](spawn::yield_context yield) {
while (true) {
int shard = next_shard;
next_shard += 1;
int next_shard = 0;
boost::asio::io_context context;
for (int i=0; i<max_aio; i++) {
- spawn::spawn(context, [&](yield_context yield) {
+ spawn::spawn(context, [&](spawn::yield_context yield) {
while (true) {
int shard = next_shard;
next_shard += 1;
pending_tokens(0),
timer(io_context) {}
- void async_wait(yield_context yield) {
+ void async_wait(spawn::yield_context yield) {
if (pending_tokens == 0) {
return;
}
// processing of a specific entry
// return whether processing was successful (true) or not (false)
EntryProcessingResult process_entry(const ConfigProxy& conf, persistency_tracker& entry_persistency_tracker,
- const cls_queue_entry& entry, yield_context yield) {
+ const cls_queue_entry& entry, spawn::yield_context yield) {
event_entry_t event_entry;
auto iter = entry.data.cbegin();
try {
}
// clean stale reservation from queue
- void cleanup_queue(const std::string& queue_name, yield_context yield) {
+ void cleanup_queue(const std::string& queue_name, spawn::yield_context yield) {
while (true) {
ldpp_dout(this, 20) << "INFO: trying to perform stale reservation cleanup for queue: " << queue_name << dendl;
const auto now = ceph::coarse_real_time::clock::now();
}
// processing of a specific queue
- void process_queue(const std::string& queue_name, yield_context yield) {
+ void process_queue(const std::string& queue_name, spawn::yield_context yield) {
constexpr auto max_elements = 1024;
auto is_idle = false;
const std::string start_marker;
// start a the cleanup coroutine for the queue
- spawn::spawn(io_context, [this, queue_name](yield_context yield) {
+ spawn::spawn(io_context, [this, queue_name](spawn::yield_context yield) {
cleanup_queue(queue_name, yield);
}, make_stack_allocator());
entries_persistency_tracker& notifs_persistency_tracker = topics_persistency_tracker[queue_name];
spawn::spawn(yield, [this, ¬ifs_persistency_tracker, &queue_name, entry_idx, total_entries, &end_marker,
- &remove_entries, &has_error, &waiter, &entry, &needs_migration_vector](yield_context yield) {
+ &remove_entries, &has_error, &waiter, &entry, &needs_migration_vector](spawn::yield_context yield) {
const auto token = waiter.make_token();
auto& persistency_tracker = notifs_persistency_tracker[entry.marker];
auto result = process_entry(this->get_cct()->_conf, persistency_tracker, entry, yield);
// process all queues
// find which of the queues is owned by this daemon and process it
- void process_queues(yield_context yield) {
+ void process_queues(spawn::yield_context yield) {
auto has_error = false;
owned_queues_t owned_queues;
if (owned_queues.insert(queue_name).second) {
ldpp_dout(this, 10) << "INFO: queue: " << queue_name << " now owned (locked) by this daemon" << dendl;
// start processing this queue
- spawn::spawn(io_context, [this, &queue_gc, &queue_gc_lock, queue_name](yield_context yield) {
+ spawn::spawn(io_context, [this, &queue_gc, &queue_gc_lock, queue_name](spawn::yield_context yield) {
process_queue(queue_name, yield);
// if queue processing ended, it means that the queue was removed or not owned anymore
// mark it for deletion
reservations_cleanup_period_s(_reservations_cleanup_period_s),
rados_store(*store)
{
- spawn::spawn(io_context, [this] (yield_context yield) {
+ spawn::spawn(io_context, [this](spawn::yield_context yield) {
process_queues(yield);
}, make_stack_allocator());
class Waiter {
using Signature = void(boost::system::error_code);
using Completion = ceph::async::Completion<Signature>;
- using CompletionInit = boost::asio::async_completion<yield_context, Signature>;
+ using CompletionInit = boost::asio::async_completion<spawn::yield_context, Signature>;
std::unique_ptr<Completion> completion = nullptr;
int ret;
template <typename Op>
Aio::OpFunc aio_abstract(librados::IoCtx ctx, Op&& op,
boost::asio::io_context& context,
- yield_context yield) {
+ spawn::yield_context yield) {
return [ctx = std::move(ctx), op = std::move(op), &context, yield] (Aio* aio, AioResult& r) mutable {
// arrange for the completion Handler to run on the yield_context's strand
// executor so it can safely call back into Aio without locking
using namespace boost::asio;
- async_completion<yield_context, void()> init(yield);
+ async_completion<spawn::yield_context, void()> init(yield);
auto ex = get_associated_executor(init.completion_handler);
librados::async_operate(context, ctx, r.obj.oid, &op, 0,
// functions must be called within the coroutine strand
class YieldingAioThrottle final : public Aio, private Throttle {
boost::asio::io_context& context;
- yield_context yield;
+ spawn::yield_context yield;
struct Handler;
// completion callback associated with the waiter
public:
YieldingAioThrottle(uint64_t window, boost::asio::io_context& context,
- yield_context yield)
+ spawn::yield_context yield)
: Throttle(window), context(context), yield(yield)
{}
CephContext* const cct;
Stream& stream;
timeout_timer& timeout;
- yield_context yield;
+ spawn::yield_context yield;
parse_buffer& buffer;
boost::system::error_code fatal_ec;
public:
StreamIO(CephContext *cct, Stream& stream, timeout_timer& timeout,
- rgw::asio::parser_type& parser, yield_context yield,
+ rgw::asio::parser_type& parser, spawn::yield_context yield,
parse_buffer& buffer, bool is_ssl,
const tcp::endpoint& local_endpoint,
const tcp::endpoint& remote_endpoint)
rgw::dmclock::Scheduler *scheduler,
const std::string& uri_prefix,
boost::system::error_code& ec,
- yield_context yield)
+ spawn::yield_context yield)
{
// don't impose a limit on the body, since we read it in pieces
static constexpr size_t body_limit = std::numeric_limits<size_t>::max();
#ifdef WITH_RADOSGW_BEAST_OPENSSL
if (l.use_ssl) {
spawn::spawn(context,
- [this, s=std::move(stream)] (yield_context yield) mutable {
+ [this, s=std::move(stream)] (spawn::yield_context yield) mutable {
auto conn = boost::intrusive_ptr{new Connection(std::move(s))};
auto c = connections.add(*conn);
// wrap the tcp stream in an ssl stream
{
#endif // WITH_RADOSGW_BEAST_OPENSSL
spawn::spawn(context,
- [this, s=std::move(stream)] (yield_context yield) mutable {
+ [this, s=std::move(stream)] (spawn::yield_context yield) mutable {
auto conn = boost::intrusive_ptr{new Connection(std::move(s))};
auto c = connections.add(*conn);
auto timeout = timeout_timer{context.get_executor(), request_timeout, conn};
}
};
- void file_aio_read_abstract(const DoutPrefixProvider *dpp, boost::asio::io_context& context, yield_context yield,
+ void file_aio_read_abstract(const DoutPrefixProvider *dpp, boost::asio::io_context& context, spawn::yield_context yield,
std::string& cache_location, off_t read_ofs, off_t read_len,
rgw::Aio* aio, rgw::AioResult& r) {
using namespace boost::asio;
- async_completion<yield_context, void()> init(yield);
+ async_completion<spawn::yield_context, void()> init(yield);
auto ex = get_associated_executor(init.completion_handler);
ldpp_dout(dpp, 20) << "D3nDataCache: " << __func__ << "(): oid=" << r.obj.oid << dendl;
return aio_count < max_aio;
});
aio_count++;
- spawn::spawn(y.get_yield_context(), [this, &y, &aio_count, obj_key, &formatter_flush_cond] (yield_context yield) {
+ spawn::spawn(y.get_yield_context(), [this, &y, &aio_count, obj_key, &formatter_flush_cond] (spawn::yield_context yield) {
handle_individual_object(obj_key, optional_yield { y.get_io_context(), yield }, &*formatter_flush_cond);
aio_count--;
});
entry.pipe = pipe;
// fetch remote markers
- spawn::spawn(ioctx, [&] (yield_context yield) {
+ spawn::spawn(ioctx, [&] (spawn::yield_context yield) {
auto y = optional_yield{ioctx, yield};
rgw_bucket_index_marker_info info;
int r = source_bilog_info(dpp, store->svc()->zone, entry.pipe,
entry.latest_gen = info.latest_gen;
});
// fetch source bucket info
- spawn::spawn(ioctx, [&] (yield_context yield) {
+ spawn::spawn(ioctx, [&] (spawn::yield_context yield) {
auto y = optional_yield{ioctx, yield};
int r = store->getRados()->get_bucket_instance_info(
*entry.pipe.source.bucket, entry.source_bucket_info,
{
boost::asio::io_context context;
- spawn::spawn(context, [&] (yield_context yield) {
+ spawn::spawn(context, [&] (spawn::yield_context yield) {
ClientCounters counters(g_ceph_context);
AsyncScheduler queue(g_ceph_context, context, std::ref(counters), nullptr,
[] (client_id client) -> ClientInfo* {
RGWReshardWait waiter(wait_duration);
boost::asio::io_context context;
- spawn::spawn(context, [&] (yield_context yield) {
+ spawn::spawn(context, [&] (spawn::yield_context yield) {
EXPECT_EQ(0, waiter.wait(optional_yield{context, yield}));
});
boost::asio::io_context context;
spawn::spawn(context,
- [&] (yield_context yield) {
+ [&] (spawn::yield_context yield) {
EXPECT_EQ(-ECANCELED, long_waiter.wait(optional_yield{context, yield}));
});
// spawn 4 coroutines
boost::asio::io_context context;
{
- auto async_waiter = [&] (yield_context yield) {
+ auto async_waiter = [&] (spawn::yield_context yield) {
EXPECT_EQ(-ECANCELED, long_waiter.wait(optional_yield{context, yield}));
};
spawn::spawn(context, async_waiter);
boost::asio::io_context context;
spawn::spawn(context,
- [&] (yield_context yield) {
+ [&] (spawn::yield_context yield) {
YieldingAioThrottle throttle(4, context, yield);
scoped_completion op;
auto c = throttle.get(obj, wait_on(op), 8, 0);
boost::asio::io_context context;
spawn::spawn(context,
- [&] (yield_context yield) {
+ [&] (spawn::yield_context yield) {
YieldingAioThrottle throttle(window, context, yield);
for (uint64_t i = 0; i < total; i++) {
using namespace std::chrono_literals;
projects / ceph.git / commitdiff