innov
/
dragonfly
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Refactor Populate command. Name the helper fibers

This commit is contained in:
Roman Gershman 2022-01-23 20:53:45 +02:00
parent 65f35f2cac
commit f119e66199
3 changed files with 49 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
helio

@ -1 +1 @@
Subproject commit bfd4e2e15b16c187b907c50a45465abdab23edf9 Subproject commit f7c9d00016ea48866632583918b042b744f261b9

46
server/debugcmd.cc
View File

@ -5,15 +5,20 @@
#include <absl/strings/str_cat.h> #include <absl/strings/str_cat.h>
#include <boost/fiber/operations.hpp>
#include "base/logging.h" #include "base/logging.h"
#include "server/engine_shard_set.h" #include "server/engine_shard_set.h"
#include "server/error.h" #include "server/error.h"
#include "server/string_family.h" #include "server/string_family.h"
#include "util/uring/uring_fiber_algo.h"
namespace dfly { namespace dfly {
using namespace boost;
using namespace std; using namespace std;
using namespace util;
namespace this_fiber = ::boost::this_fiber;
using boost::fibers::fiber;
struct PopulateBatch { struct PopulateBatch {
DbIndex dbid; DbIndex dbid;
@ -24,8 +29,8 @@ struct PopulateBatch {
} }
}; };
void DoPopulateBatch(std::string_view prefix, size_t val_size, void DoPopulateBatch(std::string_view prefix, size_t val_size, const SetCmd::SetParams& params,
const SetCmd::SetParams& params, const PopulateBatch& ps) { const PopulateBatch& ps) {
SetCmd sg(&EngineShard::tlocal()->db_slice()); SetCmd sg(&EngineShard::tlocal()->db_slice());
for (unsigned i = 0; i < ps.sz; ++i) { for (unsigned i = 0; i < ps.sz; ++i) {
@ -99,7 +104,22 @@ void DebugCmd::Populate(CmdArgList args) {
} }
ranges.emplace_back(from, total_count - from); ranges.emplace_back(from, total_count - from);
auto distribute_cb = [this, val_size, prefix](uint64_t from, uint64_t len) { vector<fiber> fb_arr(ranges.size());
for (size_t i = 0; i < ranges.size(); ++i) {
fb_arr[i] = ess_->pool()->at(i)->LaunchFiber([&] {
this->PopulateRangeFiber(ranges[i].first, ranges[i].second, prefix, val_size);
});
}
for (auto& fb : fb_arr)
fb.join();
cntx_->SendOk();
}
void DebugCmd::PopulateRangeFiber(uint64_t from, uint64_t len, std::string_view prefix,
unsigned value_len) {
this_fiber::properties<FiberProps>().set_name("populate_range");
string key = absl::StrCat(prefix, ":"); string key = absl::StrCat(prefix, ":");
size_t prefsize = key.size(); size_t prefsize = key.size();
DbIndex db_indx = 0; // TODO DbIndex db_indx = 0; // TODO
@ -115,8 +135,8 @@ void DebugCmd::Populate(CmdArgList args) {
pops.index[pops.sz++] = i; pops.index[pops.sz++] = i;
if (pops.sz == 32) { if (pops.sz == 32) {
ess_->Add(sid, [=, p = pops] { ess_->Add(sid, [=, p = pops] {
DoPopulateBatch(prefix, val_size, params, p); DoPopulateBatch(prefix, value_len, params, p);
if (i % 100 == 0) { if (i % 50 == 0) {
this_fiber::yield(); this_fiber::yield();
} }
}); });
@ -126,19 +146,9 @@ void DebugCmd::Populate(CmdArgList args) {
} }
} }
ess_->RunBriefInParallel( ess_->RunBriefInParallel([&](EngineShard* shard) {
[&](EngineShard* shard) { DoPopulateBatch(prefix, value_len, params, ps[shard->shard_id()]);
DoPopulateBatch(prefix, val_size, params, ps[shard->shard_id()]);
}); });
};
vector<fibers::fiber> fb_arr(ranges.size());
for (size_t i = 0; i < ranges.size(); ++i) {
fb_arr[i] = ess_->pool()->at(i)->LaunchFiber(distribute_cb, ranges[i].first, ranges[i].second);
}
for (auto& fb : fb_arr)
fb.join();
cntx_->SendOk();
} }
} // namespace dfly } // namespace dfly

1
server/debugcmd.h
View File

@ -18,6 +18,7 @@ class DebugCmd {
private: private:
void Populate(CmdArgList args); void Populate(CmdArgList args);
void PopulateRangeFiber(uint64_t from, uint64_t len, std::string_view prefix, unsigned value_len);
EngineShardSet* ess_; EngineShardSet* ess_;
ConnectionContext* cntx_; ConnectionContext* cntx_;