allow forced compaction generated by external script (blockstore-pcompact-tablets) to advance in presense of heavy tablet compaction

cloud/blockstore/libs/storage/core/compaction_options.cpp

@@ -0,0 +1 @@
+#include "compaction_options.h"

cloud/blockstore/libs/storage/core/compaction_options.h

@@ -0,0 +1,26 @@
+#pragma once
+
+#include <util/system/types.h>
+
+#include <bitset>
+
+namespace NCloud::NBlockStore::NStorage {
+
+////////////////////////////////////////////////////////////////////////////////
+
+enum class ECompactionOption: size_t
+{
+    Full,            // non-incremental compaction
+    Forced,          // compaction initiated externally
+    MaxFieldNumber,
+};
+
+constexpr size_t ToBit(ECompactionOption option)
+{
+    return static_cast<size_t>(option);
+}
+
+using TCompactionOptions =
+    std::bitset<ToBit(ECompactionOption::MaxFieldNumber)>;
+
+}   // namespace NCloud::NBlockStore::NStorage

cloud/blockstore/libs/storage/core/compaction_type.cpp

@@ -0,0 +1 @@
+#include "compaction_type.h"

cloud/blockstore/libs/storage/core/compaction_type.h

@@ -0,0 +1,15 @@
+#pragma once
+
+#include <util/system/types.h>
+
+namespace NCloud::NBlockStore::NStorage {
+
+////////////////////////////////////////////////////////////////////////////////
+
+enum class ECompactionType: ui32
+{
+    Forced,  // compaction initiated externally
+    Tablet,  // compaction initiated by tablet
+};
+
+}   // namespace NCloud::NBlockStore::NStorage

cloud/blockstore/libs/storage/core/ya.make

@@ -5,7 +5,9 @@ GENERATE_ENUM_SERIALIZATION(mount_token.h)
 SRCS(
     block_handler.cpp
     compaction_map.cpp
+    compaction_options.cpp
     compaction_policy.cpp
+    compaction_type.cpp
     config.cpp
     disk_counters.cpp
     disk_validation.cpp

cloud/blockstore/libs/storage/partition/part_actor_compaction.cpp

@@ -149,6 +149,7 @@ class TCompactionActor final
     const ui32 MaxAffectedBlocksPerCompaction;
     const IBlockDigestGeneratorPtr BlockDigestGenerator;
     const TDuration ReadBlobTimeout;
+    const ECompactionType CompactionType;
 
     const ui64 CommitId;
 
@@ -181,6 +182,7 @@ class TCompactionActor final
         ui32 maxAffectedBlocksPerCompaction,
         IBlockDigestGeneratorPtr blockDigestGenerator,
         TDuration readBlobTimeout,
+        ECompactionType compactionType,
         ui64 commitId,
         TVector<TRangeCompactionInfo> rangeCompactionInfos,
         TVector<TRequest> requests);
@@ -242,6 +244,7 @@ TCompactionActor::TCompactionActor(
         ui32 maxAffectedBlocksPerCompaction,
         IBlockDigestGeneratorPtr blockDigestGenerator,
         TDuration readBlobTimeout,
+        ECompactionType compactionType,
         ui64 commitId,
         TVector<TRangeCompactionInfo> rangeCompactionInfos,
         TVector<TRequest> requests)
@@ -253,6 +256,7 @@ TCompactionActor::TCompactionActor(
     , MaxAffectedBlocksPerCompaction(maxAffectedBlocksPerCompaction)
     , BlockDigestGenerator(std::move(blockDigestGenerator))
     , ReadBlobTimeout(readBlobTimeout)
+    , CompactionType(compactionType)
     , CommitId(commitId)
     , RangeCompactionInfos(std::move(rangeCompactionInfos))
     , Requests(std::move(requests))
@@ -736,6 +740,7 @@ void TCompactionActor::NotifyCompleted(
         request->AffectedRanges.push_back(ConvertRangeSafe(rc.BlockRange));
     }
     request->AffectedBlockInfos = std::move(AffectedBlockInfos);
+    request->CompactionType = CompactionType;
 
     NCloud::Send(ctx, Tablet, std::move(request));
 }
