Skip to content

Commit 4c0f422

Browse files
adamemersonadamemerson
committed
rgw/multi: Give tasks a reference to RGWDataChangesLog
Also run them in strands. Also `datalog_rados` is a `shared_ptr`, now. Probably make it intrusive later. Signed-off-by: Adam C. Emerson <[email protected]>
1 parent 24e67fa commit 4c0f422

File tree

9 files changed

+94
-94
lines changed

9 files changed

+94
-94
lines changed

‎src/include/neorados/RADOS.hpp‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1386,6 +1386,10 @@ class RADOS final
13861386
executor_type get_executor() const;
13871387
boost::asio::io_context& get_io_context();
13881388

1389+
operator bool() {
1390+
return bool(impl);
1391+
}
1392+
13891393
private:
13901394
template<typename CompletionToken>
13911395
auto consign(CompletionToken&& token) {

‎src/rgw/driver/rados/rgw_datalog.cc‎

Lines changed: 41 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717

1818
#include <boost/system/system_error.hpp>
1919

20-
#include "common/async/parallel_for_each.h"
2120
#include "include/fs_types.h"
2221
#include "include/neorados/RADOS.hpp"
2322

@@ -356,17 +355,18 @@ class RGWDataChangesFIFO final : public RGWDataChangesBE {
356355
}
357356
};
358357

359-
RGWDataChangesLog::RGWDataChangesLog(CephContext* cct)
360-
: cct(cct),
358+
RGWDataChangesLog::RGWDataChangesLog(rgw::sal::RadosStore* driver)
359+
: cct(driver->ctx()), rados(driver->get_neorados()),
360+
executor(driver->get_io_context().get_executor()),
361361
num_shards(cct->_conf->rgw_data_log_num_shards),
362362
prefix(get_prefix()),
363363
changes(cct->_conf->rgw_data_log_changes_size) {}
364364

