From babe14b2f204f5b2b4af56ed2e6ee4c9afd4fc3b Mon Sep 17 00:00:00 2001 From: hai007 Date: Tue, 16 Mar 2021 22:31:09 -0700 Subject: [PATCH] Internal change PiperOrigin-RevId: 363347152 --- cpp/core/internal/payload_manager.cc | 215 ++++++++++++++------------- cpp/core/internal/payload_manager.h | 9 +- 2 files changed, 122 insertions(+), 102 deletions(-) diff --git a/cpp/core/internal/payload_manager.cc b/cpp/core/internal/payload_manager.cc index 06fd25b8..d9a0ba6c 100644 --- a/cpp/core/internal/payload_manager.cc +++ b/cpp/core/internal/payload_manager.cc @@ -263,14 +263,16 @@ PayloadManager::~PayloadManager() { CountDownLatch stop_latch(1); // Clear our tracked pending payloads. - RunOnStatusUpdateThread([this, &stop_latch]() { - NEARBY_LOG(INFO, "PayloadManager: stop tracking payloads; self=%p", this); - MutexLock lock(&mutex_); - for (const auto& pending_id : pending_payloads_.GetAllPayloads()) { - pending_payloads_.StopTrackingPayload(pending_id); - } - stop_latch.CountDown(); - }); + RunOnStatusUpdateThread( + [this, &stop_latch]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() { + NEARBY_LOG(INFO, "PayloadManager: stop tracking payloads; self=%p", + this); + MutexLock lock(&mutex_); + for (const auto& pending_id : pending_payloads_.GetAllPayloads()) { + pending_payloads_.StopTrackingPayload(pending_id); + } + stop_latch.CountDown(); + }); stop_latch.Await(); NEARBY_LOG(INFO, "PayloadManager: turn down notification executor; self=%p", @@ -329,8 +331,10 @@ void PayloadManager::SendPayload(ClientProxy* client, should_continue = SendPayloadLoop(client, *pending_payload, payload_header, next_chunk_offset); } - RunOnStatusUpdateThread( - [this, payload_id]() { DestroyPendingPayload(payload_id); }); + RunOnStatusUpdateThread([this, payload_id]() + RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() { + DestroyPendingPayload(payload_id); + }); }); NEARBY_LOG(INFO, "PayloadManager: xfer scheduled: self=%p; id=%" PRIX64 ", type=%d", @@ -394,40 +398,42 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client, barrier.CountDown(); return; } - RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() mutable { - // Iterate through all our payloads and look for payloads associated - // with this endpoint. - MutexLock lock(&mutex_); - for (const auto& payload_id : pending_payloads_.GetAllPayloads()) { - auto* pending_payload = pending_payloads_.GetPayload(payload_id); - if (!pending_payload) continue; - auto endpoint_info = pending_payload->GetEndpoint(endpoint_id); - if (!endpoint_info) continue; - std::int64_t endpoint_offset = endpoint_info->offset; - // Stop tracking the endpoint for this payload. - pending_payload->RemoveEndpoints({endpoint_id}); - // |endpoint_info| is longer valid after calling RemoveEndpoints. - endpoint_info = nullptr; + RunOnStatusUpdateThread( + [this, client, endpoint_id, barrier]() + RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() mutable { + // Iterate through all our payloads and look for payloads associated + // with this endpoint. + MutexLock lock(&mutex_); + for (const auto& payload_id : pending_payloads_.GetAllPayloads()) { + auto* pending_payload = pending_payloads_.GetPayload(payload_id); + if (!pending_payload) continue; + auto endpoint_info = pending_payload->GetEndpoint(endpoint_id); + if (!endpoint_info) continue; + std::int64_t endpoint_offset = endpoint_info->offset; + // Stop tracking the endpoint for this payload. + pending_payload->RemoveEndpoints({endpoint_id}); + // |endpoint_info| is longer valid after calling RemoveEndpoints. + endpoint_info = nullptr; - std::int64_t payload_total_size = - pending_payload->GetInternalPayload()->GetTotalSize(); + std::int64_t payload_total_size = + pending_payload->GetInternalPayload()->GetTotalSize(); - // If no endpoints are left for this payload, close it. - if (pending_payload->GetEndpoints().empty()) { - pending_payload->Close(); - } + // If no endpoints are left for this payload, close it. + if (pending_payload->GetEndpoints().empty()) { + pending_payload->Close(); + } - // Create the payload transfer update. - PayloadProgressInfo update{payload_id, - PayloadProgressInfo::Status::kFailure, - payload_total_size, endpoint_offset}; + // Create the payload transfer update. + PayloadProgressInfo update{payload_id, + PayloadProgressInfo::Status::kFailure, + payload_total_size, endpoint_offset}; - // Send a client notification of a payload transfer failure. - client->OnPayloadProgress(endpoint_id, update); - } + // Send a client notification of a payload transfer failure. + client->OnPayloadProgress(endpoint_id, update); + } - barrier.CountDown(); - }); + barrier.CountDown(); + }); } proto::connections::PayloadStatus @@ -546,7 +552,8 @@ void PayloadManager::SendClientCallbacksForFinishedOutgoingPayload( std::int64_t num_bytes_successfully_transferred, proto::connections::PayloadStatus status) { RunOnStatusUpdateThread([this, client, finished_endpoint_ids, payload_header, - num_bytes_successfully_transferred, status]() { + num_bytes_successfully_transferred, + status]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() { // Make sure we're still tracking this payload. PendingPayload* pending_payload = GetPayload(payload_header.id()); if (!pending_payload) { @@ -583,7 +590,8 @@ void PayloadManager::SendClientCallbacksForFinishedIncomingPayload( const PayloadTransferFrame::PayloadHeader& payload_header, std::int64_t offset_bytes, proto::connections::PayloadStatus status) { RunOnStatusUpdateThread( - [this, client, endpoint_id, payload_header, offset_bytes, status]() { + [this, client, endpoint_id, payload_header, offset_bytes, + status]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() { // Make sure we're still tracking this payload. PendingPayload* pending_payload = GetPayload(payload_header.id()); if (!pending_payload) { @@ -684,42 +692,44 @@ void PayloadManager::HandleSuccessfulOutgoingChunk( const PayloadTransferFrame::PayloadHeader& payload_header, std::int32_t payload_chunk_flags, std::int64_t payload_chunk_offset, std::int64_t payload_chunk_body_size) { - RunOnStatusUpdateThread([this, client, endpoint_id, payload_header, - payload_chunk_flags, payload_chunk_offset, - payload_chunk_body_size]() { - // Make sure we're still tracking this payload and its associated - // endpoint. - PendingPayload* pending_payload = GetPayload(payload_header.id()); - if (!pending_payload || !pending_payload->GetEndpoint(endpoint_id)) { - NEARBY_LOG(INFO, - "HandleSuccessfulOutgoingChunk: endpoint not found: id=%s", - endpoint_id.c_str()); - return; - } + RunOnStatusUpdateThread( + [this, client, endpoint_id, payload_header, payload_chunk_flags, + payload_chunk_offset, + payload_chunk_body_size]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() { + // Make sure we're still tracking this payload and its associated + // endpoint. + PendingPayload* pending_payload = GetPayload(payload_header.id()); + if (!pending_payload || !pending_payload->GetEndpoint(endpoint_id)) { + NEARBY_LOG(INFO, + "HandleSuccessfulOutgoingChunk: endpoint not found: id=%s", + endpoint_id.c_str()); + return; + } - bool is_last_chunk = (payload_chunk_flags & - PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0; - PayloadProgressInfo update{ - payload_header.id(), - is_last_chunk ? PayloadProgressInfo::Status::kSuccess - : PayloadProgressInfo::Status::kInProgress, - payload_header.total_size(), - is_last_chunk ? payload_chunk_offset - : payload_chunk_offset + payload_chunk_body_size}; + bool is_last_chunk = + (payload_chunk_flags & + PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0; + PayloadProgressInfo update{ + payload_header.id(), + is_last_chunk ? PayloadProgressInfo::Status::kSuccess + : PayloadProgressInfo::Status::kInProgress, + payload_header.total_size(), + is_last_chunk ? payload_chunk_offset + : payload_chunk_offset + payload_chunk_body_size}; - // Notify the client. - client->OnPayloadProgress(endpoint_id, update); + // Notify the client. + client->OnPayloadProgress(endpoint_id, update); - if (is_last_chunk) { - // Stop tracking this endpoint. - pending_payload->RemoveEndpoints({endpoint_id}); + if (is_last_chunk) { + // Stop tracking this endpoint. + pending_payload->RemoveEndpoints({endpoint_id}); - // Close the payload if no endpoints remain. - if (pending_payload->GetEndpoints().empty()) { - pending_payload->Close(); - } - } - }); + // Close the payload if no endpoints remain. + if (pending_payload->GetEndpoints().empty()) { + pending_payload->Close(); + } + } + }); } // @PayloadManagerStatusUpdateThread @@ -746,28 +756,30 @@ void PayloadManager::HandleSuccessfulIncomingChunk( const PayloadTransferFrame::PayloadHeader& payload_header, std::int32_t payload_chunk_flags, std::int64_t payload_chunk_offset, std::int64_t payload_chunk_body_size) { - RunOnStatusUpdateThread([this, client, endpoint_id, payload_header, - payload_chunk_flags, payload_chunk_offset, - payload_chunk_body_size]() { - // Make sure we're still tracking this payload. - PendingPayload* pending_payload = GetPayload(payload_header.id()); - if (!pending_payload) { - return; - } + RunOnStatusUpdateThread( + [this, client, endpoint_id, payload_header, payload_chunk_flags, + payload_chunk_offset, + payload_chunk_body_size]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() { + // Make sure we're still tracking this payload. + PendingPayload* pending_payload = GetPayload(payload_header.id()); + if (!pending_payload) { + return; + } - bool is_last_chunk = (payload_chunk_flags & - PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0; - PayloadProgressInfo update{ - payload_header.id(), - is_last_chunk ? PayloadProgressInfo::Status::kSuccess - : PayloadProgressInfo::Status::kInProgress, - payload_header.total_size(), - is_last_chunk ? payload_chunk_offset - : payload_chunk_offset + payload_chunk_body_size}; + bool is_last_chunk = + (payload_chunk_flags & + PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0; + PayloadProgressInfo update{ + payload_header.id(), + is_last_chunk ? PayloadProgressInfo::Status::kSuccess + : PayloadProgressInfo::Status::kInProgress, + payload_header.total_size(), + is_last_chunk ? payload_chunk_offset + : payload_chunk_offset + payload_chunk_body_size}; - // Notify the client of this update. - NotifyClientOfIncomingPayloadProgressInfo(client, endpoint_id, update); - }); + // Notify the client of this update. + NotifyClientOfIncomingPayloadProgressInfo(client, endpoint_id, update); + }); } // @EndpointManagerDataPool @@ -792,13 +804,16 @@ void PayloadManager::ProcessDataPacket( } // Also, let the client know of this new incoming payload. - RunOnStatusUpdateThread([to_client, from_endpoint_id, pending_payload]() { - NEARBY_LOG(INFO, "ProcessDataPacket [new]: id=%s; payload_id=%" PRIX64, - from_endpoint_id.c_str(), pending_payload->GetId()); - to_client->OnPayload( - from_endpoint_id, - pending_payload->GetInternalPayload()->ReleasePayload()); - }); + RunOnStatusUpdateThread( + [to_client, from_endpoint_id, pending_payload]() + RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() { + NEARBY_LOG(INFO, + "ProcessDataPacket [new]: id=%s; payload_id=%" PRIX64, + from_endpoint_id.c_str(), pending_payload->GetId()); + to_client->OnPayload( + from_endpoint_id, + pending_payload->GetInternalPayload()->ReleasePayload()); + }); } else { pending_payload = GetPayload(payload_header.id()); if (!pending_payload) { diff --git a/cpp/core/internal/payload_manager.h b/cpp/core/internal/payload_manager.h index bf2dffaa..9c187565 100644 --- a/cpp/core/internal/payload_manager.h +++ b/cpp/core/internal/payload_manager.h @@ -39,6 +39,11 @@ namespace location { namespace nearby { namespace connections { +// Annotations for methods that need to run on PayloadStatusUpdateThread. +// Use only in PayloadManager +#define RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() \ + ABSL_EXCLUSIVE_LOCKS_REQUIRED(payload_status_update_executor_) + class PayloadManager : public EndpointManager::FrameProcessor { public: using EndpointIds = std::vector; @@ -265,10 +270,10 @@ class PayloadManager : public EndpointManager::FrameProcessor { const std::string& from_endpoint_id, PayloadTransferFrame& payload_transfer_frame); - // @PayloadStatusUpdateThread void NotifyClientOfIncomingPayloadProgressInfo( ClientProxy* client, const std::string& endpoint_id, - const PayloadProgressInfo& payload_transfer_update); + const PayloadProgressInfo& payload_transfer_update) + RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD(); SingleThreadExecutor* GetOutgoingPayloadExecutor(Payload::Type payload_type);