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

Sync with the latest version of helio

This commit is contained in:
Roman Gershman 2022-01-16 23:56:47 +02:00
parent ee15f98142
commit 0d57b25124
6 changed files with 11 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
helio

@ -1 +1 @@
Subproject commit de1c2b5578f835f1d659f437c0a7ea7d991faa24
Subproject commit 9db022200ac61ef2faa832aaf03d5d480f945071

2
server/dragonfly_test.cc
View File

@ -197,7 +197,7 @@ TEST_F(DflyEngineTest, FlushDb) {
}
});
pp_->at(1)->AwaitBlocking([&] {
pp_->at(1)->Await([&] {
for (unsigned i = 0; i < 100; ++i) {
Run({"mset", kKey1, "1", kKey4, "2"});
auto resp = Run({"exists", kKey1, kKey4});

4
server/engine_shard_set.h
View File

@ -217,7 +217,7 @@ void EngineShardSet::RunBriefInParallel(U&& func, P&& pred) const {
bc.Add(1);
util::ProactorBase* dest = pp_->at(i);
dest->AsyncBrief([f = std::forward<U>(func), bc]() mutable {
dest->DispatchBrief([f = std::forward<U>(func), bc]() mutable {
f(EngineShard::tlocal());
bc.Dec();
});
@ -230,7 +230,7 @@ template <typename U> void EngineShardSet::RunBlockingInParallel(U&& func) {
for (uint32_t i = 0; i < size(); ++i) {
util::ProactorBase* dest = pp_->at(i);
dest->AsyncFiber([func, bc]() mutable {
dest->Dispatch([func, bc]() mutable {
func(EngineShard::tlocal());
bc.Dec();
});

10
server/list_family_test.cc
View File

@ -95,15 +95,15 @@ TEST_F(ListFamilyTest, BLPopBlocking) {
});
this_fiber::sleep_for(50us);
/*auto fb1 = pp_->at(1)->LaunchFiber([&] {
auto fb1 = pp_->at(1)->LaunchFiber([&] {
resp1 = Run({"blpop", "x", "0"});
LOG(INFO) << "pop1";
});*/
});
this_fiber::sleep_for(30us);
pp_->at(1)->AwaitBlocking([&] { Run({"lpush", "x", "2", "1"}); });
pp_->at(1)->Await([&] { Run({"lpush", "x", "2", "1"}); });
fb0.join();
// fb1.join();
fb1.join();
// fb0 should start first and be the first transaction blocked. Therefore, it should pop '1'.
// sometimes order is switched, need to think how to fix it.
@ -129,7 +129,7 @@ TEST_F(ListFamilyTest, BLPopMultiple) {
resp0 = Run({"blpop", kKey1, kKey2, "0"});
});
pp_->at(1)->AwaitBlocking([&] { Run({"lpush", kKey1, "1", "2", "3"}); });
pp_->at(1)->Await([&] { Run({"lpush", kKey1, "1", "2", "3"}); });
fb1.join();
EXPECT_THAT(resp0, ElementsAre(StrArg(kKey1), StrArg("3")));
ASSERT_FALSE(IsLocked(0, kKey1));

2
server/main_service.cc
View File

@ -69,7 +69,7 @@ void Service::Init(util::AcceptServer* acceptor, const InitOpts& opts) {
uint32_t shard_num = pp_.size() > 1 ? pp_.size() - 1 : pp_.size();
shard_set_.Init(shard_num);
pp_.AwaitOnAll([&](uint32_t index, ProactorBase* pb) {
pp_.Await([&](uint32_t index, ProactorBase* pb) {
if (index < shard_count()) {
shard_set_.InitThreadLocal(pb, !opts.disable_time_update);
}

2
server/test_utils.cc
View File

@ -158,7 +158,7 @@ void BaseFamilyTest::UpdateTime(uint64_t ms) {
RespVec BaseFamilyTest::Run(initializer_list<std::string_view> list) {
if (!ProactorBase::IsProactorThread()) {
return pp_->at(0)->AwaitBlocking([&] { return this->Run(list); });
return pp_->at(0)->Await([&] { return this->Run(list); });
}
mu_.lock();