Temporarily fix data race in PDisk (ydb-platform#9988)

uzhastik · Oct 3, 2024 · 6e20d54 · 6e20d54
1 parent 7982835
commit 6e20d54
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 2 deletions.
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp
@@ -2485,6 +2485,9 @@ void TPDisk::ProcessFastOperationsQueue() {
             case ERequestType::RequestPushUnformattedMetadataSector:
                 ProcessPushUnformattedMetadataSector(static_cast<TPushUnformattedMetadataSector&>(*req));
                 break;
+            case ERequestType::RequestContinueReadMetadata:
+                static_cast<TContinueReadMetadata&>(*req).Execute(PCtx->ActorSystem);
+                break;
             default:
                 Y_FAIL_S("Unexpected request type# " << (ui64)req->GetType());
                 break;
@@ -3078,6 +3081,7 @@ bool TPDisk::PreprocessRequest(TRequestBase *request) {
         case ERequestType::RequestPushUnformattedMetadataSector:
         case ERequestType::RequestReadMetadata:
         case ERequestType::RequestWriteMetadata:
+        case ERequestType::RequestContinueReadMetadata:
             break;
         case ERequestType::RequestStopDevice:
             BlockDevice->Stop();

diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_metadata.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_metadata.cpp
@@ -59,8 +59,16 @@ namespace NKikimr::NPDisk {
                         const size_t bytesToRead = PDisk->Format.RoundUpToSectorSize(sizeof(TMetadataHeader) + header->Length);
                         Buffer = TRcBuf::UninitializedPageAligned(bytesToRead); // __header is not valid anymore__
                         const ui64 offset = PDisk->Format.Offset(Req->Key.ChunkIdx, Req->Key.OffsetInSectors);
-                        PDisk->BlockDevice->PreadAsync(GetBuffer(), bytesToRead, offset, this, Req->ReqId, nullptr);
-                        return;
+                        auto buffer = GetBuffer();
+                        auto reqId = Req->ReqId;
+                        auto callback = [buffer, bytesToRead, offset, this, reqId](bool success, TActorSystem *actorSystem) {
+                            if (success) {
+                                PDisk->BlockDevice->PreadAsync(buffer, bytesToRead, offset, this, reqId, nullptr);
+                            } else {
+                                Release(actorSystem);
+                            }
+                        };
+                        return PDisk->InputRequest(PDisk->ReqCreator.CreateFromArgs<TContinueReadMetadata>(callback));
                     }
                 } else {
                     Req->ErrorReason = "header checksum does not pass validation";

diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_request_id.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_request_id.h
@@ -88,6 +88,7 @@ struct TReqId {
         WriteMetadata = 69,
         WriteMetadataResult = 70,
         PushUnformattedMetadataSector = 71,
+        ContinueReadMetadata = 72,
     };
 
     // 56 bit idx, 8 bit source
@@ -151,6 +152,7 @@ enum class ERequestType {
     RequestWriteMetadata,
     RequestWriteMetadataResult,
     RequestPushUnformattedMetadataSector,
+    RequestContinueReadMetadata,
 };
 
 inline IOutputStream& operator <<(IOutputStream& out, const TReqId& reqId) {

diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_requestimpl.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_requestimpl.h
@@ -1095,5 +1095,27 @@ class TPushUnformattedMetadataSector : public TRequestBase {
     }
 };
 
+class TContinueReadMetadata : public TRequestBase {
+    std::function<void(bool, TActorSystem*)> Callback;
+
+public:
+    TContinueReadMetadata(std::function<void(bool, TActorSystem*)> callback, TAtomicBase reqIdx)
+        : TRequestBase({}, TReqId(TReqId::ContinueReadMetadata, reqIdx), OwnerSystem, 0, NPriInternal::Other)
+        , Callback(std::move(callback))
+    {}
+
+    ERequestType GetType() const override {
+        return ERequestType::RequestContinueReadMetadata;
+    }
+
+    void Execute(TActorSystem *actorSystem) {
+        Callback(true, actorSystem);
+    }
+
+    void Abort(TActorSystem *actorSystem) override {
+        Callback(false, actorSystem);
+    }
+};
+
 } // NPDisk
 } // NKikimr