Skip to content

Commit

Permalink
NBS-4826 send message about long running blob operations to TPartitio…
Browse files Browse the repository at this point in the history 
…nActor and TVolumeActor (#151) (#192)

* NBS-4826 send message about long running blob operations to TPartitionActor and TVolumeActor

* NBS-4826 Fix review issues

* Remove unnecessary comments
  • Loading branch information
@drbasic
drbasic authored Jan 22, 2024
1 parent 921542a commit c717a58
Show file tree
Hide file tree
Showing 12 changed files with 194 additions and 74 deletions.
1 change: 1 addition & 0 deletions cloud/blockstore/libs/storage/partition/part_actor_readblob.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ void TPartitionActor::HandleReadBlob(
auto readBlobActor = std::make_unique<TReadBlobActor>(
requestInfo,
SelfId(),
VolumeActorId,
TabletID(),
State->GetBlockSize(),
StorageAccessMode,
Expand Down
14 changes: 10 additions & 4 deletions cloud/blockstore/libs/storage/partition/part_actor_writeblob.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ class TWriteBlobActor final
public:
TWriteBlobActor(
const TActorId& tabletActorId,
const TActorId& volumeActorId,
TRequestInfoPtr requestInfo,
ui64 tabletId,
std::unique_ptr<TRequest> request,
Expand Down Expand Up @@ -84,13 +85,15 @@ class TWriteBlobActor final

TWriteBlobActor::TWriteBlobActor(
const TActorId& tabletActorId,
const TActorId& volumeActorId,
TRequestInfoPtr requestInfo,
ui64 tabletId,
std::unique_ptr<TRequest> request,
TDuration longRunningThreshold,
ui32 groupId)
: TLongRunningOperationCompanion(
tabletActorId,
volumeActorId,
longRunningThreshold,
TLongRunningOperationCompanion::EOperation::WriteBlob,
groupId)
Expand Down Expand Up @@ -188,8 +191,13 @@ void TWriteBlobActor::ReplyAndDie(
RequestInfo->CallContext->RequestId);
}

if (HasError(response->GetError())) {
TLongRunningOperationCompanion::RequestCancelled(ctx);
} else {
TLongRunningOperationCompanion::RequestFinished(ctx);
}

NCloud::Reply(ctx, *RequestInfo, std::move(response));
TLongRunningOperationCompanion::RequestFinished(ctx);
Die(ctx);
}

Expand Down Expand Up @@ -340,6 +348,7 @@ void TPartitionActor::HandleWriteBlob(
channel,
std::make_unique<TWriteBlobActor>(
SelfId(),
VolumeActorId,
requestInfo,
TabletID(),
std::unique_ptr<TEvPartitionPrivate::TEvWriteBlobRequest>(
Expand Down Expand Up @@ -450,9 +459,6 @@ void TPartitionActor::HandleLongRunningBlobOperation(
State->RegisterDowntime(ctx.Now(), msg.GroupId);
Actors.MarkLongRunning(ev->Sender, msg.Operation);
}

// Forward request to TVolumeActor
ctx.Send(ev->Forward(VolumeActorId));
}

} // namespace NCloud::NBlockStore::NStorage::NPartition
133 changes: 97 additions & 36 deletions cloud/blockstore/libs/storage/partition/part_ut.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10252,7 +10252,8 @@ Y_UNIT_TEST_SUITE(TPartitionTest)
FSend sendRequest,
FReceive receiveResponse,
TEvPartitionCommonPrivate::TEvLongRunningOperation::EOperation
expectedOperation)
expectedOperation,
bool killPartition)
{
auto config = DefaultConfig();

Expand Down Expand Up @@ -10296,34 +10297,39 @@ Y_UNIT_TEST_SUITE(TPartitionTest)

// Make handler for intercepting EvLongRunningOperation message.
// Attention! counters will be doubled, because we will intercept the
// initial request, and forwarded to TVolumeActor.
// requests sent to TPartitionActor and TVolumeActor.
ui32 longRunningBeginCount = 0;
ui32 longRunningFinishCount = 0;
ui32 longRunningPingCount = 0;
ui32 longRunningCanceledCount = 0;

auto takeCounters = [&](TTestActorRuntimeBase& runtime, TAutoPtr<IEventHandle>& event)
{
using TEvLongRunningOperation =
TEvPartitionCommonPrivate::TEvLongRunningOperation;
switch (event->GetTypeRewrite()) {
case TEvPartitionCommonPrivate::EvLongRunningOperation: {
auto* msg = event->Get<TEvLongRunningOperation>();
UNIT_ASSERT_VALUES_EQUAL(
expectedOperation,
msg->Operation);
using EReason = TEvLongRunningOperation::EReason;

if (event->GetTypeRewrite() ==
TEvPartitionCommonPrivate::EvLongRunningOperation)
{
auto* msg = event->Get<TEvLongRunningOperation>();

if (msg->Reason ==
TEvLongRunningOperation::EReason::LongRunningDetected)
{
UNIT_ASSERT_VALUES_EQUAL(expectedOperation, msg->Operation);

switch (msg->Reason) {
case EReason::LongRunningDetected:
if (msg->FirstNotify) {
longRunningBeginCount++;
} else {
longRunningPingCount++;
}
} else {
break;
case EReason::Finished:
longRunningFinishCount++;
}
break;
break;
case EReason::Cancelled:
longRunningCanceledCount++;
break;
}
}
return TTestActorRuntime::DefaultObserverFunc(runtime, event);
Expand Down Expand Up @@ -10368,9 +10374,23 @@ Y_UNIT_TEST_SUITE(TPartitionTest)
runtime->DispatchEvents(options, TDuration::Seconds(1));
}

// Returning stolen requests to complete the execution of the request.
for (auto& request: stolenRequests) {
runtime->Send(request.release());
if (killPartition) {
partition.KillTablet();
} else {
// Returning stolen requests to complete the execution of the
// request.
for (auto& request: stolenRequests) {
runtime->Send(request.release());
}
}

// Wait for EvLongRunningOperation (finish or cancel) arrival.
{
TDispatchOptions options;
options.FinalEvents.emplace_back(
TEvPartitionCommonPrivate::EvLongRunningOperation);
runtime->AdvanceCurrentTime(TDuration::Seconds(60));
runtime->DispatchEvents(options, TDuration::Seconds(1));
}

// Wait for background operations completion.
Expand All @@ -10380,72 +10400,113 @@ Y_UNIT_TEST_SUITE(TPartitionTest)
{
auto response = receiveResponse(partition);
UNIT_ASSERT_VALUES_EQUAL_C(
S_OK,
killPartition ? E_REJECTED : S_OK,
response->GetStatus(),
response->GetErrorReason());
}

// Wait for EvVolumePartCounters arrived.
{
TDispatchOptions options;
options.FinalEvents.emplace_back(
TEvStatsService::EvVolumePartCounters);
runtime->DispatchEvents(options);

UNIT_ASSERT_VALUES_EQUAL(2, longRunningBeginCount);
UNIT_ASSERT_VALUES_EQUAL(4, longRunningPingCount);
UNIT_ASSERT_VALUES_EQUAL(
2,
longRunningBeginCount);
UNIT_ASSERT_VALUES_EQUAL(
2,
killPartition ? 0 : 2,
longRunningFinishCount);
UNIT_ASSERT_VALUES_EQUAL(
4,
longRunningPingCount);
killPartition ? 2 : 0,
longRunningCanceledCount);
}

// smoke test for monpage
auto channelsTab = partition.RemoteHttpInfo(
BuildRemoteHttpQuery(TestTabletId, {}, "Channels"),
HTTP_METHOD::HTTP_METHOD_GET);
if (!killPartition) {
// smoke test for monpage
auto channelsTab = partition.RemoteHttpInfo(
BuildRemoteHttpQuery(TestTabletId, {}, "Channels"),
HTTP_METHOD::HTTP_METHOD_GET);

UNIT_ASSERT_C(channelsTab->Html.Contains("svg"), channelsTab->Html);
UNIT_ASSERT_C(channelsTab->Html.Contains("svg"), channelsTab->Html);
}
}

Y_UNIT_TEST(ShouldReportLongRunningReadBlobOperations)
{
auto sendRequest = [](TPartitionClient& partition)
{
partition.SendReadBlocksRequest(0); //
partition.SendReadBlocksRequest(0);
};
auto receiveResponse = [](TPartitionClient& partition)
{
return partition.RecvReadBlocksResponse(); //
return partition.RecvReadBlocksResponse();
};

DoShouldReportLongRunningBlobOperations(
sendRequest,
receiveResponse,
TEvPartitionCommonPrivate::TEvLongRunningOperation::EOperation::
ReadBlob);
ReadBlob,
false);
}

Y_UNIT_TEST(ShouldReportLongRunningWriteBlobOperations)
{
auto sendRequest = [](TPartitionClient& partition)
{
partition.SendWriteBlocksRequest(
TBlockRange32::WithLength(0, 255)); //
TBlockRange32::WithLength(0, 255));
};
auto receiveResponse = [](TPartitionClient& partition)
{
return partition.RecvWriteBlocksResponse();
};

DoShouldReportLongRunningBlobOperations(
sendRequest,
receiveResponse,
TEvPartitionCommonPrivate::TEvLongRunningOperation::EOperation::
WriteBlob,
false);
}

Y_UNIT_TEST(ShouldReportLongRunningReadBlobOperationCancel)
{
auto sendRequest = [](TPartitionClient& partition)
{
partition.SendReadBlocksRequest(0);
};
auto receiveResponse = [](TPartitionClient& partition)
{
return partition.RecvReadBlocksResponse();
};

DoShouldReportLongRunningBlobOperations(
sendRequest,
receiveResponse,
TEvPartitionCommonPrivate::TEvLongRunningOperation::EOperation::
ReadBlob,
true);
}

Y_UNIT_TEST(ShouldReportLongRunningWriteBlobOperationCancel)
{
auto sendRequest = [](TPartitionClient& partition)
{
partition.SendWriteBlocksRequest(
TBlockRange32::WithLength(0, 255));
};
auto receiveResponse = [](TPartitionClient& partition)
{
return partition.RecvWriteBlocksResponse(); //
return partition.RecvWriteBlocksResponse();
};

DoShouldReportLongRunningBlobOperations(
sendRequest,
receiveResponse,
TEvPartitionCommonPrivate::TEvLongRunningOperation::EOperation::
WriteBlob);
WriteBlob,
true);
}
}

