Skip to content

Commit 920ac4a

Browse files
kruallkruall
authored
fix walking back start of elapsing time (#4578)
1 parent f275821 commit 920ac4a

File tree

5 files changed

+39
-20
lines changed

5 files changed

+39
-20
lines changed

‎ydb/library/actors/core/executor_pool_io.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,15 +36,15 @@ namespace NActors {
3636
ThreadQueue.Push(workerId + 1, revolvingCounter);
3737

3838
NHPTimer::STime hpnow = GetCycleCountFast();
39-
NHPTimer::STime hpprev = TlsThreadContext->StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
39+
NHPTimer::STime hpprev = TlsThreadContext->UpdateStartOfElapsingTime(hpnow);
4040
TlsThreadContext->ElapsingActorActivity.store(Max<ui64>(), std::memory_order_release);
4141
wctx.AddElapsedCycles(ActorSystemIndex, hpnow - hpprev);
4242

4343
if (threadCtx.WaitingPad.Park())
4444
return 0;
4545

4646
hpnow = GetCycleCountFast();
47-
hpprev = TlsThreadContext->StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
47+
hpprev = TlsThreadContext->UpdateStartOfElapsingTime(hpnow);
4848
TlsThreadContext->ElapsingActorActivity.store(ActorSystemIndex, std::memory_order_release);
4949
wctx.AddParkedCycles(hpnow - hpprev);
5050
}

‎ydb/library/actors/core/executor_thread.cpp‎

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -199,16 +199,16 @@ namespace NActors {
199199
bool firstEvent = true;
200200
bool preempted = false;
201201
bool wasWorking = false;
202-
NHPTimer::STime hpnow = Ctx.HPStart;
203-
NHPTimer::STime hpprev = TlsThreadContext->StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
202+
NHPTimer::STime hpnow = Ctx.HPStart;
203+
NHPTimer::STime hpprev = TlsThreadContext->UpdateStartOfElapsingTime(hpnow);
204204
Ctx.AddElapsedCycles(ActorSystemIndex, hpnow - hpprev);
205-
hpprev = Ctx.HPStart;
205+
NHPTimer::STime eventStart = Ctx.HPStart;
206206

207207
for (; Ctx.ExecutedEvents < Ctx.EventsPerMailbox; ++Ctx.ExecutedEvents) {
208208
if (TAutoPtr<IEventHandle> evExt = mailbox->Pop()) {
209209
mailbox->ProcessEvents(mailbox);
210210
recipient = evExt->GetRecipientRewrite();
211-
TActorContext ctx(*mailbox, *this, hpprev, recipient);
211+
TActorContext ctx(*mailbox, *this, eventStart, recipient);
212212
TlsActivationContext = &ctx; // ensure dtor (if any) is called within actor system
213213
// move for destruct before ctx;
214214
auto ev = std::move(evExt);
@@ -250,7 +250,7 @@ namespace NActors {
250250
actor->Receive(ev);
251251

252252
hpnow = GetCycleCountFast();
253-
hpprev = TlsThreadContext->StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
253+
hpprev = TlsThreadContext->UpdateStartOfElapsingTime(hpnow);
254254

255255
mailbox->ProcessEvents(mailbox);
256256
actor->OnDequeueEvent();
@@ -265,8 +265,9 @@ namespace NActors {
265265

266266
if (mailbox->IsEmpty()) // was not-free and become free, we must reclaim mailbox
267267
reclaimAsFree = true;
268-
269-
NHPTimer::STime elapsed = Ctx.AddEventProcessingStats(hpprev, hpnow, activityType, CurrentActorScheduledEventsCounter);
268+
269+
Ctx.AddElapsedCycles(activityType, hpnow - hpprev);
270+
NHPTimer::STime elapsed = Ctx.AddEventProcessingStats(eventStart, hpnow, activityType, CurrentActorScheduledEventsCounter);
270271
if (elapsed > 1000000) {
271272
LwTraceSlowEvent(ev.Get(), evTypeForTracing, actorType, Ctx.PoolId, CurrentRecipient, NHPTimer::GetSeconds(elapsed) * 1000.0);
272273
}
@@ -286,9 +287,10 @@ namespace NActors {
286287
Ctx.IncrementNonDeliveredEvents();
287288
}
288289
hpnow = GetCycleCountFast();
289-
hpprev = TlsThreadContext->StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
290+
hpprev = TlsThreadContext->UpdateStartOfElapsingTime(hpnow);
290291
Ctx.AddElapsedCycles(ActorSystemIndex, hpnow - hpprev);
291292
}
293+
eventStart = hpnow;
292294

293295
if (TlsThreadContext->CapturedType == ESendingType::Tail) {
294296
AtomicStore(&mailbox->ScheduleMoment, hpnow);
@@ -778,7 +780,7 @@ namespace NActors {
778780
void TGenericExecutorThread::GetCurrentStats(TExecutorThreadStats& statsCopy) const {
779781
NHPTimer::STime hpnow = GetCycleCountFast();
780782
ui64 activityType = TlsThreadCtx.ElapsingActorActivity.load(std::memory_order_acquire);
781-
NHPTimer::STime hpprev = TlsThreadCtx.StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
783+
NHPTimer::STime hpprev = TlsThreadCtx.UpdateStartOfElapsingTime(hpnow);
782784
if (activityType == Max<ui64>()) {
783785
Ctx.AddParkedCycles(hpnow - hpprev);
784786
} else {
@@ -790,7 +792,7 @@ namespace NActors {
790792
void TGenericExecutorThread::GetSharedStats(i16 poolId, TExecutorThreadStats &statsCopy) const {
791793
NHPTimer::STime hpnow = GetCycleCountFast();
792794
ui64 activityType = TlsThreadCtx.ElapsingActorActivity.load(std::memory_order_acquire);
793-
NHPTimer::STime hpprev = TlsThreadCtx.StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
795+
NHPTimer::STime hpprev = TlsThreadCtx.UpdateStartOfElapsingTime(hpnow);
794796
if (activityType == Max<ui64>()) {
795797
Ctx.AddParkedCycles(hpnow - hpprev);
796798
} else {

‎ydb/library/actors/core/executor_thread_ctx.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -97,14 +97,14 @@ namespace NActors {
9797
}
9898

9999
NHPTimer::STime hpnow = GetCycleCountFast();
100-
NHPTimer::STime hpprev = TlsThreadContext->StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
100+
NHPTimer::STime hpprev = TlsThreadContext->UpdateStartOfElapsingTime(hpnow);
101101
TlsThreadContext->ElapsingActorActivity.store(Max<ui64>(), std::memory_order_release);
102102
TlsThreadContext->WorkerCtx->AddElapsedCycles(TlsThreadContext->ActorSystemIndex, hpnow - hpprev);
103103
do {
104104
if (WaitingPad.Park()) // interrupted
105105
return true;
106106
hpnow = GetCycleCountFast();
107-
hpprev = TlsThreadContext->StartOfElapsingTime.exchange(hpnow, std::memory_order_acq_rel);
107+
hpprev = TlsThreadContext->UpdateStartOfElapsingTime(hpnow);
108108
TlsThreadContext->WorkerCtx->AddParkedCycles(hpnow - hpprev);
109109
state = GetState<TWaitState>();
110110
} while (static_cast<EThreadState>(state) == EThreadState::Sleep && !stopFlag->load(std::memory_order_relaxed));

‎ydb/library/actors/core/thread_context.h‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
#include "defs.h"
44

5+
#include <atomic>
56
#include <ydb/library/actors/util/datetime.h>
67
#include <ydb/library/actors/util/mpmc_ring_queue.h>
78

@@ -29,10 +30,23 @@ namespace NActors {
2930
bool IsCurrentRecipientAService = false;
3031
TMPMCRingQueue<20>::EPopMode ActivationPopMode = TMPMCRingQueue<20>::EPopMode::ReallySlow;
3132

32-
std::atomic<ui64> StartOfElapsingTime = 0;
33+
std::atomic<i64> StartOfElapsingTime = 0;
3334
std::atomic<ui64> ElapsingActorActivity = 0;
3435
TWorkerContext *WorkerCtx = nullptr;
3536
ui32 ActorSystemIndex = 0;
37+
38+
ui64 UpdateStartOfElapsingTime(i64 newValue) {
39+
i64 oldValue = StartOfElapsingTime.load(std::memory_order_acquire);
40+
for (;;) {
41+
if (newValue - oldValue <= 0) {
42+
break;
43+
}
44+
if (StartOfElapsingTime.compare_exchange_strong(oldValue, newValue, std::memory_order_acq_rel)) {
45+
break;
46+
}
47+
}
48+
return oldValue;
49+
}
3650
};
3751

3852
extern Y_POD_THREAD(TThreadContext*) TlsThreadContext; // in actor.cpp

‎ydb/library/actors/core/worker_context.h‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -51,13 +51,17 @@ namespace NActors {
5151
}
5252

5353
void AddElapsedCycles(ui32 activityType, i64 elapsed) {
54-
Y_DEBUG_ABORT_UNLESS(activityType < Stats->MaxActivityType());
55-
RelaxedStore(&Stats->ElapsedTicks, RelaxedLoad(&Stats->ElapsedTicks) + elapsed);
56-
RelaxedStore(&Stats->ElapsedTicksByActivity[activityType], RelaxedLoad(&Stats->ElapsedTicksByActivity[activityType]) + elapsed);
54+
if (Y_LIKELY(elapsed > 0)) {
55+
Y_DEBUG_ABORT_UNLESS(activityType < Stats->MaxActivityType());
56+
RelaxedStore(&Stats->ElapsedTicks, RelaxedLoad(&Stats->ElapsedTicks) + elapsed);
57+
RelaxedStore(&Stats->ElapsedTicksByActivity[activityType], RelaxedLoad(&Stats->ElapsedTicksByActivity[activityType]) + elapsed);
58+
}
5759
}
5860

5961
void AddParkedCycles(i64 elapsed) {
60-
RelaxedStore(&Stats->ParkedTicks, RelaxedLoad(&Stats->ParkedTicks) + elapsed);
62+
if (Y_LIKELY(elapsed > 0)) {
63+
RelaxedStore(&Stats->ParkedTicks, RelaxedLoad(&Stats->ParkedTicks) + elapsed);
64+
}
6165
}
6266

6367
void AddBlockedCycles(i64 elapsed) {
@@ -126,7 +130,6 @@ namespace NActors {
126130
RelaxedStore(&Stats->ReceivedEvents, RelaxedLoad(&Stats->ReceivedEvents) + 1);
127131
RelaxedStore(&Stats->ReceivedEventsByActivity[activityType], RelaxedLoad(&Stats->ReceivedEventsByActivity[activityType]) + 1);
128132
RelaxedStore(&Stats->ScheduledEventsByActivity[activityType], RelaxedLoad(&Stats->ScheduledEventsByActivity[activityType]) + scheduled);
129-
AddElapsedCycles(activityType, elapsed);
130133
return elapsed;
131134
}
132135

0 commit comments

Comments
 (0)