@@ -1029,7 +1034,9 @@ void TPartitionActor::ChangeRangeCountPerRunIfNeeded(
 
 void TPartitionActor::EnqueueCompactionIfNeeded(const TActorContext& ctx)
 {
-    if (State->GetCompactionState().Status != EOperationStatus::Idle) {
+    if (State->GetCompactionState(ECompactionType::Tablet).Status !=
+        EOperationStatus::Idle)
+    {
         // already enqueued
         return;
     }
@@ -1125,7 +1132,8 @@ void TPartitionActor::EnqueueCompactionIfNeeded(const TActorContext& ctx)
         throttlingAllowed = false;
     }
 
-    State->GetCompactionState().SetStatus(EOperationStatus::Enqueued);
+    State->GetCompactionState(ECompactionType::Tablet).SetStatus(
+        EOperationStatus::Enqueued);
 
     if (Config->GetCompactionCountPerRunIncreasingThreshold()
         && Config->GetCompactionCountPerRunDecreasingThreshold()
@@ -1176,7 +1184,7 @@ void TPartitionActor::EnqueueCompactionIfNeeded(const TActorContext& ctx)
     if (mode == TEvPartitionPrivate::GarbageCompaction
             || !diskGarbageBelowThreshold)
     {
-        request->ForceFullCompaction = true;
+        request->CompactionOptions.set(ToBit(ECompactionOption::Full));
     }
 
     if (throttlingAllowed && Config->GetMaxCompactionDelay()) {
@@ -1245,13 +1253,21 @@ void TPartitionActor::HandleCompaction(
         NCloud::Reply(ctx, requestInfo, std::move(response));
     };
 
-    if (State->GetCompactionState().Status == EOperationStatus::Started) {
+    const auto compactionType =
+        msg->CompactionOptions.test(ToBit(ECompactionOption::Forced)) ?
+        ECompactionType::Forced:
+        ECompactionType::Tablet;
+
+    if (State->GetCompactionState(compactionType).Status ==
+        EOperationStatus::Started)
+    {
         replyError(ctx, *requestInfo, E_TRY_AGAIN, "compaction already started");
         return;
     }
 
     if (!State->IsCompactionAllowed()) {
-        State->GetCompactionState().SetStatus(EOperationStatus::Idle);
+        State->GetCompactionState(compactionType).SetStatus(
+            EOperationStatus::Idle);
 
         replyError(ctx, *requestInfo, E_BS_OUT_OF_SPACE, "all channels readonly");
         return;
@@ -1287,7 +1303,8 @@ void TPartitionActor::HandleCompaction(
     }
 
     if (tops.empty() || !tops.front().Stat.BlobCount) {
-        State->GetCompactionState().SetStatus(EOperationStatus::Idle);
+        State->GetCompactionState(compactionType).SetStatus(
+            EOperationStatus::Idle);
 
         replyError(ctx, *requestInfo, S_ALREADY, "nothing to compact");
         return;
@@ -1310,9 +1327,10 @@ void TPartitionActor::HandleCompaction(
             State->GetBlocksCount() - 1);
 
         LOG_DEBUG(ctx, TBlockStoreComponents::PARTITION,
-            "[%lu] Start compaction @%lu (range: %s, blobs: %u, blocks: %u"
+            "[%lu] Start %s compaction @%lu (range: %s, blobs: %u, blocks: %u"
             ", reads: %u, blobsread: %u, blocksread: %u, score: %f)",
             TabletID(),
+            compactionType == ECompactionType::Forced ? "forced" : "tablet",
             commitId,
             DescribeRange(blockRange).c_str(),
             x.Stat.BlobCount,
@@ -1325,7 +1343,7 @@ void TPartitionActor::HandleCompaction(
         ranges.emplace_back(rangeIdx, blockRange);
     }
 
-    State->GetCompactionState().SetStatus(EOperationStatus::Started);
+    State->GetCompactionState(compactionType).SetStatus(EOperationStatus::Started);
 
     State->GetCommitQueue().AcquireBarrier(commitId);
     State->GetCleanupQueue().AcquireBarrier(commitId);
@@ -1336,7 +1354,7 @@ void TPartitionActor::HandleCompaction(
     auto tx = CreateTx<TCompaction>(
         requestInfo,
         commitId,
-        msg->ForceFullCompaction,
+        msg->CompactionOptions,
         std::move(ranges));
 
     ui64 minCommitId = State->GetCommitQueue().GetMinCommitId();
@@ -1401,7 +1419,8 @@ void TPartitionActor::HandleCompactionCompleted(
     State->GetCleanupQueue().ReleaseBarrier(commitId);
     State->GetGarbageQueue().ReleaseBarrier(commitId);
 
-    State->GetCompactionState().SetStatus(EOperationStatus::Idle);
+    State->GetCompactionState(msg->CompactionType).SetStatus(
+        EOperationStatus::Idle);
 
     Actors.Erase(ev->Sender);
 
@@ -1458,7 +1477,7 @@ void PrepareRangeCompaction(
     const TString& folderId,
     const TString& diskId,
     const ui64 commitId,
-    const bool forceFullCompaction,
+    const bool fullCompaction,
     const TActorContext& ctx,
     const ui64 tabletId,
     THashSet<TPartialBlobId, TPartialBlobIdHash>& affectedBlobIds,
@@ -1503,7 +1522,7 @@ void PrepareRangeCompaction(
 
     if (ready
             && incrementalCompactionEnabled
-            && !forceFullCompaction)
+            && !fullCompaction)
     {
         THashMap<TPartialBlobId, ui32, TPartialBlobIdHash> liveBlocks;
         for (const auto& m: args.BlockMarks) {
@@ -1885,7 +1904,7 @@ bool TPartitionActor::PrepareCompaction(
             PartitionConfig.GetFolderId(),
             PartitionConfig.GetDiskId(),
             args.CommitId,
-            args.ForceFullCompaction,
+            args.CompactionOptions.test(ToBit(ECompactionOption::Full)),
             ctx,
             TabletID(),
             affectedBlobIds,
@@ -1962,6 +1981,11 @@ void TPartitionActor::CompleteCompaction(
         Config->GetBlobStorageAsyncGetTimeoutSSD() :
         Config->GetBlobStorageAsyncGetTimeoutHDD();
 
+    const auto compactionType =
+        args.CompactionOptions.test(ToBit(ECompactionOption::Forced)) ?
+            ECompactionType::Forced:
+            ECompactionType::Tablet;
+
     auto actor = NCloud::Register<TCompactionActor>(
         ctx,
         args.RequestInfo,
@@ -1972,6 +1996,7 @@ void TPartitionActor::CompleteCompaction(
         Config->GetMaxAffectedBlocksPerCompaction(),
         BlockDigestGenerator,
         readBlobTimeout,
+        compactionType,
         args.CommitId,
         std::move(rangeCompactionInfos),
         std::move(requests));

cloud/blockstore/libs/storage/partition/part_actor_compactrange.cpp

@@ -93,7 +93,9 @@ void TForcedCompactionActor::SendCompactionRequest(const TActorContext& ctx)
     auto request = std::make_unique<TEvPartitionPrivate::TEvCompactionRequest>(
         MakeIntrusive<TCallContext>(),
         RangesToCompact[CurrentBlock],
-        true);
+        TCompactionOptions().
+            set(ToBit(ECompactionOption::Forced)).
+            set(ToBit(ECompactionOption::Full)));
 
     NCloud::Send(ctx, Tablet, std::move(request));
 }

cloud/blockstore/libs/storage/partition/part_events_private.h

@@ -6,6 +6,8 @@
 #include <cloud/blockstore/libs/diagnostics/profile_log.h>
 #include <cloud/blockstore/libs/kikimr/components.h>
 #include <cloud/blockstore/libs/kikimr/events.h>
+#include <cloud/blockstore/libs/storage/core/compaction_options.h>
+#include <cloud/blockstore/libs/storage/core/compaction_type.h>
 #include <cloud/blockstore/libs/storage/core/request_info.h>
 #include <cloud/blockstore/libs/storage/model/channel_data_kind.h>
 #include <cloud/blockstore/libs/storage/model/channel_permissions.h>
@@ -441,14 +443,19 @@ struct TEvPartitionPrivate
     {
         ECompactionMode Mode = RangeCompaction;
         TMaybe<ui32> BlockIndex;
-        bool ForceFullCompaction = false;
+        TCompactionOptions CompactionOptions;
 
         TCompactionRequest() = default;
 
-        TCompactionRequest(ui32 blockIndex, bool forceFullCompaction)
-            : Mode(RangeCompaction)
-            , BlockIndex(blockIndex)
-            , ForceFullCompaction(forceFullCompaction)
+        TCompactionRequest(
+                ui32 blockIndex,
+                TCompactionOptions compactionOptions)
+            : BlockIndex(blockIndex)
+            , CompactionOptions(compactionOptions)
+        {}
+
+        TCompactionRequest(ui32 blockIndex)
+            : TCompactionRequest(blockIndex, {})
         {}
 
         TCompactionRequest(ECompactionMode mode)
@@ -773,6 +780,17 @@ struct TEvPartitionPrivate
     {
     };
 
+
+    //
+    // CompactionCompleted
+    //
+
+    struct TCompactionCompleted
+        : TOperationCompleted
+    {
+        ECompactionType CompactionType;
+    };
+
     //
     // MetadataRebuildCompleted
     //
@@ -880,7 +898,7 @@ struct TEvPartitionPrivate
     using TEvWriteBlocksCompleted = TResponseEvent<TWriteBlocksCompleted, EvWriteBlocksCompleted>;
     using TEvZeroBlocksCompleted = TResponseEvent<TOperationCompleted, EvZeroBlocksCompleted>;
     using TEvFlushCompleted = TResponseEvent<TFlushCompleted, EvFlushCompleted>;
-    using TEvCompactionCompleted = TResponseEvent<TOperationCompleted, EvCompactionCompleted>;
+    using TEvCompactionCompleted = TResponseEvent<TCompactionCompleted, EvCompactionCompleted>;
     using TEvCollectGarbageCompleted = TResponseEvent<TOperationCompleted, EvCollectGarbageCompleted>;
     using TEvForcedCompactionCompleted = TResponseEvent<TForcedCompactionCompleted, EvForcedCompactionCompleted>;
     using TEvMetadataRebuildCompleted = TResponseEvent<TOperationCompleted, EvMetadataRebuildCompleted>;

cloud/blockstore/libs/storage/partition/part_state.cpp

@@ -1008,6 +1008,16 @@ ui64 TPartitionState::GetMixedIndexCacheMemSize() const
     return MixedIndexCacheAllocator.GetBytesAllocated();
 }
 
+////////////////////////////////////////////////////////////////////////////////
+// Compaction
+
+TOperationState& TPartitionState::GetCompactionState(ECompactionType type)
+{
+    return type == ECompactionType::Forced ?
+        ForcedCompactionState.State :
+        CompactionState;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 void TPartitionState::SetUsedBlocks(

cloud/blockstore/libs/storage/partition/part_state.h

@@ -9,6 +9,7 @@
 #include <cloud/blockstore/libs/diagnostics/downtime_history.h>
 #include <cloud/blockstore/libs/storage/api/partition.h>
 #include <cloud/blockstore/libs/storage/core/compaction_map.h>
+#include <cloud/blockstore/libs/storage/core/compaction_type.h>
 #include <cloud/blockstore/libs/storage/core/request_buffer.h>
 #include <cloud/blockstore/libs/storage/core/request_info.h>
 #include <cloud/blockstore/libs/storage/core/ts_ring_buffer.h>
@@ -78,6 +79,7 @@ struct TForcedCompactionState
     ui32 Progress = 0;
     ui32 RangesCount = 0;
     TString OperationId;
+    TOperationState State;
 };
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -800,10 +802,7 @@ class TPartitionState
     TInstant LastCompactionRangeCountPerRunTs;
 
 public:
-    TOperationState& GetCompactionState()
-    {
-        return CompactionState;
-    }
+    TOperationState& GetCompactionState(ECompactionType type);
 
     TCompactionMap& GetCompactionMap()
     {

cloud/blockstore/libs/storage/partition/part_tx.h

@@ -8,6 +8,7 @@
 #include <cloud/blockstore/libs/storage/api/partition.h>
 #include <cloud/blockstore/libs/storage/core/block_handler.h>
 #include <cloud/blockstore/libs/storage/core/compaction_map.h>
+#include <cloud/blockstore/libs/storage/core/compaction_options.h>
 #include <cloud/blockstore/libs/storage/core/request_info.h>
 #include <cloud/blockstore/libs/storage/partition/model/blob_to_confirm.h>
 #include <cloud/blockstore/libs/storage/partition/model/block.h>
@@ -430,18 +431,18 @@ struct TTxPartition
     {
         const TRequestInfoPtr RequestInfo;
         const ui64 CommitId;
-        const bool ForceFullCompaction;
+        const TCompactionOptions CompactionOptions;
 
         TVector<TRangeCompaction> RangeCompactions;
 
         TCompaction(
                 TRequestInfoPtr requestInfo,
                 ui64 commitId,
-                bool forceFullCompaction,
+                TCompactionOptions compactionOptions,
                 const TVector<std::pair<ui32, TBlockRange32>>& ranges)
             : RequestInfo(std::move(requestInfo))
             , CommitId(commitId)
-            , ForceFullCompaction(forceFullCompaction)
+            , CompactionOptions(compactionOptions)
         {
             RangeCompactions.reserve(ranges.size());
             for (const auto& range: ranges) {