365365
RGWDataChangesLog::RGWDataChangesLog(CephContext *cct, bool log_data,
366-
neorados::RADOS *rados,
366+
neorados::RADOS rados,
367367
std::optional<int> num_shards,
368368
std::optional<uint64_t> sem_max_keys)
369-
: cct(cct), rados(rados), log_data(log_data),
369+
: cct(cct), rados(rados), log_data(log_data), executor(rados.get_executor()),
370370
num_shards(num_shards ? *num_shards :
371371
cct->_conf->rgw_data_log_num_shards),
372372
prefix(get_prefix()), changes(cct->_conf->rgw_data_log_changes_size),
@@ -438,15 +438,13 @@ void DataLogBackends::handle_empty_to(uint64_t new_tail) {
438438
int RGWDataChangesLog::start(const DoutPrefixProvider *dpp,
439439
const RGWZone* zone,
440440
const RGWZoneParams& zoneparams,
441-
rgw::sal::RadosStore* store,
442441
bool background_tasks)
443442
{
444443
log_data = zone->log_data;
445-
rados = &store->get_neorados();
446444
try {
447445
// Blocking in startup code, not ideal, but won't hurt anything.
448446
std::exception_ptr eptr
449-
= asio::co_spawn(store->get_io_context(),
447+
= asio::co_spawn(executor,
450448
start(dpp, zoneparams.log_pool,
451449
background_tasks, background_tasks,
452450
background_tasks),
@@ -476,7 +474,6 @@ RGWDataChangesLog::start(const DoutPrefixProvider *dpp,
476474
bool renew)
477475
{
478476
down_flag = false;
479-
cancel_strand = asio::make_strand(rados->get_executor());
480477
ran_background = (recovery || watch || renew);
481478

482479
auto defbacking = to_log_type(
@@ -512,10 +509,10 @@ RGWDataChangesLog::start(const DoutPrefixProvider *dpp,
512509

513510
if (renew) {
514511
asio::co_spawn(
515-
co_await asio::this_coro::executor,
516-
renew_run(renew_signal),
517-
asio::bind_cancellation_slot(renew_signal->slot(),
518-
asio::bind_executor(*cancel_strand,
512+
renew_strand,
513+
renew_run(shared_from_this()),
514+
asio::bind_cancellation_slot(renew_signal.slot(),
515+
asio::bind_executor(renew_strand,
519516
asio::detached)));
520517
}
521518
if (watch) {
@@ -527,20 +524,20 @@ RGWDataChangesLog::start(const DoutPrefixProvider *dpp,
527524
"Unable to establish recovery watch!"};
528525
}
529526
asio::co_spawn(
530-
co_await asio::this_coro::executor,
531-
watch_loop(watch_signal),
532-
asio::bind_cancellation_slot(watch_signal->slot(),
533-
asio::bind_executor(*cancel_strand,
527+
watch_strand,
528+
watch_loop(shared_from_this()),
529+
asio::bind_cancellation_slot(watch_signal.slot(),
530+
asio::bind_executor(watch_strand,
534531
asio::detached)));
535532
}
536533
if (recovery) {
537534
// Recovery can run concurrent with normal operation, so we don't
538535
// have to block startup while we do all that I/O.
539536
asio::co_spawn(
540-
co_await asio::this_coro::executor,
541-
recover(dpp, recovery_signal),
542-
asio::bind_cancellation_slot(recovery_signal->slot(),
543-
asio::bind_executor(*cancel_strand,
537+
recovery_strand,
538+
recover(dpp, shared_from_this()),
539+
asio::bind_cancellation_slot(recovery_signal.slot(),
540+
asio::bind_executor(recovery_strand,
544541
asio::detached)));
545542
}
546543
co_return;
@@ -667,7 +664,9 @@ RGWDataChangesLog::process_notification(const DoutPrefixProvider* dpp,
667664
}
668665
}
669666

670-
asio::awaitable<void> RGWDataChangesLog::watch_loop(decltype(watch_signal)) {
667+
asio::awaitable<void>
668+
RGWDataChangesLog::watch_loop(std::shared_ptr<RGWDataChangesLog>)
669+
{
671670
const DoutPrefix dp(cct, dout_subsys, "rgw data changes log: ");
672671
const auto oid = get_sem_set_oid(0);
673672
bool need_rewatch = false;
@@ -1363,21 +1362,18 @@ asio::awaitable<void> RGWDataChangesLog::shutdown() {
13631362
}
13641363
renew_stop();
13651364
// Revisit this later
1366-
if (renew_signal)
1367-
asio::dispatch(*cancel_strand,
1368-
[this]() {
1369-
renew_signal->emit(asio::cancellation_type::terminal);
1370-
});
1371-
if (recovery_signal)
1372-
asio::dispatch(*cancel_strand,
1373-
[this]() {
1374-
recovery_signal->emit(asio::cancellation_type::terminal);
1375-
});
1376-
if (watch_signal)
1377-
asio::dispatch(*cancel_strand,
1378-
[this]() {
1379-
watch_signal->emit(asio::cancellation_type::terminal);
1380-
});
1365+
asio::dispatch(renew_strand,
1366+
[this]() {
1367+
renew_signal.emit(asio::cancellation_type::terminal);
1368+
});
1369+
asio::dispatch(recovery_strand,
1370+
[this]() {
1371+
recovery_signal.emit(asio::cancellation_type::terminal);
1372+
});
1373+
asio::dispatch(watch_strand,
1374+
[this]() {
1375+
watch_signal.emit(asio::cancellation_type::terminal);
1376+
});
13811377
if (watchcookie && rados->check_watch(watchcookie)) {
13821378
auto wc = watchcookie;
13831379
watchcookie = 0;
@@ -1390,15 +1386,6 @@ asio::awaitable<void> RGWDataChangesLog::shutdown_or_timeout() {
13901386
using namespace asio::experimental::awaitable_operators;
13911387
asio::steady_timer t(co_await asio::this_coro::executor, 3s);
13921388
co_await (shutdown() || t.async_wait(asio::use_awaitable));
1393-
if (renew_signal) {
1394-
renew_signal->emit(asio::cancellation_type::terminal);
1395-
}
1396-
if (recovery_signal) {
1397-
recovery_signal->emit(asio::cancellation_type::terminal);
1398-
}
1399-
if (watch_signal) {
1400-
watch_signal->emit(asio::cancellation_type::terminal);
1401-
}
14021389
}
14031390

14041391
RGWDataChangesLog::~RGWDataChangesLog() {
@@ -1429,7 +1416,8 @@ void RGWDataChangesLog::blocking_shutdown() {
14291416
}
14301417
}
14311418

1432-
asio::awaitable<void> RGWDataChangesLog::renew_run(decltype(renew_signal)) {
1419+
asio::awaitable<void> RGWDataChangesLog::renew_run(
1420+
std::shared_ptr<RGWDataChangesLog>) {
14331421
static constexpr auto runs_per_prune = 150;
14341422
auto run = 0;
14351423
renew_timer.emplace(co_await asio::this_coro::executor);
@@ -1733,14 +1721,14 @@ RGWDataChangesLog::recover_shard(const DoutPrefixProvider* dpp, int index)
17331721
co_return;
17341722
}
17351723

1736-
asio::awaitable<void> RGWDataChangesLog::recover(const DoutPrefixProvider* dpp,
1737-
decltype(recovery_signal))
1724+
asio::awaitable<void> RGWDataChangesLog::recover(
1725+
const DoutPrefixProvider* dpp,
1726+
std::shared_ptr<RGWDataChangesLog>)
17381727
{
1739-
auto strand = asio::make_strand(co_await asio::this_coro::executor);
17401728
co_await asio::co_spawn(
1741-
strand,
1742-
[this](const DoutPrefixProvider* dpp)-> asio::awaitable<void, decltype(strand)> {
1743-
auto ex = co_await boost::asio::this_coro::executor;
1729+
recovery_strand,
1730+
[this](const DoutPrefixProvider* dpp)-> asio::awaitable<void, strand_t> {
1731+
auto ex = recovery_strand;
17441732
auto group = async::spawn_group{ex, static_cast<size_t>(num_shards)};
17451733
for (auto i = 0; i < num_shards; ++i) {
17461734
boost::asio::co_spawn(ex, recover_shard(dpp, i), group);

‎src/rgw/driver/rados/rgw_datalog.h‎

Lines changed: 20 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -350,23 +350,27 @@ struct hash<BucketGen> {
350350
};
351351
}
352352

353-
class RGWDataChangesLog {
353+
class RGWDataChangesLog
354+
: public std::enable_shared_from_this<RGWDataChangesLog> {
354355
friend class DataLogTestBase;
355356
friend DataLogBackends;
356357
CephContext *cct;
357-
neorados::RADOS* rados;
358-
std::optional<asio::strand<asio::io_context::executor_type>> cancel_strand;
358+
std::optional<neorados::RADOS> rados;
359359
neorados::IOContext loc;
360360
rgw::BucketChangeObserver *observer = nullptr;
361361
bool log_data = false;
362362
std::unique_ptr<DataLogBackends> bes;
363363

364-
std::shared_ptr<asio::cancellation_signal> renew_signal =
365-
std::make_shared<asio::cancellation_signal>();
366-
std::shared_ptr<asio::cancellation_signal> watch_signal =
367-
std::make_shared<asio::cancellation_signal>();
368-
std::shared_ptr<asio::cancellation_signal> recovery_signal =
369-
std::make_shared<asio::cancellation_signal>();
364+
using executor_t = asio::io_context::executor_type;
365+
executor_t executor;
366+
using strand_t = asio::strand<executor_t>;
367+
strand_t renew_strand{executor};
368+
asio::cancellation_signal renew_signal = asio::cancellation_signal();
369+
strand_t watch_strand{executor};
370+
asio::cancellation_signal watch_signal = asio::cancellation_signal();
371+
strand_t recovery_strand{executor};
372+
asio::cancellation_signal recovery_signal = asio::cancellation_signal();
373+
370374
ceph::mono_time last_recovery = ceph::mono_clock::zero();
371375

372376
const int num_shards;
@@ -410,7 +414,8 @@ class RGWDataChangesLog {
410414
ceph::real_time expiration);
411415

412416
std::optional<asio::steady_timer> renew_timer;
413-
asio::awaitable<void> renew_run(decltype(renew_signal) renew_signal);
417+
asio::awaitable<void> renew_run(
418+
std::shared_ptr<RGWDataChangesLog> renew_signal);
414419
void renew_stop();
415420

416421
std::function<bool(const rgw_bucket& bucket, optional_yield y,
@@ -425,10 +430,10 @@ class RGWDataChangesLog {
425430

426431
public:
427432

428-
RGWDataChangesLog(CephContext* cct);
433+
RGWDataChangesLog(rgw::sal::RadosStore* driver);
429434
// For testing.
430435
RGWDataChangesLog(CephContext* cct, bool log_data,
431-
neorados::RADOS* rados,
436+
neorados::RADOS rados,
432437
std::optional<int> num_shards = std::nullopt,
433438
std::optional<uint64_t> sem_max_keys = std::nullopt);
434439
~RGWDataChangesLog();
@@ -441,13 +446,12 @@ class RGWDataChangesLog {
441446
bool recovery, bool watch, bool renew);
442447

443448
int start(const DoutPrefixProvider *dpp, const RGWZone* _zone,
444-
const RGWZoneParams& zoneparams, rgw::sal::RadosStore* store,
445-
bool background_tasks);
449+
const RGWZoneParams& zoneparams, bool background_tasks);
446450
asio::awaitable<bool> establish_watch(const DoutPrefixProvider* dpp,
447451
std::string_view oid);
448452
asio::awaitable<void> process_notification(const DoutPrefixProvider* dpp,
449453
std::string_view oid);
450-
asio::awaitable<void> watch_loop(decltype(watch_signal));
454+
asio::awaitable<void> watch_loop(std::shared_ptr<RGWDataChangesLog>);
451455
int choose_oid(const rgw_bucket_shard& bs);
452456
asio::awaitable<void> add_entry(const DoutPrefixProvider *dpp,
453457
const RGWBucketInfo& bucket_info,
@@ -529,7 +533,7 @@ class RGWDataChangesLog {
529533
bc::flat_map<std::string, uint64_t>&& semcount);
530534
asio::awaitable<void> recover_shard(const DoutPrefixProvider* dpp, int index);
531535
asio::awaitable<void> recover(const DoutPrefixProvider* dpp,
532-
decltype(recovery_signal));
536+
std::shared_ptr<RGWDataChangesLog>);
533537
asio::awaitable<void> shutdown();
534538
asio::awaitable<void> shutdown_or_timeout();
535539
void blocking_shutdown();

‎src/rgw/driver/rados/rgw_rados.cc‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1467,12 +1467,7 @@ int RGWRados::init_begin(CephContext* _cct, const DoutPrefixProvider *dpp,
14671467
const rgw::SiteConfig& site, rgw::sal::ConfigStore* cfgstore)
14681468
{
14691469
set_context(_cct);
1470-
int ret = driver->init_neorados(dpp);
1471-
if (ret < 0) {
1472-
ldpp_dout(dpp, 0) << "ERROR: failed to initialize neorados (ret=" << cpp_strerror(-ret) << ")" << dendl;
1473-
return ret;
1474-
}
1475-
ret = init_rados();
1470+
auto ret = init_rados();
14761471
if (ret < 0) {
14771472
ldpp_dout(dpp, 0) << "ERROR: failed to initialize librados (ret=" << cpp_strerror(-ret) << ")" << dendl;
14781473
return ret;

‎src/rgw/driver/rados/rgw_sal_rados.cc‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5483,11 +5483,20 @@ int RadosRole::delete_obj(const DoutPrefixProvider *dpp, optional_yield y)
54835483

54845484
extern "C" {
54855485

5486-
void* newRadosStore(void* io_context)
5486+
void* newRadosStore(void* io_context, void* dpp_)
54875487
{
5488+
auto dpp = static_cast<DoutPrefixProvider*>(dpp_);
54885489
rgw::sal::RadosStore* store = new rgw::sal::RadosStore(
54895490
*static_cast<boost::asio::io_context*>(io_context));
54905491
if (store) {
5492+
int ret = store->init_neorados(dpp);
5493+
if (ret < 0) {
5494+
ldpp_dout(dpp, 0) << "ERROR: failed to initialize neorados (ret=" << cpp_strerror(-ret) << ")" << dendl;
5495+
delete store;
5496+
store = nullptr;
5497+
return store;
5498+
}
5499+
54915500
RGWRados* rados = new RGWRados();
54925501

54935502
if (!rados) {

‎src/rgw/driver/rados/rgw_service.cc‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ int RGWServices_Def::init(CephContext *cct,
6464
bilog_rados = std::make_unique<RGWSI_BILog_RADOS>(cct);
6565
cls = std::make_unique<RGWSI_Cls>(cct);
6666
config_key_rados = std::make_unique<RGWSI_ConfigKey_RADOS>(cct);
67-
datalog_rados = std::make_unique<RGWDataChangesLog>(cct);
67+
datalog_rados = std::make_shared<RGWDataChangesLog>(driver);
6868
mdlog = std::make_unique<RGWSI_MDLog>(cct, run_sync, cfgstore);
6969
notify = std::make_unique<RGWSI_Notify>(cct);
7070
zone = std::make_unique<RGWSI_Zone>(cct, cfgstore, site);
@@ -139,7 +139,7 @@ int RGWServices_Def::init(CephContext *cct,
139139

140140
r = datalog_rados->start(dpp, &zone->get_zone(),
141141
zone->get_zone_params(),
142-
driver, background_tasks);
142+
background_tasks);
143143
if (r < 0) {
144144
ldpp_dout(dpp, 0) << "ERROR: failed to start datalog_rados service (" << cpp_strerror(-r) << dendl;
145145
return r;

‎src/rgw/driver/rados/rgw_service.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -101,7 +101,7 @@ struct RGWServices_Def
101101
std::unique_ptr<RGWSI_SysObj_Core> sysobj_core;
102102
std::unique_ptr<RGWSI_SysObj_Cache> sysobj_cache;
103103
std::unique_ptr<RGWSI_User_RADOS> user_rados;
104-
std::unique_ptr<RGWDataChangesLog> datalog_rados;
104+
std::shared_ptr<RGWDataChangesLog> datalog_rados;
105105
std::unique_ptr<RGWAsyncRadosProcessor> async_processor;
106106

107107
RGWServices_Def();

‎src/rgw/rgw_sal.cc‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@
4949

5050
extern "C" {
5151
#ifdef WITH_RADOSGW_RADOS
52-
extern rgw::sal::Driver* newRadosStore(boost::asio::io_context* io_context);
52+
extern rgw::sal::Driver* newRadosStore(void* io_context, const void* dpp);
5353
#endif
5454
#ifdef WITH_RADOSGW_DBSTORE
5555
extern rgw::sal::Driver* newDBStore(CephContext *cct);
@@ -90,7 +90,7 @@ rgw::sal::Driver* DriverManager::init_storage_provider(const DoutPrefixProvider*
9090

9191
if (cfg.store_name.compare("rados") == 0) {
9292
#ifdef WITH_RADOSGW_RADOS
93-
driver = newRadosStore(&io_context);
93+
driver = newRadosStore(&io_context, dpp);
9494
RGWRados* rados = static_cast<rgw::sal::RadosStore* >(driver)->getRados();
9595

9696
if ((*rados).set_use_cache(use_cache)
@@ -248,7 +248,7 @@ rgw::sal::Driver* DriverManager::init_raw_storage_provider(const DoutPrefixProvi
248248
rgw::sal::Driver* driver = nullptr;
249249
if (cfg.store_name.compare("rados") == 0) {
250250
#ifdef WITH_RADOSGW_RADOS
251-
driver = newRadosStore(&io_context);
251+
driver = newRadosStore(&io_context, dpp);
252252
RGWRados* rados = static_cast<rgw::sal::RadosStore* >(driver)->getRados();
253253

254254
rados->set_context(cct);

0 commit comments

Comments
 (0)