Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

additional logging and checker #15976

Draft
wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

additional logging and checker #15976

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bc76ca8
additional logging and checker
ivanmorozov333 Mar 19, 2025
d92f681
fix
ivanmorozov333 Mar 19, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion ydb/core/kqp/compute_actor/kqp_scan_compute_manager.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,7 @@ class TInFlightComputes {
private:
YDB_READONLY_DEF(NActors::TActorId, ActorId);
YDB_READONLY(ui64, FreeSpace, 0);
YDB_READONLY(TMonotonic, LastAckInstant, TMonotonic::Now());
std::deque<std::unique_ptr<TComputeTaskData>> DataQueue;

bool SendData() {
Expand All @@ -190,6 +191,7 @@ class TInFlightComputes {
}

void OnAckReceived(const ui64 freeSpace) {
LastAckInstant = TMonotonic::Now();
AFL_ENSURE(!FreeSpace);
AFL_ENSURE(freeSpace);
FreeSpace = freeSpace;
Expand All @@ -209,9 +211,11 @@ class TInFlightComputes {

private:
std::deque<std::unique_ptr<TComputeTaskData>> UndefinedShardTaskData;
std::deque<TComputeActorInfo> ComputeActors;
YDB_READONLY_DEF(std::deque<TComputeActorInfo>, ComputeActors);
YDB_READONLY(TMonotonic, StartInstant, TMonotonic::Now());
THashMap<TActorId, TComputeActorInfo*> ComputeActorsById;
public:

ui32 GetPacksToSendCount() const {
ui32 result = UndefinedShardTaskData.size();
for (auto&& i : ComputeActors) {
Expand All @@ -220,6 +224,15 @@ class TInFlightComputes {
return result;
}

THashMap<NActors::TActorId, ui32> GetPacksToSend() const {
THashMap<NActors::TActorId, ui32> result;
result.emplace(NActors::TActorId(), UndefinedShardTaskData.size());
for (auto&& i : ComputeActors) {
AFL_ENSURE(result.emplace(i.GetActorId(), i.GetPacksToSendCount()).second);
}
return result;
}

TInFlightComputes(const std::vector<NActors::TActorId>& computeActorIds) {
for (auto&& i : computeActorIds) {
ComputeActors.emplace_back(i);
Expand Down
30 changes: 30 additions & 0 deletions ydb/core/kqp/compute_actor/kqp_scan_fetcher_actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,9 +81,39 @@ void TKqpScanFetcherActor::Bootstrap() {
}
AFL_DEBUG(NKikimrServices::KQP_COMPUTE)("event", "bootstrap")("compute", ComputeActorIds.size())("shards", PendingShards.size());
StartTableScan();
Schedule(TDuration::Seconds(30), new NActors::TEvents::TEvWakeup);
Become(&TKqpScanFetcherActor::StateFunc);
}

void TKqpScanFetcherActor::HandleExecute(NActors::TEvents::TEvWakeup::TPtr& /*ev*/) {
const TMonotonic now = TMonotonic::Now();
TMonotonic maxInstant = TMonotonic::Zero();
for (auto&& i : InFlightComputes.GetComputeActors()) {
if (maxInstant < i.GetLastAckInstant()) {
maxInstant = i.GetLastAckInstant();
}
}
TStringBuilder sb;
sb << "[";
for (auto&& i : InFlightComputes.GetPacksToSend()) {
sb << "{" << i.first << ":" << i.second << "}";
}
sb << "];";
sb << "[";
for (auto&& i : InFlightComputes.GetComputeActors()) {
sb << "{" << i.GetActorId() << ":" << now - i.GetLastAckInstant() << ":" << i.IsFree() << "}";
}
sb << "]";
if (TDuration::Seconds(120) < now - maxInstant) {
AFL_ENSURE(false)("in_flight", sb)("count", InFlightComputes.GetComputeActors().size())
("shards", InFlightShards.GetShardsCount())("scans", InFlightShards.GetScansCount())("duration", now - InFlightComputes.GetStartInstant());
} else {
AFL_INFO(NKikimrServices::KQP_COMPUTE)("in_flight", sb)("count", InFlightComputes.GetComputeActors().size())
("shards", InFlightShards.GetShardsCount())("scans", InFlightShards.GetScansCount())("duration", now - InFlightComputes.GetStartInstant());
}
Schedule(TDuration::Seconds(30), new NActors::TEvents::TEvWakeup);
}

void TKqpScanFetcherActor::HandleExecute(TEvScanExchange::TEvAckData::TPtr& ev) {
AFL_ENSURE(ev->Get()->GetFreeSpace());
AFL_DEBUG(NKikimrServices::KQP_COMPUTE)("event", "AckDataFromCompute")("self_id", SelfId())("scan_id", ScanId)
Expand Down
3 changes: 2 additions & 1 deletion ydb/core/kqp/compute_actor/kqp_scan_fetcher_actor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ class TKqpScanFetcherActor: public NActors::TActorBootstrapped<TKqpScanFetcherAc
hFunc(TEvInterconnect::TEvNodeDisconnected, HandleExecute);
hFunc(TEvScanExchange::TEvTerminateFromCompute, HandleExecute);
hFunc(TEvScanExchange::TEvAckData, HandleExecute);
hFunc(NActors::TEvents::TEvWakeup, HandleExecute);
IgnoreFunc(TEvInterconnect::TEvNodeConnected);
IgnoreFunc(TEvTxProxySchemeCache::TEvInvalidateTableResult);
default:
Expand All @@ -92,7 +93,7 @@ class TKqpScanFetcherActor: public NActors::TActorBootstrapped<TKqpScanFetcherAc
StopOnError("unexpected exception: " + CurrentExceptionMessage());
}
}

void HandleExecute(NActors::TEvents::TEvWakeup::TPtr& ev);
void HandleExecute(TEvScanExchange::TEvAckData::TPtr& ev);

void HandleExecute(TEvScanExchange::TEvTerminateFromCompute::TPtr& ev);
Expand Down
Loading