| // Copyright 2018 The Fuchsia Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| #include "src/connectivity/overnet/lib/packet_protocol/packet_protocol.h" |
| |
| #include <iostream> |
| #include <sstream> |
| |
| namespace overnet { |
| |
| namespace { |
| |
| class AckSendRequest final : public PacketProtocol::SendRequest { |
| public: |
| AckSendRequest() : SendRequest(true) {} |
| Slice GenerateBytes(LazySliceArgs args) override { return Slice(); } |
| void Ack(const Status& status) override {} |
| }; |
| |
| AckSendRequest ack_send_request; |
| |
| } // namespace |
| |
| /////////////////////////////////////////////////////////////////////////////// |
| // Initialization/shutdown/control paths |
| |
| PacketProtocol::PacketProtocol(Timer* timer, RandFunc rand, |
| PacketSender* packet_sender, const Codec* codec, |
| uint64_t mss, bool probe_tails) |
| : codec_(codec), |
| timer_(timer), |
| packet_sender_(packet_sender), |
| probe_tails_(probe_tails), |
| maximum_send_size_(mss) { |
| state_.Reset(this, std::move(rand)); |
| } |
| |
| void PacketProtocol::Close(Callback<void> quiesced) { |
| ScopedModule<PacketProtocol> in_pp(this); |
| OVERNET_TRACE(DEBUG) << "Close"; |
| InTransaction([&](Transaction* transaction) { |
| transaction->QuiesceOnCompletion(std::move(quiesced)); |
| }); |
| } |
| |
| PacketProtocol::AckSender::AckSender(PacketProtocol* protocol) |
| : protocol_(protocol) {} |
| |
| PacketProtocol::OutstandingMessages::OutstandingMessages( |
| PacketProtocol* protocol) |
| : protocol_(protocol) {} |
| |
| void PacketProtocol::Quiesce() { |
| ScopedModule<PacketProtocol> in_pp(this); |
| OVERNET_TRACE(DEBUG) << "Quiesce"; |
| state_.Reset(); |
| primary_ref_.Drop(); |
| } |
| |
| PacketProtocol::Transaction::Transaction(PacketProtocol* protocol) |
| : protocol_(protocol) { |
| ScopedModule<PacketProtocol> in_pp(protocol_); |
| OVERNET_TRACE(DEBUG) << "Transaction.Begin"; |
| assert(protocol_->active_transaction_ == nullptr); |
| protocol_->active_transaction_ = this; |
| } |
| |
| PacketProtocol::Transaction::~Transaction() { |
| OVERNET_TRACE(DEBUG) << "Transaction.Finalize"; |
| ProtocolRef protocol(protocol_); |
| for (;;) { |
| if (quiesce_) { |
| protocol->Quiesce(); |
| } else if (protocol->state_.has_value()) { |
| if (bbr_ack_.has_value()) { |
| protocol->state_->bbr.OnAck(bbr_ack_.Take()); |
| continue; |
| } else if (start_sending_) { |
| start_sending_ = false; |
| protocol->state_->outstanding_messages.StartSending(); |
| continue; |
| } else if (increment_outstanding_tip_) { |
| increment_outstanding_tip_ = false; |
| protocol->state_->outstanding_messages.IncrementTip(); |
| continue; |
| } |
| } |
| |
| OVERNET_TRACE(DEBUG) << "Transaction.End"; |
| assert(protocol->active_transaction_ == this); |
| protocol->active_transaction_ = nullptr; |
| return; |
| } |
| } |
| |
| void PacketProtocol::Transaction::QuiesceOnCompletion(Callback<void> callback) { |
| OVERNET_TRACE(DEBUG) << "Schedule Quiesce"; |
| assert(!quiesce_); |
| assert(protocol_->quiesce_.empty()); |
| protocol_->quiesce_ = std::move(callback); |
| quiesce_ = true; |
| } |
| |
| /////////////////////////////////////////////////////////////////////////////// |
| // PacketProtocol::Send and friends. |
| // Defines the send path. |
| |
| void PacketProtocol::Send(SendRequestHdl send_request) { |
| ScopedModule<PacketProtocol> in_pp(this); |
| OVERNET_TRACE(DEBUG) << "Send"; |
| InTransaction([&](Transaction* transaction) { |
| if (!state_.has_value()) { |
| return; |
| } |
| state_->outstanding_messages.Schedule(transaction, std::move(send_request)); |
| }); |
| } |
| |
| void PacketProtocol::OutstandingMessages::Schedule(Transaction* transaction, |
| SendRequestHdl hdl) { |
| Schedule(transaction, OutstandingPacket::Pending{std::move(hdl)}); |
| } |
| |
| std::string PacketProtocol::OutstandingMessages::OutstandingString() const { |
| std::ostringstream out; |
| out << "{tip=" << send_tip_ << ":" << unsent_tip_ << "|"; |
| bool first = true; |
| for (const auto& pkt : outstanding_) { |
| if (!first) { |
| out << ","; |
| } |
| first = false; |
| out << pkt.state; |
| } |
| out << "}"; |
| return out.str(); |
| } |
| |
| void PacketProtocol::OutstandingMessages::Schedule( |
| Transaction* transaction, |
| OutstandingPacket::State outstanding_packet_state) { |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.Schedule " |
| << outstanding_packet_state |
| << " outstanding=" << OutstandingString(); |
| tail_probe_timeout_.Reset(); |
| auto make_packet = [&] { |
| return OutstandingPacket{ |
| protocol_->timer_->Now(), |
| protocol_->state_->received_queue.max_seen_sequence(), |
| std::move(outstanding_packet_state)}; |
| }; |
| if (!outstanding_.empty() && |
| std::holds_alternative<OutstandingPacket::PendingTailProbe>( |
| outstanding_.back().state)) { |
| if (!std::holds_alternative<OutstandingPacket::PendingTailProbe>( |
| outstanding_packet_state)) { |
| outstanding_.back() = make_packet(); |
| } |
| } else { |
| if (unsent_tip_ - send_tip_ == outstanding_.size()) { |
| transaction->StartSendingOnCompletion(); |
| } |
| outstanding_.emplace_back(make_packet()); |
| } |
| ScheduleRetransmit(); |
| max_outstanding_size_ = |
| std::max(outstanding_.size(), size_t(max_outstanding_size_)); |
| } |
| |
| void PacketProtocol::OutstandingMessages::StartSending() { |
| ScopedModule<PacketProtocol> in_pp(protocol_); |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.StartSending: " |
| << OutstandingString(); |
| |
| assert(!outstanding_.empty()); |
| assert(unsent_tip_ >= send_tip_); |
| assert(unsent_tip_ - send_tip_ < outstanding_.size()); |
| |
| transmit_request_.Reset( |
| &protocol_->state_->bbr, [this](const Status& status) { |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.Send status=" << status; |
| if (status.is_error()) { |
| return; |
| } |
| SeqNum seq_num(unsent_tip_, max_outstanding_size_ + 1); |
| protocol_->packet_sender_->SendPacket( |
| seq_num, PacketSend(protocol_, transmit_request_.Take())); |
| }); |
| } |
| |
| PacketProtocol::PacketSend::PacketSend(PacketProtocol* protocol, |
| BBR::TransmitRequest bbr_request) |
| : protocol_(protocol), bbr_request_(std::move(bbr_request)) { |
| assert(protocol); |
| } |
| |
| Slice PacketProtocol::PacketSend::operator()(LazySliceArgs args) { |
| auto protocol = std::move(protocol_); |
| protocol->stats_.outgoing_packet_count++; |
| return protocol->InTransaction( |
| [=, protocol = protocol.get()](Transaction* transaction) { |
| ScopedModule<PacketProtocol> in_pp(protocol); |
| Slice output; |
| if (protocol->state_.has_value()) { |
| output = protocol->state_->outstanding_messages.GeneratePacket( |
| transaction, std::move(bbr_request_), args); |
| } |
| if (protocol->state_.has_value()) { |
| protocol->state_->outstanding_messages.SentMessage(transaction); |
| } |
| return output; |
| }); |
| } |
| |
| PacketProtocol::PacketSend::~PacketSend() { |
| if (protocol_.has_value() && protocol_->state_.has_value()) { |
| protocol_->InTransaction([this](Transaction* transaction) { |
| protocol_->state_->outstanding_messages.CancelledMessage(transaction); |
| }); |
| } |
| } |
| |
| void PacketProtocol::OutstandingMessages::CancelledMessage( |
| Transaction* transaction) { |
| SentMessage(transaction); |
| } |
| |
| Slice PacketProtocol::OutstandingMessages::GeneratePacket( |
| Transaction* transaction, BBR::TransmitRequest bbr_request, |
| LazySliceArgs args) { |
| if (unsent_tip_ - send_tip_ == outstanding_.size()) { |
| // nack before send? |
| OVERNET_TRACE(DEBUG) << "Seq " << unsent_tip_ << " nacked before sending"; |
| return Slice(); |
| } |
| |
| const uint64_t seq_idx = unsent_tip_; |
| auto* outstanding_packet = &outstanding_[seq_idx - send_tip_]; |
| |
| OVERNET_TRACE(DEBUG) << "GeneratePacket: " << outstanding_packet->state; |
| |
| SendRequestHdl request; |
| Slice send; |
| |
| if (std::holds_alternative<OutstandingPacket::Pending>( |
| outstanding_packet->state)) { |
| auto& pending_packet = |
| std::get<OutstandingPacket::Pending>(outstanding_packet->state); |
| request = std::move(pending_packet.request); |
| last_send_was_tail_probe_ = false; |
| } else if (std::holds_alternative<OutstandingPacket::PendingTailProbe>( |
| outstanding_packet->state)) { |
| request = SendRequestHdl(&ack_send_request); |
| last_send_was_tail_probe_ = true; |
| } else { |
| abort(); |
| } |
| |
| auto send_status = protocol_->codec_->Encode( |
| seq_idx, |
| protocol_->FormatPacket(transaction, seq_idx, request.borrow(), args)); |
| if (send_status.is_error()) { |
| OVERNET_TRACE(ERROR) << "Failed to encode packet: " |
| << send_status.AsStatus(); |
| } else { |
| send = std::move(*send_status); |
| } |
| |
| // outstanding_ should not have changed |
| assert(unsent_tip_ == seq_idx); |
| assert(outstanding_packet == &outstanding_[seq_idx - send_tip_]); |
| outstanding_packet->state = OutstandingPacket::Sent{ |
| std::move(request), |
| bbr_request.Sent(BBR::OutgoingPacket{seq_idx, send.length()}), |
| protocol_->state_->bdp_estimator.SentPacket(seq_idx)}; |
| unsent_tip_++; |
| ScheduleRetransmit(); |
| return send; |
| } |
| |
| Slice PacketProtocol::FormatPacket(Transaction* transaction, uint64_t seq_idx, |
| SendRequest* request, LazySliceArgs args) { |
| assert(state_.has_value()); |
| |
| const auto max_length = std::min(static_cast<uint64_t>(args.max_length), |
| static_cast<uint64_t>(maximum_send_size_)); |
| auto make_args = [=](uint64_t prefix_length, bool has_other_content) { |
| auto maxlen = max_length; |
| auto subtract_from_max = |
| std::min(maxlen, prefix_length + codec_->border.Total()); |
| maxlen -= subtract_from_max; |
| return LazySliceArgs{args.desired_border.WithAddedPrefix( |
| prefix_length + codec_->border.prefix), |
| maxlen, has_other_content || args.has_other_content}; |
| }; |
| |
| OVERNET_TRACE(DEBUG) << "FormatPacket: can_build_ack=" |
| << state_->received_queue.CanBuildAck() |
| << " request.must_send_ack=" << request->must_send_ack() |
| << " should_send_ack=" |
| << state_->ack_sender.ShouldSendAck(); |
| |
| if (state_->received_queue.CanBuildAck() && |
| (request->must_send_ack() || state_->ack_sender.ShouldSendAck())) { |
| stats_.acks_sent++; |
| auto ack = state_->received_queue.BuildAck( |
| transaction, seq_idx, timer_->Now(), |
| varint::MaximumLengthWithPrefix(max_length), &state_->ack_sender); |
| AckFrame::Writer ack_writer(&ack); |
| const uint8_t ack_length_length = |
| varint::WireSizeFor(ack_writer.wire_length()); |
| const uint64_t prefix_length = ack_length_length + ack_writer.wire_length(); |
| auto payload = request->GenerateBytes(make_args(prefix_length, true)); |
| stats_.pure_acks_sent += payload.length() == 0; |
| return std::move(payload).WithPrefix( |
| prefix_length, [&ack_writer, ack_length_length](uint8_t* p) { |
| ack_writer.Write( |
| varint::Write(ack_writer.wire_length(), ack_length_length, p)); |
| }); |
| } else { |
| return request->GenerateBytes(make_args(1, false)) |
| .WithPrefix(1, [](uint8_t* p) { *p = 0; }); |
| } |
| } |
| |
| uint64_t PacketProtocol::ReceivedQueue::max_seen_sequence() const { |
| if (received_packets_.empty()) { |
| return received_tip_; |
| } |
| size_t idx = received_packets_.size() - 1; |
| while (idx > 0 && received_packets_[idx].state == ReceiveState::UNKNOWN) { |
| idx--; |
| } |
| return received_tip_ + idx; |
| } |
| |
| AckFrame PacketProtocol::ReceivedQueue::BuildAck(Transaction* transaction, |
| uint64_t seq_idx, |
| TimeStamp now, |
| uint32_t max_length, |
| AckSender* ack_sender) { |
| OVERNET_TRACE(DEBUG) << "BuildAck seq:" << seq_idx |
| << " from received packets: " << ReceivedPacketSummary(); |
| |
| auto packet_delay = [this, now](uint64_t seq_idx) -> uint64_t { |
| const auto& received_packet = received_packets_[seq_idx - received_tip_]; |
| assert(received_packet.state != ReceiveState::UNKNOWN); |
| return (now - received_packet.when).as_us(); |
| }; |
| |
| const auto max_seen = max_seen_sequence(); |
| assert(max_seen > 0); |
| |
| AckFrame ack(max_seen, packet_delay(max_seen)); |
| for (uint64_t seq_idx = max_seen; seq_idx > received_tip_; seq_idx--) { |
| auto& received_packet = received_packets_[seq_idx - received_tip_]; |
| switch (received_packet.state) { |
| case ReceiveState::UNKNOWN: |
| OVERNET_TRACE(DEBUG) |
| << "Mark unseen packet " << seq_idx << " as NOT_RECEIVED"; |
| received_packet = ReceivedPacket{ReceiveState::NOT_RECEIVED, now}; |
| stats_->unseen_packets_marked_not_received++; |
| [[fallthrough]]; |
| case ReceiveState::NOT_RECEIVED: |
| ack.AddNack(seq_idx); |
| break; |
| case ReceiveState::RECEIVED: |
| case ReceiveState::RECEIVED_PURE_ACK: |
| case ReceiveState::RECEIVED_AND_ACKED_IMMEDIATELY: |
| break; |
| } |
| } |
| |
| ack.AdjustForMSS(max_length, packet_delay); |
| |
| ack_sender->AckSent(transaction, seq_idx, ack.partial()); |
| |
| OVERNET_TRACE(DEBUG) << "BuildAck generates: " << ack << " bytes=" |
| << Slice::FromWriters(AckFrame::Writer(&ack)); |
| |
| return ack; |
| } |
| |
| void PacketProtocol::OutstandingMessages::SentMessage( |
| Transaction* transaction) { |
| ScopedModule<PacketProtocol> in_pp(protocol_); |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.SentMessage: " |
| << OutstandingString(); |
| if (unsent_tip_ - send_tip_ != outstanding_.size()) { |
| transaction->StartSendingOnCompletion(); |
| } |
| if (protocol_->probe_tails_ && !last_send_was_tail_probe_) { |
| ScheduleAck(); |
| } |
| } |
| |
| void PacketProtocol::OutstandingMessages::ScheduleAck() { |
| ScopedModule<PacketProtocol> in_pp(protocol_); |
| const auto when = protocol_->timer_->Now() + protocol_->TailLossProbeDelay(); |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.ScheduleTailLossProbe"; |
| if (unsent_tip_ - send_tip_ != outstanding_.size()) { |
| protocol_->stats_ |
| .tail_loss_probes_cancelled_because_requests_already_queued++; |
| return; |
| } |
| if (tail_probe_timeout_.has_value() && tail_probe_timeout_->when <= when) { |
| protocol_->stats_ |
| .tail_loss_probes_cancelled_because_probe_already_scheduled++; |
| return; |
| } |
| tail_probe_timeout_.Reset( |
| protocol_->timer_, when, [this](const Status& status) { |
| ScopedModule<PacketProtocol> in_pp(protocol_); |
| OVERNET_TRACE(DEBUG) |
| << "OutstandingMessages.ScheduleTailLossProbe --> " << status; |
| if (status.is_error()) { |
| protocol_->stats_.tail_loss_probes_cancelled_after_timer_created++; |
| return; |
| } |
| tail_probe_timeout_.Reset(); |
| protocol_->stats_.tail_loss_probes_scheduled++; |
| protocol_->InTransaction( |
| [this](Transaction* transaction) { ForceSendAck(transaction); }); |
| }); |
| } |
| |
| void PacketProtocol::OutstandingMessages::ForceSendAck( |
| Transaction* transaction) { |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.ForceSendAck"; |
| Schedule(transaction, OutstandingPacket::PendingTailProbe{}); |
| } |
| |
| /////////////////////////////////////////////////////////////////////////////// |
| // PacketProtocol::Process and friends. |
| // Defines the receive path. |
| |
| void PacketProtocol::Process(TimeStamp received, SeqNum seq_num, Slice slice, |
| ProcessCallback handle_message) { |
| assert(refs_); |
| ScopedModule<PacketProtocol> scoped_module(this); |
| Transaction transaction(this); |
| |
| OVERNET_TRACE(DEBUG) << "Process: " << slice; |
| |
| stats_.incoming_packet_count++; |
| |
| if (!state_.has_value()) { |
| return; |
| } |
| |
| state_->bdp_estimator.ReceivedBytes(slice.length()); |
| state_->ack_sender.NeedAck( |
| &transaction, |
| state_->received_queue.Received(seq_num, received, [&](uint64_t seq_idx) { |
| return ProcessMessage(&transaction, seq_idx, std::move(slice), received, |
| std::move(handle_message)); |
| })); |
| state_->outstanding_messages.ReceivedPacket(); |
| } |
| |
| template <class F> |
| PacketProtocol::AckUrgency PacketProtocol::ReceivedQueue::Received( |
| SeqNum seq_num, TimeStamp received, F logic) { |
| const auto seq_idx = |
| seq_num.Reconstruct(received_tip_ + received_packets_.size()); |
| |
| OVERNET_TRACE(DEBUG) << "Process seq:" << seq_idx; |
| |
| if (seq_idx < received_tip_) { |
| stats_->ack_not_required_historic_sequence++; |
| return AckUrgency::NOT_REQUIRED; |
| } |
| |
| if (!EnsureValidReceivedPacket(seq_idx, received)) { |
| return AckUrgency::NOT_REQUIRED; |
| } |
| |
| auto* received_packet = &received_packets_[seq_idx - received_tip_]; |
| if (received_packet->state != ReceiveState::UNKNOWN) { |
| OVERNET_TRACE(DEBUG) << "frozen as " << received_packet->state; |
| stats_->ack_not_required_frozen_sequence++; |
| return AckUrgency::NOT_REQUIRED; |
| } |
| |
| auto pmr = logic(seq_idx); |
| OVERNET_TRACE(DEBUG) << "Process seq:" << seq_idx |
| << " process_message_result=" << pmr << " is_last=" |
| << (seq_idx == |
| received_tip_ + received_packets_.size() - 1) |
| << " optional_ack_run_length=" |
| << optional_ack_run_length_ |
| << " received_packets=" << ReceivedPacketSummary(); |
| |
| // received_packet may not be valid anymore, but the packet *must* exist |
| assert(seq_idx >= received_tip_); |
| received_packet = &received_packets_[seq_idx - received_tip_]; |
| |
| switch (pmr) { |
| case ProcessMessageResult::NOT_PROCESSED: |
| // Failed processing packets shouldn't generate traffic. |
| stats_->ack_not_required_invalid_packet++; |
| return AckUrgency::NOT_REQUIRED; |
| case ProcessMessageResult::NACK: |
| optional_ack_run_length_ = 0; |
| *received_packet = ReceivedPacket{ReceiveState::NOT_RECEIVED, received}; |
| stats_->ack_required_immediately_due_to_nack++; |
| // Always send a nack as soon as we realize one is necessary. |
| return AckUrgency::SEND_IMMEDIATELY; |
| case ProcessMessageResult::OPTIONAL_ACK: |
| // If we get a single ack without a payload, we suppress sending a reply |
| // ack. |
| optional_ack_run_length_++; |
| *received_packet = |
| ReceivedPacket{ReceiveState::RECEIVED_PURE_ACK, received}; |
| if (optional_ack_run_length_ < 5) { |
| stats_->ack_not_required_short_optional_run++; |
| return AckUrgency::NOT_REQUIRED; |
| } |
| stats_->ack_required_soon_ack_received++; |
| optional_ack_run_length_ = 0; |
| return AckUrgency::SEND_SOON; |
| case ProcessMessageResult::ACK: { |
| optional_ack_run_length_ = 0; |
| *received_packet = ReceivedPacket{ReceiveState::RECEIVED, received}; |
| static const uint64_t kMaxIncomingBeforeForcedAck = 3; |
| if (seq_idx - received_tip_ >= kMaxIncomingBeforeForcedAck) { |
| for (auto idx = seq_idx - received_tip_ - kMaxIncomingBeforeForcedAck; |
| idx < seq_idx - received_tip_; idx++) { |
| if (received_packets_[idx].state != ReceiveState::RECEIVED) { |
| goto dont_send_immediately; |
| } |
| } |
| *received_packet = ReceivedPacket{ |
| ReceiveState::RECEIVED_AND_ACKED_IMMEDIATELY, received}; |
| stats_->ack_required_immediately_due_to_multiple_receives++; |
| return AckUrgency::SEND_IMMEDIATELY; |
| } |
| dont_send_immediately: |
| // Got some data, make sure there's an ack scheduled soon. |
| stats_->ack_required_soon_data_received++; |
| return AckUrgency::SEND_SOON; |
| } |
| case ProcessMessageResult::ACK_URGENTLY: { |
| optional_ack_run_length_ = 0; |
| *received_packet = ReceivedPacket{ReceiveState::RECEIVED, received}; |
| stats_->ack_required_immediately_due_to_partial_ack++; |
| return AckUrgency::SEND_IMMEDIATELY; |
| } |
| } |
| } |
| |
| std::string PacketProtocol::ReceivedQueue::ReceivedPacketSummary() const { |
| std::ostringstream out; |
| out << "{"; |
| uint64_t seq_idx = received_tip_; |
| for (const auto& pkt : received_packets_) { |
| if (seq_idx != received_tip_) { |
| out << ", "; |
| } |
| out << "[" << seq_idx << "] " << pkt.state; |
| seq_idx++; |
| } |
| out << "}"; |
| return out.str(); |
| } |
| |
| PacketProtocol::ProcessMessageResult PacketProtocol::ProcessMessage( |
| Transaction* transaction, uint64_t seq_idx, Slice slice, TimeStamp received, |
| ProcessCallback handle_message) { |
| using StatusType = StatusOr<IncomingMessage*>; |
| |
| // Decode slice |
| auto slice_status = codec_->Decode(seq_idx, std::move(slice)); |
| if (slice_status.is_error()) { |
| handle_message(slice_status.AsStatus()); |
| return ProcessMessageResult::NOT_PROCESSED; |
| } |
| slice = std::move(*slice_status); |
| |
| const uint8_t* p = slice.begin(); |
| const uint8_t* end = slice.end(); |
| |
| if (p == end) { |
| // Empty packet |
| handle_message(nullptr); |
| return ProcessMessageResult::OPTIONAL_ACK; |
| } |
| |
| uint64_t ack_length; |
| if (!varint::Read(&p, end, &ack_length)) { |
| handle_message(StatusType(StatusCode::INVALID_ARGUMENT, |
| "Failed to parse ack length from message")); |
| return ProcessMessageResult::NOT_PROCESSED; |
| } |
| slice.TrimBegin(p - slice.begin()); |
| |
| OVERNET_TRACE(DEBUG) << "ack_length=" << ack_length; |
| |
| if (ack_length > slice.length()) { |
| handle_message(StatusType(StatusCode::INVALID_ARGUMENT, |
| "Ack frame claimed to be past end of message")); |
| return ProcessMessageResult::NOT_PROCESSED; |
| } |
| |
| Optional<AckFrame> ack; |
| |
| ProcessMessageResult ack_result = ProcessMessageResult::OPTIONAL_ACK; |
| if (ack_length > 0) { |
| auto ack_status = AckFrame::Parse(slice.TakeUntilOffset(ack_length)); |
| if (ack_status.is_error()) { |
| handle_message(ack_status.AsStatus()); |
| return ProcessMessageResult::NOT_PROCESSED; |
| } |
| |
| if (auto ack_valid = state_->outstanding_messages.ValidateAck(*ack_status); |
| ack_valid.is_error()) { |
| handle_message(ack_valid); |
| return ProcessMessageResult::NACK; |
| } |
| |
| OVERNET_TRACE(DEBUG) << "Process seq:" << seq_idx |
| << " got-ack:" << *ack_status; |
| ack = std::move(*ack_status); |
| if (ack->partial()) { |
| ack_result = ProcessMessageResult::ACK_URGENTLY; |
| } |
| } |
| |
| if (slice.length() > 0) { |
| IncomingMessage msg(std::move(slice)); |
| |
| // Process the message body: |
| handle_message(&msg); |
| |
| if (msg.was_nacked()) { |
| // Note: ack not processed |
| return ProcessMessageResult::NACK; |
| } else if (ack_result != ProcessMessageResult::ACK_URGENTLY) { |
| ack_result = ProcessMessageResult::ACK; |
| } |
| } else { |
| // Handle no message: |
| handle_message(nullptr); |
| } |
| if (ack.has_value()) { |
| state_->outstanding_messages.ProcessValidAck(transaction, ack.Take(), |
| received); |
| } |
| return ack_result; |
| } |
| |
| Status PacketProtocol::OutstandingMessages::ValidateAck( |
| const AckFrame& ack) const { |
| if (ack.ack_to_seq() < send_tip_) { |
| return Status::Ok(); |
| } |
| if (ack.ack_to_seq() >= send_tip_ + outstanding_.size()) { |
| std::ostringstream msg; |
| msg << "Ack packet past sending sequence: ack_seq=" << ack.ack_to_seq() |
| << " max_sent=" << (send_tip_ + outstanding_.size() - 1) |
| << " outstanding_window=" << outstanding_.size(); |
| return Status(StatusCode::INVALID_ARGUMENT, msg.str()); |
| } |
| for (auto nack_seq : ack.nack_seqs()) { |
| if (nack_seq < send_tip_) { |
| continue; |
| } |
| const OutstandingPacket& pkt = outstanding_[nack_seq - send_tip_]; |
| if (std::holds_alternative<OutstandingPacket::Acked>(pkt.state)) { |
| // Previously acked packet becomes nacked: this is an error. |
| return Status(StatusCode::INVALID_ARGUMENT, |
| "Previously acked packet becomes nacked"); |
| } |
| } |
| for (size_t i = 0; i < ack.ack_to_seq() - send_tip_; i++) { |
| if (std::holds_alternative<OutstandingPacket::Pending>( |
| outstanding_[i].state)) { |
| return Status(StatusCode::INVALID_ARGUMENT, "Ack/nack unsent sequence"); |
| } |
| } |
| return Status::Ok(); |
| } |
| |
| void PacketProtocol::OutstandingMessages::ProcessValidAck( |
| Transaction* transaction, AckFrame ack, TimeStamp received) { |
| // Basic validation. Can assert anything that should be an error because |
| // ValidateAck should have been called prior. |
| if (ack.ack_to_seq() < send_tip_) { |
| return; |
| } |
| assert(ack.ack_to_seq() < send_tip_ + outstanding_.size()); |
| |
| // Move receive window forward. |
| const auto max_seen_sequence_at_send = |
| outstanding_[ack.ack_to_seq() - send_tip_].max_seen_sequence_at_send; |
| if (max_seen_sequence_at_send > 0) { |
| protocol_->state_->received_queue.SetTip(max_seen_sequence_at_send, |
| received); |
| } |
| |
| const auto queue_delay = TimeDelta::FromMicroseconds(ack.ack_delay_us()); |
| |
| // Fail any nacked packets. |
| // Iteration is from oldest packet to newest, such that the OLDEST nacked |
| // message is the most likely to be sent first. This has the important |
| // consequence that if the packet was a fragment of a large message that was |
| // rejected due to buffering, the earlier pieces (that are more likely to |
| // fit) are retransmitted first. |
| for (auto nack_seq : ack.nack_seqs()) { |
| Nack(transaction, nack_seq, queue_delay, Status::Unavailable()); |
| } |
| |
| // Clear out outstanding packet references, propagating acks. |
| for (size_t i = send_tip_; i <= ack.ack_to_seq(); i++) { |
| OutstandingPacket& pkt = outstanding_[i - send_tip_]; |
| if (std::holds_alternative<OutstandingPacket::Sent>(pkt.state)) { |
| Ack(transaction, i, queue_delay); |
| } |
| } |
| } |
| |
| void PacketProtocol::OutstandingMessages::Ack(Transaction* transaction, |
| uint64_t ack_seq, |
| TimeDelta queue_delay) { |
| OutstandingPacket& pkt = outstanding_[ack_seq - send_tip_]; |
| auto& sent_packet = std::get<OutstandingPacket::Sent>(pkt.state); |
| auto request = std::move(sent_packet.request); |
| auto& bbr_sent_packet = sent_packet.bbr_sent_packet; |
| bbr_sent_packet.send_time += queue_delay; |
| transaction->QueueAck(bbr_sent_packet); |
| protocol_->state_->bdp_estimator.AckPacket(sent_packet.bdp_packet); |
| pkt.state = OutstandingPacket::Acked{}; |
| transaction->IncrementOutstandingTipOnCompletion(); |
| request.Ack(Status::Ok()); |
| } |
| |
| void PacketProtocol::OutstandingMessages::Nack(Transaction* transaction, |
| uint64_t nack_seq, |
| TimeDelta queue_delay, |
| const Status& status) { |
| OVERNET_TRACE(DEBUG) << "AckProcessor.Nack: seq=" << nack_seq |
| << " status=" << status << " send_tip=" << send_tip_; |
| assert(status.is_error()); |
| if (protocol_->state_.has_value()) { |
| protocol_->state_->ack_sender.OnNack( |
| transaction, nack_seq, status.code() == StatusCode::CANCELLED); |
| } |
| if (nack_seq < send_tip_) { |
| return; |
| } |
| OutstandingPacket& pkt = outstanding_[nack_seq - send_tip_]; |
| OVERNET_TRACE(DEBUG) << "AckProcessor.Nack: seq=" << nack_seq |
| << " state=" << pkt.state; |
| if (std::holds_alternative<OutstandingPacket::Pending>(pkt.state) || |
| std::holds_alternative<OutstandingPacket::PendingTailProbe>(pkt.state)) { |
| pkt.state = OutstandingPacket::Nacked{}; |
| transaction->IncrementOutstandingTipOnCompletion(); |
| } else if (auto* sent_packet = |
| std::get_if<OutstandingPacket::Sent>(&pkt.state)) { |
| assert(sent_packet->bbr_sent_packet.outgoing.sequence == nack_seq); |
| auto& bbr_sent_packet = sent_packet->bbr_sent_packet; |
| bbr_sent_packet.send_time += queue_delay; |
| transaction->QueueNack(bbr_sent_packet); |
| auto request = std::move(sent_packet->request); |
| pkt.state = OutstandingPacket::Nacked{}; |
| transaction->IncrementOutstandingTipOnCompletion(); |
| request.Ack(status); |
| } else if (std::holds_alternative<OutstandingPacket::Nacked>(pkt.state)) { |
| } else { |
| // Previously acked packet becomes nacked: this is an error that should be |
| // diagnosed during validation. |
| abort(); |
| } |
| } |
| |
| void PacketProtocol::OutstandingMessages::IncrementTip() { |
| while (!outstanding_.empty() && outstanding_.front().is_finalized()) { |
| send_tip_++; |
| outstanding_.pop_front(); |
| } |
| if (unsent_tip_ < send_tip_) { |
| unsent_tip_ = send_tip_; |
| } |
| } |
| |
| void PacketProtocol::ReceivedQueue::SetTip(uint64_t seq_idx, |
| TimeStamp received) { |
| assert(seq_idx >= 1); |
| uint64_t tip_idx = seq_idx - 1; |
| if (tip_idx <= received_tip_) { |
| return; |
| } |
| if (!EnsureValidReceivedPacket(tip_idx, received)) { |
| abort(); |
| } |
| received_packets_.erase( |
| received_packets_.begin(), |
| received_packets_.begin() + (tip_idx - received_tip_)); |
| received_tip_ = tip_idx; |
| } |
| |
| /////////////////////////////////////////////////////////////////////////////// |
| // Ack scheduling |
| |
| std::string PacketProtocol::AckSender::SentFullAcksString() const { |
| std::ostringstream out; |
| bool first = true; |
| for (uint64_t ack : sent_full_acks_) { |
| if (!first) { |
| out << ","; |
| } |
| first = false; |
| out << ack; |
| } |
| return "{" + out.str() + "}"; |
| } |
| |
| void PacketProtocol::AckSender::NeedAck(Transaction* transaction, |
| AckUrgency urgency) { |
| OVERNET_TRACE(DEBUG) << "AckSender.NeedAck" |
| << " urgency=" << urgency << " (from " << urgency_ << ")" |
| << " all_acks_acknowledged=" << all_acks_acknowledged_ |
| << " sent_full_acks=" << SentFullAcksString(); |
| |
| if (urgency <= urgency_) { |
| return; |
| } |
| |
| urgency_ = urgency; |
| assert(protocol_->state_.has_value()); |
| sent_full_acks_.clear(); |
| all_acks_acknowledged_ = false; |
| switch (urgency_) { |
| case AckUrgency::NOT_REQUIRED: |
| abort(); |
| case AckUrgency::SEND_SOON: { |
| const auto when = |
| protocol_->timer_->Now() + protocol_->TailLossProbeDelay(); |
| OVERNET_TRACE(DEBUG) << "AckSender.NeedAck: schedule ack start for " |
| << when; |
| suppress_need_acks_ = true; |
| send_ack_timer_.Reset(protocol_->timer_, when, [this](const Status& status) { |
| OVERNET_TRACE(DEBUG) << "AckSender.NeedAck: ack start --> " << status; |
| if (status.is_error()) { |
| return; |
| } |
| suppress_need_acks_ = false; |
| protocol_->stats_ |
| .tail_loss_probe_scheduled_because_ack_required_soon_timer_expired++; |
| protocol_->state_->outstanding_messages.ScheduleAck(); |
| }); |
| } break; |
| case AckUrgency::SEND_IMMEDIATELY: |
| suppress_need_acks_ = false; |
| send_ack_timer_.Reset(); |
| protocol_->state_->outstanding_messages.ForceSendAck(transaction); |
| break; |
| } |
| } |
| |
| void PacketProtocol::AckSender::AckSent(Transaction* transaction, |
| uint64_t seq_idx, bool partial) { |
| OVERNET_TRACE(DEBUG) << "AckSender.AckSent seq_idx=" << seq_idx |
| << " partial=" << partial |
| << " all_acks_acknowledged=" << all_acks_acknowledged_ |
| << " sent_full_acks=" << SentFullAcksString(); |
| |
| send_ack_timer_.Reset(); |
| if (!sent_full_acks_.empty()) { |
| assert(seq_idx > sent_full_acks_.back()); |
| } |
| urgency_ = AckUrgency::NOT_REQUIRED; |
| if (!partial) { |
| sent_full_acks_.push_back(seq_idx); |
| } else if (sent_full_acks_.empty()) { |
| protocol_->stats_.ack_required_soon_continue_partial_after_ack++; |
| NeedAck(transaction, AckUrgency::SEND_SOON); |
| } |
| } |
| |
| void PacketProtocol::AckSender::OnNack(Transaction* transaction, uint64_t seq, |
| bool shutting_down) { |
| OVERNET_TRACE(DEBUG) << "AckSender.OnNack" |
| << " sent_full_acks=" << SentFullAcksString() |
| << " seq=" << seq |
| << " all_acks_acknowledged=" << all_acks_acknowledged_; |
| auto it = |
| std::lower_bound(sent_full_acks_.begin(), sent_full_acks_.end(), seq); |
| if (it == sent_full_acks_.end() || *it != seq) { |
| return; |
| } |
| sent_full_acks_.erase(it); |
| if (sent_full_acks_.empty() && !shutting_down) { |
| protocol_->stats_.ack_required_soon_all_acks_nacked++; |
| NeedAck(transaction, AckUrgency::SEND_SOON); |
| } |
| } |
| |
| void PacketProtocol::AckSender::OnAck(uint64_t seq) { |
| OVERNET_TRACE(DEBUG) << "AckSender.OnAck" |
| << " sent_full_acks=" << SentFullAcksString() |
| << " seq=" << seq |
| << " all_acks_acknowledged=" << all_acks_acknowledged_; |
| auto it = |
| std::lower_bound(sent_full_acks_.begin(), sent_full_acks_.end(), seq); |
| if (it == sent_full_acks_.end() || *it != seq) { |
| return; |
| } |
| sent_full_acks_.clear(); |
| all_acks_acknowledged_ = true; |
| } |
| |
| /////////////////////////////////////////////////////////////////////////////// |
| // Retransmit scheduling |
| |
| void PacketProtocol::OutstandingMessages::ScheduleRetransmit() { |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.ScheduleRetransmit: rto_timer=" |
| << retransmit_timeout_.has_value() |
| << " deadline=" << RetransmitDeadline(); |
| if (retransmit_timeout_.has_value()) { |
| return; |
| } |
| if (auto timeout = RetransmitDeadline(); timeout.has_value()) { |
| retransmit_timeout_.Reset( |
| protocol_->timer_, *timeout, |
| [protocol = protocol_](const Status& status) { |
| ScopedModule in_pp(protocol); |
| protocol->InTransaction([=](Transaction* transaction) { |
| OVERNET_TRACE(DEBUG) |
| << "OutstandingMessages.ScheduleRetransmit --> " << status |
| << " protocol_open=" << !transaction->Closing(); |
| if (transaction->Closing()) { |
| return; |
| } |
| if (status.is_error()) { |
| protocol->state_->outstanding_messages.NackAll(transaction); |
| return; |
| } |
| protocol->state_->outstanding_messages.CheckRetransmit(transaction); |
| }); |
| }); |
| } |
| } |
| |
| Optional<TimeStamp> PacketProtocol::OutstandingMessages::RetransmitDeadline() { |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.RetransmitDeadline: " |
| << OutstandingString(); |
| for (const auto& outstanding : outstanding_) { |
| if (!outstanding.is_finalized()) { |
| return outstanding.sent + protocol_->RetransmitDelay(); |
| } |
| } |
| return Nothing; |
| } |
| |
| void PacketProtocol::OutstandingMessages::CheckRetransmit( |
| Transaction* transaction) { |
| if (!protocol_->state_.has_value()) { |
| return; |
| } |
| retransmit_timeout_.Reset(); |
| const auto nack_before = |
| protocol_->timer_->Now() - protocol_->RetransmitDelay(); |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.CheckRetransmit: nack_before=" |
| << nack_before |
| << " (current_rtt=" << protocol_->CurrentRTT() << ")"; |
| for (size_t i = 0; i < outstanding_.size(); i++) { |
| if (outstanding_[i].is_finalized()) { |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.CheckRetransmit: seq " |
| << (send_tip_ + i) << " finalized: STOP"; |
| break; |
| } |
| const auto sent = outstanding_[i].sent; |
| if (sent > nack_before) { |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.CheckRetransmit: seq " |
| << (send_tip_ + i) << " sent at " << sent |
| << ": STOP"; |
| break; |
| } |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.CheckRetransmit: seq " |
| << (send_tip_ + i) << " sent at " << sent << ": NACK"; |
| Nack(transaction, send_tip_ + i, TimeDelta::Zero(), Status::Unavailable()); |
| } |
| ScheduleRetransmit(); |
| } |
| |
| PacketProtocol::OutstandingMessages::~OutstandingMessages() { |
| protocol_->InTransaction( |
| [this](Transaction* transaction) { NackAll(transaction); }); |
| } |
| |
| void PacketProtocol::OutstandingMessages::NackAll(Transaction* transaction) { |
| OVERNET_TRACE(DEBUG) << "OutstandingMessages.NackAll"; |
| for (uint64_t i = send_tip_, end = send_tip_ + outstanding_.size(); i < end; |
| i++) { |
| if (outstanding_[i - send_tip_].is_finalized()) { |
| continue; |
| } |
| Nack(transaction, i, TimeDelta::Zero(), Status::Cancelled()); |
| } |
| } |
| |
| /////////////////////////////////////////////////////////////////////////////// |
| // Utilities |
| |
| TimeDelta PacketProtocol::CurrentRTT() const { |
| return std::max(TimeDelta::FromMilliseconds(1), state_->bbr.rtt()); |
| } |
| |
| TimeDelta PacketProtocol::RetransmitDelay() const { |
| constexpr const auto kMinRetransmitDelay = TimeDelta::FromSeconds(1); |
| constexpr const int kRTTScaling = 4; |
| auto rtt = CurrentRTT(); |
| if (rtt == TimeDelta::PositiveInf() || |
| rtt < kMinRetransmitDelay / kRTTScaling) { |
| return kMinRetransmitDelay; |
| } |
| return kRTTScaling * rtt; |
| } |
| |
| TimeDelta PacketProtocol::TailLossProbeDelay() const { |
| constexpr const auto kMinTailLossProbeDelay = TimeDelta::FromMilliseconds(1); |
| constexpr const int kRTTScaling = 4; |
| auto rtt = CurrentRTT(); |
| if (rtt == TimeDelta::PositiveInf() || |
| rtt < kRTTScaling * kMinTailLossProbeDelay) { |
| return kMinTailLossProbeDelay; |
| } |
| return rtt / kRTTScaling; |
| } |
| |
| PacketProtocol::Codec* PacketProtocol::NullCodec() { |
| class NullCodec final : public Codec { |
| public: |
| NullCodec() : Codec(Border::None()) {} |
| StatusOr<Slice> Encode(uint64_t seq_idx, Slice src) const override { |
| return src; |
| } |
| StatusOr<Slice> Decode(uint64_t seq_idx, Slice src) const override { |
| return src; |
| } |
| }; |
| static NullCodec null_codec; |
| return &null_codec; |
| } |
| |
| } // namespace overnet |
