| // Copyright 2022 The Pigweed Authors |
| // |
| // Licensed under the Apache License, Version 2.0 (the "License"); you may not |
| // use this file except in compliance with the License. You may obtain a copy of |
| // the License at |
| // |
| // https://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| // WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| // License for the specific language governing permissions and limitations under |
| // the License. |
| |
| #define PW_LOG_MODULE_NAME "TRN" |
| |
| #include "pw_transfer/transfer_thread.h" |
| |
| #include "pw_assert/check.h" |
| #include "pw_log/log.h" |
| #include "pw_transfer/internal/chunk.h" |
| |
| PW_MODIFY_DIAGNOSTICS_PUSH(); |
| PW_MODIFY_DIAGNOSTIC(ignored, "-Wmissing-field-initializers"); |
| |
| namespace pw::transfer::internal { |
| |
| void TransferThread::Terminate() { |
| next_event_ownership_.acquire(); |
| next_event_.type = EventType::kTerminate; |
| event_notification_.release(); |
| } |
| |
| void TransferThread::SimulateTimeout(EventType type, uint32_t session_id) { |
| next_event_ownership_.acquire(); |
| |
| next_event_.type = type; |
| next_event_.chunk = {}; |
| next_event_.chunk.session_id = session_id; |
| |
| event_notification_.release(); |
| |
| WaitUntilEventIsProcessed(); |
| } |
| |
| void TransferThread::Run() { |
| // Next event starts freed. |
| next_event_ownership_.release(); |
| |
| while (true) { |
| if (event_notification_.try_acquire_until(GetNextTransferTimeout())) { |
| if (next_event_.type == EventType::kTerminate) { |
| return; |
| } |
| |
| HandleEvent(next_event_); |
| |
| // Finished processing the event. Allow the next_event struct to be |
| // overwritten. |
| next_event_ownership_.release(); |
| } |
| |
| // Regardless of whether an event was received or not, check for any |
| // transfers which have timed out and process them if so. |
| for (Context& context : client_transfers_) { |
| if (context.timed_out()) { |
| context.HandleEvent({.type = EventType::kClientTimeout}); |
| } |
| } |
| for (Context& context : server_transfers_) { |
| if (context.timed_out()) { |
| context.HandleEvent({.type = EventType::kServerTimeout}); |
| } |
| } |
| } |
| } |
| |
| chrono::SystemClock::time_point TransferThread::GetNextTransferTimeout() const { |
| chrono::SystemClock::time_point timeout = |
| chrono::SystemClock::TimePointAfterAtLeast(kMaxTimeout); |
| |
| for (Context& context : client_transfers_) { |
| auto ctx_timeout = context.timeout(); |
| if (ctx_timeout.has_value() && ctx_timeout.value() < timeout) { |
| timeout = ctx_timeout.value(); |
| } |
| } |
| for (Context& context : server_transfers_) { |
| auto ctx_timeout = context.timeout(); |
| if (ctx_timeout.has_value() && ctx_timeout.value() < timeout) { |
| timeout = ctx_timeout.value(); |
| } |
| } |
| |
| return timeout; |
| } |
| |
| void TransferThread::StartTransfer(TransferType type, |
| uint32_t session_id, |
| uint32_t resource_id, |
| stream::Stream* stream, |
| const TransferParameters& max_parameters, |
| Function<void(Status)>&& on_completion, |
| chrono::SystemClock::duration timeout, |
| uint8_t max_retries) { |
| // Block until the last event has been processed. |
| next_event_ownership_.acquire(); |
| |
| bool is_client_transfer = stream != nullptr; |
| |
| next_event_.type = is_client_transfer ? EventType::kNewClientTransfer |
| : EventType::kNewServerTransfer; |
| next_event_.new_transfer = { |
| .type = type, |
| .session_id = session_id, |
| .resource_id = resource_id, |
| .max_parameters = &max_parameters, |
| .timeout = timeout, |
| .max_retries = max_retries, |
| .transfer_thread = this, |
| }; |
| |
| staged_on_completion_ = std::move(on_completion); |
| |
| // The transfer is initialized with either a stream (client-side) or a handler |
| // (server-side). If no stream is provided, try to find a registered handler |
| // with the specified ID. |
| if (is_client_transfer) { |
| next_event_.new_transfer.stream = stream; |
| next_event_.new_transfer.rpc_writer = &static_cast<rpc::Writer&>( |
| type == TransferType::kTransmit ? client_write_stream_ |
| : client_read_stream_); |
| } else { |
| auto handler = std::find_if(handlers_.begin(), |
| handlers_.end(), |
| [&](auto& h) { return h.id() == resource_id; }); |
| if (handler != handlers_.end()) { |
| next_event_.new_transfer.handler = &*handler; |
| next_event_.new_transfer.rpc_writer = &static_cast<rpc::Writer&>( |
| type == TransferType::kTransmit ? server_read_stream_ |
| : server_write_stream_); |
| } else { |
| // No handler exists for the transfer: return a NOT_FOUND. |
| next_event_.type = EventType::kSendStatusChunk; |
| next_event_.send_status_chunk = { |
| .session_id = session_id, |
| .status = Status::NotFound().code(), |
| .stream = type == TransferType::kTransmit |
| ? TransferStream::kServerRead |
| : TransferStream::kServerWrite, |
| }; |
| } |
| } |
| |
| event_notification_.release(); |
| } |
| |
| void TransferThread::ProcessChunk(EventType type, ConstByteSpan chunk) { |
| // If this assert is hit, there is a bug in the transfer implementation. |
| // Contexts' max_chunk_size_bytes fields should be set based on the size of |
| // chunk_buffer_. |
| PW_CHECK(chunk.size() <= chunk_buffer_.size(), |
| "Transfer received a larger chunk than it can handle."); |
| |
| Result<uint32_t> session_id = Chunk::ExtractSessionId(chunk); |
| if (!session_id.ok()) { |
| PW_LOG_ERROR("Received a malformed chunk without a session ID"); |
| return; |
| } |
| |
| // Block until the last event has been processed. |
| next_event_ownership_.acquire(); |
| |
| std::memcpy(chunk_buffer_.data(), chunk.data(), chunk.size()); |
| |
| next_event_.type = type; |
| next_event_.chunk = { |
| .session_id = *session_id, |
| .data = chunk_buffer_.data(), |
| .size = chunk.size(), |
| }; |
| |
| event_notification_.release(); |
| } |
| |
| void TransferThread::EndTransfer(EventType type, |
| uint32_t session_id, |
| Status status, |
| bool send_status_chunk) { |
| // Block until the last event has been processed. |
| next_event_ownership_.acquire(); |
| |
| next_event_.type = type; |
| next_event_.end_transfer = { |
| .session_id = session_id, |
| .status = status.code(), |
| .send_status_chunk = send_status_chunk, |
| }; |
| |
| event_notification_.release(); |
| } |
| |
| void TransferThread::SetClientStream(TransferStream type, |
| rpc::RawClientReaderWriter& stream) { |
| // Block until the last event has been processed. |
| next_event_ownership_.acquire(); |
| |
| next_event_.type = EventType::kSetTransferStream; |
| next_event_.set_transfer_stream = type; |
| staged_client_stream_ = std::move(stream); |
| |
| event_notification_.release(); |
| } |
| |
| void TransferThread::SetServerStream(TransferStream type, |
| rpc::RawServerReaderWriter& stream) { |
| // Block until the last event has been processed. |
| next_event_ownership_.acquire(); |
| |
| next_event_.type = EventType::kSetTransferStream; |
| next_event_.set_transfer_stream = type; |
| staged_server_stream_ = std::move(stream); |
| |
| event_notification_.release(); |
| } |
| |
| void TransferThread::TransferHandlerEvent(EventType type, |
| internal::Handler& handler) { |
| // Block until the last event has been processed. |
| next_event_ownership_.acquire(); |
| |
| next_event_.type = type; |
| if (type == EventType::kAddTransferHandler) { |
| next_event_.add_transfer_handler = &handler; |
| } else { |
| next_event_.remove_transfer_handler = &handler; |
| } |
| |
| event_notification_.release(); |
| } |
| |
| void TransferThread::HandleEvent(const internal::Event& event) { |
| switch (event.type) { |
| case EventType::kSendStatusChunk: |
| SendStatusChunk(event.send_status_chunk); |
| break; |
| |
| case EventType::kSetTransferStream: |
| switch (event.set_transfer_stream) { |
| case TransferStream::kClientRead: |
| client_read_stream_ = std::move(staged_client_stream_); |
| break; |
| |
| case TransferStream::kClientWrite: |
| client_write_stream_ = std::move(staged_client_stream_); |
| break; |
| |
| case TransferStream::kServerRead: |
| server_read_stream_ = std::move(staged_server_stream_); |
| break; |
| |
| case TransferStream::kServerWrite: |
| server_write_stream_ = std::move(staged_server_stream_); |
| break; |
| } |
| return; |
| |
| case EventType::kAddTransferHandler: |
| handlers_.push_front(*event.add_transfer_handler); |
| return; |
| |
| case EventType::kRemoveTransferHandler: |
| for (ServerContext& server_context : server_transfers_) { |
| if (server_context.handler() == event.remove_transfer_handler) { |
| server_context.HandleEvent({ |
| .type = EventType::kServerEndTransfer, |
| .end_transfer = |
| { |
| .session_id = server_context.session_id(), |
| .status = Status::Aborted().code(), |
| .send_status_chunk = false, |
| }, |
| }); |
| } |
| } |
| handlers_.remove(*event.remove_transfer_handler); |
| return; |
| |
| default: |
| // Other events are handled by individual transfer contexts. |
| break; |
| } |
| |
| if (Context* ctx = FindContextForEvent(event); ctx != nullptr) { |
| if (event.type == EventType::kNewClientTransfer) { |
| // TODO(frolv): This is terrible. |
| static_cast<ClientContext*>(ctx)->set_on_completion( |
| std::move(staged_on_completion_)); |
| } |
| |
| ctx->HandleEvent(event); |
| } |
| } |
| |
| Context* TransferThread::FindContextForEvent( |
| const internal::Event& event) const { |
| switch (event.type) { |
| case EventType::kNewClientTransfer: |
| return FindNewTransfer(client_transfers_, event.new_transfer.session_id); |
| case EventType::kNewServerTransfer: |
| return FindNewTransfer(server_transfers_, event.new_transfer.session_id); |
| |
| case EventType::kClientChunk: |
| return FindActiveTransfer(client_transfers_, event.chunk.session_id); |
| case EventType::kServerChunk: |
| return FindActiveTransfer(server_transfers_, event.chunk.session_id); |
| |
| case EventType::kClientTimeout: // Manually triggered client timeout |
| return FindActiveTransfer(client_transfers_, event.chunk.session_id); |
| case EventType::kServerTimeout: // Manually triggered server timeout |
| return FindActiveTransfer(server_transfers_, event.chunk.session_id); |
| |
| case EventType::kClientEndTransfer: |
| return FindActiveTransfer(client_transfers_, |
| event.end_transfer.session_id); |
| case EventType::kServerEndTransfer: |
| return FindActiveTransfer(server_transfers_, |
| event.end_transfer.session_id); |
| |
| default: |
| return nullptr; |
| } |
| } |
| |
| void TransferThread::SendStatusChunk( |
| const internal::SendStatusChunkEvent& event) { |
| rpc::Writer& destination = stream_for(event.stream); |
| |
| Result<ConstByteSpan> result = |
| Chunk::Final(ProtocolVersion::kLegacy, event.session_id, event.status) |
| .Encode(chunk_buffer_); |
| |
| if (!result.ok()) { |
| PW_LOG_ERROR("Failed to encode final chunk for transfer %u", |
| static_cast<unsigned>(event.session_id)); |
| return; |
| } |
| |
| if (!destination.Write(result.value()).ok()) { |
| PW_LOG_ERROR("Failed to send final chunk for transfer %u", |
| static_cast<unsigned>(event.session_id)); |
| return; |
| } |
| } |
| |
| } // namespace pw::transfer::internal |
| |
| PW_MODIFY_DIAGNOSTICS_POP(); |