Skip to content

Commit 41196f0

Browse files
lll-phill-llllll-phill-lll
authored
[YQL-17709] [channel] split spilling into actor and non-actor parts (#1983)
1 parent d99f5c5 commit 41196f0

File tree

11 files changed

+182
-342
lines changed

11 files changed

+182
-342
lines changed

‎ydb/core/tx/datashard/datashard_kqp.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -983,7 +983,7 @@ class TKqpTaskRunnerExecutionContext: public NDq::IDqTaskRunnerExecutionContext
983983
return {};
984984
}
985985

986-
NDq::IDqChannelStorage::TPtr CreateChannelStorage(ui64 /* channelId */, bool /* withSpilling */, TActorSystem* /* actorSystem */, bool /*isConcurrent*/) const override {
986+
NDq::IDqChannelStorage::TPtr CreateChannelStorage(ui64 /* channelId */, bool /* withSpilling */, TActorSystem* /* actorSystem */) const override {
987987
return {};
988988
}
989989

‎ydb/library/yql/dq/actors/compute/dq_task_runner_exec_ctx.cpp

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,12 +13,12 @@ TDqTaskRunnerExecutionContext::TDqTaskRunnerExecutionContext(TTxId txId, IDqChan
1313
}
1414

1515
IDqChannelStorage::TPtr TDqTaskRunnerExecutionContext::CreateChannelStorage(ui64 channelId, bool withSpilling) const {
16-
return CreateChannelStorage(channelId, withSpilling, nullptr, false);
16+
return CreateChannelStorage(channelId, withSpilling, NActors::TlsActivationContext->ActorSystem());
1717
}
1818

19-
IDqChannelStorage::TPtr TDqTaskRunnerExecutionContext::CreateChannelStorage(ui64 channelId, bool withSpilling, NActors::TActorSystem* actorSystem, bool isConcurrent) const {
19+
IDqChannelStorage::TPtr TDqTaskRunnerExecutionContext::CreateChannelStorage(ui64 channelId, bool withSpilling, NActors::TActorSystem* actorSystem) const {
2020
if (withSpilling) {
21-
return CreateDqChannelStorage(TxId_, channelId, WakeUp_, actorSystem, isConcurrent);
21+
return CreateDqChannelStorage(TxId_, channelId, WakeUp_, actorSystem);
2222
} else {
2323
return nullptr;
2424
}

‎ydb/library/yql/dq/actors/compute/dq_task_runner_exec_ctx.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ class TDqTaskRunnerExecutionContext : public TDqTaskRunnerExecutionContextBase {
1212
TDqTaskRunnerExecutionContext(TTxId txId, IDqChannelStorage::TWakeUpCallback&& wakeUp);
1313

1414
IDqChannelStorage::TPtr CreateChannelStorage(ui64 channelId, bool withSpilling) const override;
15-
IDqChannelStorage::TPtr CreateChannelStorage(ui64 channelId, bool withSpilling, NActors::TActorSystem* actorSystem, bool isConcurrent) const override;
15+
IDqChannelStorage::TPtr CreateChannelStorage(ui64 channelId, bool withSpilling, NActors::TActorSystem* actorSystem) const override;
1616

1717
std::function<void()> GetWakeupCallback() const override;
1818

‎ydb/library/yql/dq/actors/spilling/channel_storage.cpp

Lines changed: 76 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -20,62 +20,108 @@ using namespace NActors;
2020

2121
namespace {
2222

23+
24+
constexpr ui32 MAX_INFLIGHT_BLOBS_COUNT = 10;
25+
constexpr ui64 MAX_INFLIGHT_BLOBS_SIZE = 50_MB;
26+
2327
class TDqChannelStorage : public IDqChannelStorage {
28+
struct TWritingBlobInfo {
29+
ui64 BlobSize_;
30+
NThreading::TFuture<void> IsBlobWrittenFuture_;
31+
};
2432
public:
25-
TDqChannelStorage(TTxId txId, ui64 channelId, TWakeUpCallback&& wakeUp, TActorSystem* actorSystem, bool isConcurrent)
33+
TDqChannelStorage(TTxId txId, ui64 channelId, TWakeUpCallback&& wakeUp, TActorSystem* actorSystem)
2634
: ActorSystem_(actorSystem)
2735
{
28-
if (isConcurrent) {
29-
SelfActor_ = CreateConcurrentDqChannelStorageActor(txId, channelId, std::move(wakeUp), actorSystem);
30-
} else {
31-
SelfActor_ = CreateDqChannelStorageActor(txId, channelId, std::move(wakeUp), actorSystem);
32-
}
33-
SelfActorId_ = TlsActivationContext->AsActorContext().RegisterWithSameMailbox(SelfActor_->GetActor());
34-
SelfId_ = TlsActivationContext->AsActorContext().SelfID;
36+
ChannelStorageActor_ = CreateDqChannelStorageActor(txId, channelId, std::move(wakeUp), actorSystem);
37+
ChannelStorageActorId_ = ActorSystem_->Register(ChannelStorageActor_->GetActor());
3538
}
3639

3740
~TDqChannelStorage() {
38-
if (ActorSystem_) {
39-
ActorSystem_->Send(
40-
new IEventHandle(
41-
SelfActorId_,
42-
SelfId_,
43-
new TEvents::TEvPoison,
44-
/*flags=*/0,
45-
/*cookie=*/0));
46-
} else {
47-
TlsActivationContext->AsActorContext().Send(SelfActorId_, new TEvents::TEvPoison);
48-
}
41+
ActorSystem_->Send(ChannelStorageActorId_, new TEvents::TEvPoison);
4942
}
5043

51-
bool IsEmpty() const override {
52-
return SelfActor_->IsEmpty();
44+
bool IsEmpty() override {
45+
UpdateWriteStatus();
46+
47+
return WritingBlobs_.empty() && StoredBlobsCount_ == 0 && LoadingBlobs_.empty();
5348
}
5449

55-
bool IsFull() const override {
56-
return SelfActor_->IsFull();
50+
bool IsFull() override {
51+
UpdateWriteStatus();
52+
53+
return WritingBlobs_.size() > MAX_INFLIGHT_BLOBS_COUNT || WritingBlobsTotalSize_ > MAX_INFLIGHT_BLOBS_SIZE;
5754
}
5855

5956
void Put(ui64 blobId, TRope&& blob, ui64 cookie = 0) override {
60-
SelfActor_->Put(blobId, std::move(blob), cookie);
57+
UpdateWriteStatus();
58+
59+
auto promise = NThreading::NewPromise<void>();
60+
auto future = promise.GetFuture();
61+
62+
ui64 blobSize = blob.size();
63+
64+
ActorSystem_->Send(ChannelStorageActorId_, new TEvDqChannelSpilling::TEvPut(blobId, std::move(blob), std::move(promise)), /*flags*/0, cookie);
65+
66+
WritingBlobs_.emplace(blobId, TWritingBlobInfo(blobSize, std::move(future)));
67+
WritingBlobsTotalSize_ += blobSize;
6168
}
6269

6370
bool Get(ui64 blobId, TBuffer& blob, ui64 cookie = 0) override {
64-
return SelfActor_->Get(blobId, blob, cookie);
71+
UpdateWriteStatus();
72+
73+
const auto it = LoadingBlobs_.find(blobId);
74+
// If we didn't request loading blob from spilling -> request it
75+
if (it == LoadingBlobs_.end()) {
76+
auto promise = NThreading::NewPromise<TBuffer>();
77+
auto future = promise.GetFuture();
78+
ActorSystem_->Send(ChannelStorageActorId_, new TEvDqChannelSpilling::TEvGet(blobId, std::move(promise)), /*flags*/0, cookie);
79+
80+
LoadingBlobs_.emplace(blobId, std::move(future));
81+
return false;
82+
}
83+
// If we requested loading blob, but it's not loaded -> wait
84+
if (!it->second.HasValue()) return false;
85+
86+
blob = std::move(it->second.ExtractValue());
87+
LoadingBlobs_.erase(it);
88+
--StoredBlobsCount_;
89+
90+
return true;
91+
}
92+
93+
private:
94+
void UpdateWriteStatus() {
95+
for (auto it = WritingBlobs_.begin(); it != WritingBlobs_.end();) {
96+
if (it->second.IsBlobWrittenFuture_.HasValue()) {
97+
WritingBlobsTotalSize_ -= it->second.BlobSize_;
98+
++StoredBlobsCount_;
99+
it = WritingBlobs_.erase(it);
100+
} else {
101+
++it;
102+
}
103+
}
65104
}
66105

67106
private:
68-
IDqChannelStorageActor* SelfActor_;
69-
TActorId SelfActorId_;
70-
TActorId SelfId_;
107+
IDqChannelStorageActor* ChannelStorageActor_;
108+
TActorId ChannelStorageActorId_;
71109
TActorSystem *ActorSystem_;
110+
111+
// BlobId -> future with requested blob
112+
std::unordered_map<ui64, NThreading::TFuture<TBuffer>> LoadingBlobs_;
113+
// BlobId -> future with some additional info
114+
std::unordered_map<ui64, TWritingBlobInfo> WritingBlobs_;
115+
ui64 WritingBlobsTotalSize_ = 0;
116+
117+
ui64 StoredBlobsCount_ = 0;
72118
};
73119

74120
} // anonymous namespace
75121

76-
IDqChannelStorage::TPtr CreateDqChannelStorage(TTxId txId, ui64 channelId, IDqChannelStorage::TWakeUpCallback wakeUp, TActorSystem* actorSystem, bool isConcurrent)
122+
IDqChannelStorage::TPtr CreateDqChannelStorage(TTxId txId, ui64 channelId, IDqChannelStorage::TWakeUpCallback wakeUp, TActorSystem* actorSystem)
77123
{
78-
return new TDqChannelStorage(txId, channelId, std::move(wakeUp), actorSystem, isConcurrent);
124+
return new TDqChannelStorage(txId, channelId, std::move(wakeUp), actorSystem);
79125
}
80126

81127
} // namespace NYql::NDq

‎ydb/library/yql/dq/actors/spilling/channel_storage.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,6 @@ namespace NActors {
1111
namespace NYql::NDq {
1212

1313
IDqChannelStorage::TPtr CreateDqChannelStorage(TTxId txId, ui64 channelId,
14-
IDqChannelStorage::TWakeUpCallback wakeUpCb, NActors::TActorSystem* actorSystem, bool isConcurrent);
14+
IDqChannelStorage::TWakeUpCallback wakeUpCb, NActors::TActorSystem* actorSystem);
1515

1616
} // namespace NYql::NDq

0 commit comments

Comments
 (0)