Execute LogWrite and ChunkWrite from GetEvents thread in PDisk (#8984)

Co-authored-by: Vlad Kuznecov <va-kuznecov@nebius.com>
Co-authored-by: Semyon Danilov <samvimes@yandex.ru>

Author: 3 people

ydb/core/blobstorage/pdisk/blobstorage_pdisk_blockdevice_async.cpp

@@ -368,6 +368,25 @@ class TRealBlockDevice : public IBlockDevice {
             }
         }
 
+        void ExecuteOrScheduleCompletion(TCompletionAction *action) {
+            if (action->ShouldBeExecutedInCompletionThread) {
+                Device.CompletionThread->Schedule(action);
+            } else {
+                if (action->CanHandleResult()) {
+                    action->Exec(Device.PCtx->ActorSystem);
+                } else {
+                    TString errorReason = action->ErrorReason;
+
+                    action->Release(Device.PCtx->ActorSystem);
+
+                    if (!Device.QuitCounter.IsBlocked()) {
+                        Device.BecomeErrorState(TStringBuilder()
+                                << " CompletionAction error, operation info# " << errorReason);
+                    }
+                }
+            }
+        }
+
         void Exec(TAsyncIoOperationResult *event) {
             IAsyncIoOperation *op = event->Operation;
             // Add up the execution time of all the events
@@ -425,7 +444,7 @@ class TRealBlockDevice : public IBlockDevice {
                     WaitingNoops[idx % MaxWaitingNoops] = completionAction->FlushAction;
                     completionAction->FlushAction = nullptr;
                 }
-                Device.CompletionThread->Schedule(completionAction);
+                ExecuteOrScheduleCompletion(completionAction);
                 auto seqnoL6 = AtomicGetAndIncrement(Device.Mon.SeqnoL6);
                 Device.Mon.L6.Set(duration > Device.Reordering, seqnoL6);
             }
@@ -443,7 +462,7 @@ class TRealBlockDevice : public IBlockDevice {
                     LWTRACK(PDiskDeviceGetFromWaiting, WaitingNoops[i]->Orbit);
                     double durationMs = HPMilliSecondsFloat(HPNow() - WaitingNoops[i]->GetTime);
                     Device.Mon.DeviceFlushDuration.Increment(durationMs);
-                    Device.CompletionThread->Schedule(WaitingNoops[i]);
+                    ExecuteOrScheduleCompletion(WaitingNoops[i]);
                     WaitingNoops[i] = nullptr;
                 }
                 ++NextPossibleNoop;

ydb/core/blobstorage/pdisk/blobstorage_pdisk_completion.h

@@ -18,6 +18,11 @@ struct TCompletionAction {
     NWilson::TTraceId TraceId;
     EIoResult Result = EIoResult::Unknown;
     TString ErrorReason;
+    // Only reads should be executed in a separate thread since their completions consist of
+    // time-consuming deciphering of read data. But currently some completion actions can write
+    // to BlockDevice from Exec() and it's more safe to use WhiteList to allow only
+    // LogWrite and ChunkWrite to be executed from GetThread
+    bool ShouldBeExecutedInCompletionThread = true;
 
     mutable NLWTrace::TOrbit Orbit;
 protected:

ydb/core/blobstorage/pdisk/blobstorage_pdisk_completion_impl.h

@@ -93,6 +93,7 @@ class TCompletionChunkWrite : public TCompletionAction {
         , ReqId(reqId)
         , Span(std::move(span))
     {
+        TCompletionAction::ShouldBeExecutedInCompletionThread = false;
     }
 
     ~TCompletionChunkWrite() {
@@ -144,7 +145,9 @@ class TCompletionLogWrite : public TCompletionAction {
         , LogWriteQueue(std::move(logWriteQueue))
         , Commits(std::move(commits))
         , CommitedLogChunks(std::move(commitedLogChunks))
-    {}
+    {
+        TCompletionAction::ShouldBeExecutedInCompletionThread = false;
+    }
 
     TVector<ui32>* GetCommitedLogChunksPtr() {
         return &CommitedLogChunks;