Expand Down
25 changes: 16 additions & 9 deletions cloud/blockstore/libs/storage/partition_common/actor_read_blob.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,20 +15,22 @@ LWTRACE_USING(BLOCKSTORE_STORAGE_PROVIDER);

TReadBlobActor::TReadBlobActor(
TRequestInfoPtr requestInfo,
const TActorId& tablet,
ui64 tabletId,
const TActorId& partitionActorId,
const TActorId& volumeActorId,
ui64 partitionTabletId,
ui32 blockSize,
const EStorageAccessMode storageAccessMode,
std::unique_ptr<TRequest> request,
TDuration longRunningThreshold)
: TLongRunningOperationCompanion(
tablet,
partitionActorId,
volumeActorId,
longRunningThreshold,
TLongRunningOperationCompanion::EOperation::ReadBlob,
request->GroupId)
, RequestInfo(std::move(requestInfo))
, Tablet(tablet)
, TabletId(tabletId)
, PartitionActorId(partitionActorId)
, PartitionTabletId(partitionTabletId)
, BlockSize(blockSize)
, StorageAccessMode(storageAccessMode)
, Request(std::move(request))
Expand Down Expand Up @@ -104,7 +106,7 @@ void TReadBlobActor::NotifyCompleted(
request->RequestTime = ResponseReceived - RequestSent;
request->GroupId = Request->GroupId;

NCloud::Send(ctx, Tablet, std::move(request));
NCloud::Send(ctx, PartitionActorId, std::move(request));
}

