Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

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

Merged
merged 1 commit into from
Jan 22, 2024
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cloud/blockstore/libs/storage/partition/part_actor_readblob.cpp
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Loading