PDisk noop scheduler (#9332)

Author: va-kuznecov

ydb/core/blobstorage/lwtrace_probes/blobstorage_probes.h

@@ -156,6 +156,9 @@ struct TEventTypeField {
     PROBE(PDiskAddToScheduler, GROUPS("PDisk", "PDiskRequest"), \
       TYPES(TPDiskIdField, ui64, double, ui64, bool, ui64), \
       NAMES("pdisk", "reqId", "creationTimeSec", "owner", "isFast", "priorityClass")) \
+    PROBE(PDiskAddToNoopScheduler, GROUPS("PDisk", "PDiskRequest"), \
+      TYPES(TPDiskIdField, ui64, double, ui64, bool, ui64), \
+      NAMES("pdisk", "reqId", "creationTimeSec", "owner", "isFast", "priorityClass")) \
     PROBE(PDiskRouteRequest, GROUPS("PDisk", "PDiskRequest"), \
       TYPES(TPDiskIdField, ui64, double, ui64, bool, ui64), \
       NAMES("pdisk", "reqId", "creationTimeSec", "owner", "isFast", "priorityClass")) \
@@ -265,11 +268,16 @@ struct TEventTypeField {
       NAMES("pdisk", "entireUpdateMs", "inputQueueMs", "schedulingMs", "processingMs", "waitingMs")) \
     PROBE(PDiskEnqueueAllDetails, GROUPS("PDisk"), \
       TYPES(ui64, size_t, size_t, size_t, double), \
-      NAMES("pdisk", "initialQueueSize", "processedReqs", "pushedToForsetiReqs", "spentTimeMs")) \
+      NAMES("pdisk", "initialQueueSize", "processedReqs", "pushedToSchedulerReqs", "spentTimeMs")) \
     PROBE(PDiskUpdateStarted, GROUPS("PDisk"), TYPES(ui64), NAMES("pdisk")) \
-    PROBE(PDiskProcessLogWriteQueue, GROUPS("PDisk"), TYPES(ui64, size_t, size_t), NAMES("pdisk", "logQueueSize", "commitQueueSize")) \
-    PROBE(PDiskProcessChunkReadQueue, GROUPS("PDisk"), TYPES(ui64, size_t), NAMES("pdisk", "queueSize")) \
-    PROBE(PDiskProcessChunkWriteQueue, GROUPS("PDisk"), TYPES(ui64, size_t), NAMES("pdisk", "queueSize")) \
+    PROBE(PDiskProcessLogWriteQueue, GROUPS("PDisk"), TYPES(ui64, size_t, size_t, size_t), NAMES("pdisk", "remainingLogWritesSize", "logWritesSize", "commitsSize")) \
+    PROBE(PDiskProcessLogWriteBatch, GROUPS("PDisk"), TYPES(ui64, size_t, size_t), NAMES("pdisk", "logQueueSize", "commitQueueSize")) \
+    PROBE(PDiskProcessChunkReadQueue, GROUPS("PDisk"), \
+      TYPES(ui64, size_t, size_t, size_t, double), \
+      NAMES("pdisk", "initialQueueSize", "processed", "processedBytes", "processedCostMs")) \
+    PROBE(PDiskProcessChunkWriteQueue, GROUPS("PDisk"), \
+      TYPES(ui64, size_t, size_t, size_t, double), \
+      NAMES("pdisk", "initialQueueSize", "processed", "processedBytes", "processedCostMs")) \
     PROBE(PDiskStartWaiting, GROUPS("PDisk"), TYPES(ui64), NAMES("pdisk")) \
     PROBE(PDiskUpdateEnded, GROUPS("PDisk"), TYPES(ui64, float), NAMES("pdisk", "entireUpdateMs")) \
     PROBE(DSProxyGetEnqueue, GROUPS("DSProxy", "LWTrackStart"), TYPES(), NAMES()) \

ydb/core/blobstorage/pdisk/blobstorage_pdisk_blockdevice_async.cpp

@@ -223,7 +223,7 @@ class TRealBlockDevice : public IBlockDevice {
         TRealBlockDevice &Device;
         std::shared_ptr<TPDiskCtx> &PCtx;
         TCountedQueueOneOne<IAsyncIoOperation*, 4 << 10> OperationsToBeSubmit;
-        static constexpr TAtomicBase SubmitInFlightBytesMax = 1ull << 15;
+        static constexpr TAtomicBase SubmitInFlightBytesMax = 1ull << 17;
         TMutex SubmitMtx;
         TCondVar SubmitCondVar;
         TAtomicBlockCounter SubmitQuitCounter;

ydb/core/blobstorage/pdisk/blobstorage_pdisk_completion_impl.cpp

@@ -42,6 +42,7 @@ void TCompletionLogWrite::Exec(TActorSystem *actorSystem) {
     NHPTimer::STime now = HPNow();
     for (auto it = LogWriteQueue.begin(); it != LogWriteQueue.end(); ++it) {
         TLogWrite &evLog = *(*it);
+        evLog.Replied = true;
         TLogWrite *&batch = batchMap[evLog.Owner];
         LOG_DEBUG_S(*actorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDisk->PCtx->PDiskId
                 << " ReqId# " << evLog.ReqId.Id << " TEvLogResult Sender# " << evLog.Sender.LocalId()
@@ -98,17 +99,12 @@ void TCompletionLogWrite::Exec(TActorSystem *actorSystem) {
 }
 
 void TCompletionLogWrite::Release(TActorSystem *actorSystem) {
-    switch (Result) {
-    case EIoResult::Ok:
-    case EIoResult::Unknown:
-        break;
-    default:
-        for (TLogWrite *logWrite : LogWriteQueue) {
-            auto res = MakeHolder<TEvLogResult>(NKikimrProto::CORRUPTED, NKikimrBlobStorage::StatusIsValid,
-                    ErrorReason);
-            actorSystem->Send(logWrite->Sender, res.Release());
-            PDisk->Mon.WriteLog.CountResponse();
-        }
+    for (TLogWrite *logWrite : LogWriteQueue) {
+        auto res = MakeHolder<TEvLogResult>(NKikimrProto::CORRUPTED, NKikimrBlobStorage::StatusIsValid,
+                ErrorReason);
+        logWrite->Replied = true;
+        actorSystem->Send(logWrite->Sender, res.Release());
+        PDisk->Mon.WriteLog.CountResponse();
     }
 
     delete this;

ydb/core/blobstorage/pdisk/blobstorage_pdisk_config.h

@@ -30,6 +30,11 @@ struct TPDiskSchedulerConfig {
     ui64 LoadWeight = LoadWeightDefault;
     ui64 LowReadWeight = LowWeightDefault;
 
+    size_t MaxChunkReadsPerCycle = 16;
+    double MaxChunkReadsDurationPerCycleMs = 0.25;
+    size_t MaxChunkWritesPerCycle = 8;
+    double MaxChunkWritesDurationPerCycleMs = 1;
+
     TString ToString(bool isMultiline) const {
         const char *x = isMultiline ? "\n" : "";
         TStringStream str;
@@ -44,6 +49,10 @@ struct TPDiskSchedulerConfig {
         str << " OtherReadWeight# " << OtherReadWeight << x;
         str << " LoadWeight# " << LoadWeight << x;
         str << " LowReadWeight# " << LowReadWeight << x;
+        str << " MaxChunkReadsPerCycle# " << MaxChunkReadsPerCycle << x;
+        str << " MaxChunkReadsDurationPerCycleMs# " << MaxChunkReadsDurationPerCycleMs << x;
+        str << " MaxChunkWritesPerCycle# " << MaxChunkWritesPerCycle << x;
+        str << " MaxChunkWritesDurationPerCycleMs# " << MaxChunkWritesDurationPerCycleMs << x;
         str << "}" << x;
         return str.Str();
     }
@@ -156,6 +165,7 @@ struct TPDiskConfig : public TThrRefBase {
     NKikimrBlobStorage::TPDiskSpaceColor::E SpaceColorBorder = NKikimrBlobStorage::TPDiskSpaceColor::GREEN;
 
     ui32 CompletionThreadsCount = 1;
+    bool UseNoopScheduler = false;
 
     bool MetadataOnly = false;
 
@@ -312,6 +322,7 @@ struct TPDiskConfig : public TThrRefBase {
         str << " MaxMetadataMegabytes# " << MaxMetadataMegabytes << x;
         str << " SpaceColorBorder# " << SpaceColorBorder << x;
         str << " CompletionThreadsCount# " << CompletionThreadsCount << x;
+        str << " UseNoopScheduler# " << (UseNoopScheduler ? "true" : "false") << x;
         str << "}";
         return str.Str();
     }
@@ -400,6 +411,10 @@ struct TPDiskConfig : public TThrRefBase {
         if (cfg->HasCompletionThreadsCount()) {
             CompletionThreadsCount = cfg->GetCompletionThreadsCount();
         }
+
+        if (cfg->HasUseNoopScheduler()) {
+            UseNoopScheduler = cfg->GetUseNoopScheduler();
+        }
     }
 };