void TReadBlobActor::ReplyAndDie(
Expand All @@ -123,8 +125,13 @@ void TReadBlobActor::ReplyAndDie(
RequestInfo->CallContext->RequestId);
}

if (HasError(response->GetError())) {
TLongRunningOperationCompanion::RequestCancelled(ctx);
} else {
TLongRunningOperationCompanion::RequestFinished(ctx);
}

NCloud::Reply(ctx, *RequestInfo, std::move(response));
TLongRunningOperationCompanion::RequestFinished(ctx);
Die(ctx);
}

Expand All @@ -135,7 +142,7 @@ void TReadBlobActor::ReplyError(
{
LOG_ERROR(ctx, TBlockStoreComponents::PARTITION_COMMON,
"[%lu] TEvBlobStorage::TEvGet failed: %s\n%s",
TabletId,
PartitionTabletId,
description.data(),
response.Print(false).data());

Expand Down Expand Up @@ -176,7 +183,7 @@ void TReadBlobActor::HandleGetResult(
{
LOG_WARN(ctx, TBlockStoreComponents::PARTITION_COMMON,
"[%lu] Repairing TEvBlobStorage::TEvGet %s error (%s)",
TabletId,
PartitionTabletId,
NKikimrProto::EReplyStatus_Name(response.Status).data(),
msg->Print(false).data());

Expand Down
9 changes: 5 additions & 4 deletions cloud/blockstore/libs/storage/partition_common/actor_read_blob.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,8 @@ class TReadBlobActor final
private:
const TRequestInfoPtr RequestInfo;

const NActors::TActorId Tablet;
const ui64 TabletId;
const NActors::TActorId PartitionActorId;
const ui64 PartitionTabletId;
const ui32 BlockSize;
const EStorageAccessMode StorageAccessMode;
const std::unique_ptr<TRequest> Request;
Expand All @@ -37,8 +37,9 @@ class TReadBlobActor final
public:
TReadBlobActor(
TRequestInfoPtr requestInfo,
const NActors::TActorId& tablet,
ui64 tabletId,
const NActors::TActorId& partitionActorId,
const NActors::TActorId& volumeActorId,
ui64 partitionTabletId,
ui32 blockSize,
const EStorageAccessMode storageAccessMode,
std::unique_ptr<TRequest> request,
Expand Down
Loading

0 comments on commit c717a58

Please sign in to comment.