Fixed fault

Author: GrigoriyPA

ydb/core/tx/conveyor/service/service.cpp

@@ -10,7 +10,7 @@ TWorkersPool::TWorkersPool(const TString& conveyorName, const NActors::TActorId&
     Workers.reserve(WorkersCount);
     for (ui32 i = 0; i < WorkersCount; ++i) {
         const auto usage = config.GetWorkerCPUUsage(i);
-        Workers.emplace_back(std::make_unique<TWorker>(conveyorName, usage, distributorId, i, Counters.SendFwdHistogram, Counters.SendFwdDuration));
+        Workers.emplace_back(usage, std::make_unique<TWorker>(conveyorName, usage, distributorId, i, Counters.SendFwdHistogram, Counters.SendFwdDuration));
         MaxWorkerThreads += usage;
     }
     AFL_VERIFY(WorkersCount)("name", conveyorName)("action", "conveyor_registered")("config", config.DebugString())("actor_id", distributorId)("count", WorkersCount);
@@ -46,9 +46,6 @@ void TWorkersPool::ReleaseWorker(const ui32 workerIdx) {
 
 void TWorkersPool::ChangeAmountCPULimit(const double delta) {
     AmountCPULimit += delta;
-    if (std::abs(AmountCPULimit) < Eps) {
-        AmountCPULimit = 0;
-    }
     AFL_VERIFY(AmountCPULimit >= 0);
     Counters.AmountCPULimit->Set(AmountCPULimit);
     Counters.ChangeCPULimitRate->Inc();
@@ -61,15 +58,15 @@ void TWorkersPool::ChangeAmountCPULimit(const double delta) {
     ActiveWorkersCount = 0;
     ActiveWorkerThreads = numberThreads;
     ActiveWorkersIdx.clear();
-    for (const auto& worker : Workers) {
+    for (auto& worker : Workers) {
         if (numberThreads <= 0) {
             break;
         }
         if (!worker.GetRunningTask()) {
             ActiveWorkersIdx.emplace_back(ActiveWorkersCount);
         }
-        worker.GetWorker()->UpdateCPUSoftLimit(std::min<double>(numberThreads, 1));
-        numberThreads -= worker.GetWorker()->GetCPUSoftLimit();
+        worker.ChangeCPUSoftLimit(std::min<double>(numberThreads, 1));
+        numberThreads -= worker.GetCPUSoftLimit();
         ++ActiveWorkersCount;
     }
     AFL_VERIFY(std::abs(numberThreads) < Eps);

ydb/core/tx/conveyor/service/service.h

@@ -188,14 +188,21 @@ class TProcess {
 class TWorkersPool {
     class TWorkerInfo {
         YDB_READONLY(bool, RunningTask, false);
-        YDB_READONLY(TWorker*, Worker, nullptr);
+        YDB_READONLY(double, CPUSoftLimit, 0.0);
         YDB_READONLY_DEF(TActorId, WorkerId);
     public:
-        explicit TWorkerInfo(std::unique_ptr<TWorker> worker)
-            : Worker(worker.get())
+        TWorkerInfo(const double cpuSoftLimit, std::unique_ptr<TWorker> worker)
+            : CPUSoftLimit(cpuSoftLimit)
             , WorkerId(TActivationContext::Register(worker.release())) {
         }
 
+        void ChangeCPUSoftLimit(const double newCPUSoftLimit) {
+            if (std::abs(newCPUSoftLimit - CPUSoftLimit) > Eps) {
+                TActivationContext::Send(WorkerId, std::make_unique<TEvInternal::TEvChangeCPUSoftLimit>(newCPUSoftLimit));
+                CPUSoftLimit = newCPUSoftLimit;
+            }
+        }
+
         void OnStartTask() {
             AFL_VERIFY(!RunningTask);
             RunningTask = true;

ydb/core/tx/conveyor/service/worker.cpp

@@ -62,7 +62,8 @@ void TWorker::HandleMain(TEvInternal::TEvNewTask::TPtr& ev) {
     ExecuteTask(ev->Get()->ExtractTasks());
 }
 
-void TWorker::UpdateCPUSoftLimit(const double cpuSoftLimit) {
+void TWorker::HandleMain(TEvInternal::TEvChangeCPUSoftLimit::TPtr& ev) {
+    const auto cpuSoftLimit = ev->Get()->GetCPUSoftLimit();
     AFL_VERIFY(0 < cpuSoftLimit);
     AFL_VERIFY(cpuSoftLimit <= CPUHardLimit);
     CPUSoftLimit = cpuSoftLimit;
@@ -71,9 +72,6 @@ void TWorker::UpdateCPUSoftLimit(const double cpuSoftLimit) {
         return;
     }
 
-    Sleep(TDuration::MilliSeconds(10));
-
-    AFL_VERIFY(!Instants.empty());
     auto wakeupTime = Instants.back();
     if (CPUSoftLimit < 1) {
         wakeupTime += GetWakeupDuration();

ydb/core/tx/conveyor/service/worker.h

@@ -44,6 +44,7 @@ struct TEvInternal {
     enum EEv {
         EvNewTask = EventSpaceBegin(NActors::TEvents::ES_PRIVATE),
         EvTaskProcessedResult,
+        EvChangeCPUSoftLimit,
         EvEnd
     };
 
@@ -84,6 +85,15 @@ struct TEvInternal {
             AFL_VERIFY(Instants.size() == ProcessIds.size() + 1);
         }
     };
+
+    class TEvChangeCPUSoftLimit: public NActors::TEventLocal<TEvChangeCPUSoftLimit, EvChangeCPUSoftLimit> {
+    private:
+        YDB_READONLY(double, CPUSoftLimit, 0.0);
+    public:
+        explicit TEvChangeCPUSoftLimit(const double cpuSoftLimit)
+            : CPUSoftLimit(cpuSoftLimit) {
+        }
+    };
 };
 
 class TWorker: public NActors::TActorBootstrapped<TWorker> {
@@ -104,13 +114,15 @@ class TWorker: public NActors::TActorBootstrapped<TWorker> {
     void ExecuteTask(std::vector<TWorkerTask>&& workerTasks);
     void HandleMain(TEvInternal::TEvNewTask::TPtr& ev);
     void HandleMain(NActors::TEvents::TEvWakeup::TPtr& ev);
+    void HandleMain(TEvInternal::TEvChangeCPUSoftLimit::TPtr& ev);
     void OnWakeup();
 public:
 
     STATEFN(StateMain) {
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvInternal::TEvNewTask, HandleMain);
             hFunc(NActors::TEvents::TEvWakeup, HandleMain);
+            hFunc(TEvInternal::TEvChangeCPUSoftLimit, HandleMain);
             default:
                 ALS_ERROR(NKikimrServices::TX_CONVEYOR) << "unexpected event for task executor: " << ev->GetTypeRewrite();
                 break;
@@ -132,8 +144,6 @@ class TWorker: public NActors::TActorBootstrapped<TWorker> {
         AFL_VERIFY(0 < CPUHardLimit);
         AFL_VERIFY(CPUHardLimit <= 1);
     }
-
-    void UpdateCPUSoftLimit(const double cpuSoftLimit);
 };
 
 }