src/connectivity/bluetooth/core/bt-host/l2cap/enhanced_retransmission_mode_tx_engine.cc

// Copyright 2019 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/bluetooth/core/bt-host/public/pw_bluetooth_sapphire/internal/host/l2cap/enhanced_retransmission_mode_tx_engine.h"

#include <limits>

#include "src/connectivity/bluetooth/core/bt-host/public/pw_bluetooth_sapphire/internal/host/common/assert.h"
#include "src/connectivity/bluetooth/core/bt-host/public/pw_bluetooth_sapphire/internal/host/common/log.h"
#include "src/connectivity/bluetooth/core/bt-host/public/pw_bluetooth_sapphire/internal/host/l2cap/frame_headers.h"

namespace bt::l2cap::internal {

using Engine = EnhancedRetransmissionModeTxEngine;

namespace {

// Returns the number of frames within the range from |low| to |high|, inclusive
// of |low|, but exclusive of |high|. Returns zero if |low == high|.
uint8_t NumFramesBetween(uint8_t low, uint8_t high) {
  if (high < low) {
    high += (EnhancedControlField::kMaxSeqNum + 1);
  }
  return high - low;
}

}  // namespace

Engine::EnhancedRetransmissionModeTxEngine(
    ChannelId channel_id,
    uint16_t max_tx_sdu_size,
    uint8_t max_transmissions,
    uint8_t n_frames_in_tx_window,
    TxChannel& channel,
    ConnectionFailureCallback connection_failure_callback,
    pw::async::Dispatcher& dispatcher)
    : TxEngine(channel_id, max_tx_sdu_size, channel),
      pw_dispatcher_(dispatcher),
      max_transmissions_(max_transmissions),
      n_frames_in_tx_window_(n_frames_in_tx_window),
      connection_failure_callback_(std::move(connection_failure_callback)),
      expected_ack_seq_(0),
      next_tx_seq_(0),
      last_tx_seq_(0),
      req_seqnum_(0),
      retransmitted_range_during_poll_(false),
      n_receiver_ready_polls_sent_(0),
      remote_is_busy_(false) {
  BT_DEBUG_ASSERT(n_frames_in_tx_window_);
  receiver_ready_poll_task_.set_function(
      [this](pw::async::Context /*ctx*/, pw::Status status) {
        if (!status.ok()) {
          return;
        }
        SendReceiverReadyPoll();
        StartMonitorTimer();
      });
  monitor_task_.set_function(
      [this](pw::async::Context /*ctx*/, pw::Status status) {
        if (!status.ok()) {
          return;
        }
        if (max_transmissions_ == 0 ||
            n_receiver_ready_polls_sent_ < max_transmissions_) {
          SendReceiverReadyPoll();
          StartMonitorTimer();
        } else {
          connection_failure_callback_();  // May invalidate |self|.
        }
      });
}

void Engine::NotifySduQueued() {
  std::optional<ByteBufferPtr> sdu = channel().GetNextQueuedSdu();
  BT_ASSERT(sdu);
  ProcessSdu(std::move(*sdu));
}

void Engine::ProcessSdu(ByteBufferPtr sdu) {
  BT_ASSERT(sdu);
  // TODO(https://fxbug.dev/42054330): Add support for segmentation
  if (sdu->size() > max_tx_sdu_size()) {
    bt_log(INFO,
           "l2cap",
           "SDU size exceeds channel TxMTU (channel-id: %#.4x)",
           channel_id());
    return;
  }

  const auto seq_num = GetNextTxSeq();
  SimpleInformationFrameHeader header(seq_num);
  DynamicByteBuffer frame(sizeof(header) + sdu->size());
  auto body = frame.mutable_view(sizeof(header));
  frame.WriteObj(header);
  sdu->Copy(&body);

  // TODO(https://fxbug.dev/42086227): Limit the size of the queue.
  pending_pdus_.push_back(std::move(frame));
  MaybeSendQueuedData();
}

void Engine::UpdateAckSeq(uint8_t new_seq, bool is_poll_response) {
  // TODO(quiche): Reconsider this assertion if we allow reconfiguration of the
  // TX window.
  BT_DEBUG_ASSERT_MSG(NumUnackedFrames() <= n_frames_in_tx_window_,
                      "(NumUnackedFrames() = %u, n_frames_in_tx_window_ = %u, "
                      "expected_ack_seq_ = %u, last_tx_seq_ = %u)",
                      NumUnackedFrames(),
                      n_frames_in_tx_window_,
                      expected_ack_seq_,
                      last_tx_seq_);

  const auto n_frames_acked = NumFramesBetween(expected_ack_seq_, new_seq);
  if (n_frames_acked > NumUnackedFrames()) {
    // Peer acknowledgment of our outbound data (ReqSeq) exceeds the sequence
    // numbers of yet-acknowledged data that we've sent to that peer. See
    // conditions "With-Invalid-ReqSeq" and "With-Invalid-ReqSeq-Retrans" in
    // Core Spec v5.0 Vol 3 Part A Sec 8.6.5.5.
    bt_log(WARN,
           "l2cap",
           "Received acknowledgment for %hhu frames but only %hhu frames are "
           "pending",
           n_frames_acked,
           NumUnackedFrames());
    connection_failure_callback_();  // May invalidate |self|.
    return;
  }

  // Perform Stop-MonitorTimer when "F = 1," per Core Spec v5.0, Vol 3, Part A,
  // Sec 8.6.5.7–8.
  if (is_poll_response) {
    monitor_task_.Cancel();
  }

  BT_ASSERT(!(range_request_.has_value() && single_request_.has_value()));
  if (ProcessSingleRetransmitRequest(new_seq, is_poll_response) ==
      UpdateAckSeqAction::kConsumeAckSeq) {
    return;
  }

  auto n_frames_to_discard = n_frames_acked;
  while (n_frames_to_discard) {
    BT_DEBUG_ASSERT(!pending_pdus_.empty());
    pending_pdus_.pop_front();
    --n_frames_to_discard;
  }

  expected_ack_seq_ = new_seq;
  if (expected_ack_seq_ == next_tx_seq_) {
    receiver_ready_poll_task_.Cancel();
  }

  const auto range_request = std::exchange(range_request_, std::nullopt);

  // RemoteBusy is cleared as the first action to take when receiving a REJ per
  // Core Spec v5.0 Vol 3, Part A, Sec 8.6.5.9–11, so their corresponding member
  // variables shouldn't be both set.
  BT_ASSERT(!(range_request.has_value() && remote_is_busy_));
  bool should_retransmit = range_request.has_value();

  // This implements the logic for RejActioned in the Recv {I,RR,REJ} (F=1)
  // event for all of the receiver states (Core Spec v5.0 Vol 3, Part A,
  // Sec 8.6.5.9–11).
  if (is_poll_response) {
    if (retransmitted_range_during_poll_) {
      should_retransmit = false;
      retransmitted_range_during_poll_ = false;
    } else {
      should_retransmit = true;
    }
  }

  if (remote_is_busy_) {
    return;
  }

  // This implements the logic for PbitOutstanding in the Recv REJ (F=0) event
  // for all of the receiver states (Core Spec v5.0 Vol 3, Part A,
  // Sec 8.6.5.9–11).
  if (range_request.has_value() && !is_poll_response &&
      monitor_task_.is_pending()) {
    retransmitted_range_during_poll_ = true;
  }

  if (should_retransmit) {
    const bool set_is_poll_response =
        range_request.value_or(RangeRetransmitRequest{}).is_poll_request;
    if (!RetransmitUnackedData(std::nullopt, set_is_poll_response)) {
      return;
    }
  }

  MaybeSendQueuedData();

  // TODO(quiche): Restart the receiver_ready_poll_task_, if there's any
  // remaining unacknowledged data.
}

void Engine::UpdateReqSeq(uint8_t new_seq) { req_seqnum_ = new_seq; }

void Engine::ClearRemoteBusy() {
  // TODO(quiche): Maybe clear backpressure on the Channel (subject to TxWindow
  // contraints).
  remote_is_busy_ = false;
}

void Engine::SetRemoteBusy() {
  // TODO(https://fxbug.dev/42086338): Signal backpressure to the Channel.
  remote_is_busy_ = true;
  receiver_ready_poll_task_.Cancel();
}

void Engine::SetSingleRetransmit(bool is_poll_request) {
  BT_ASSERT(!single_request_.has_value());
  BT_ASSERT(!range_request_.has_value());
  // Store SREJ state for UpdateAckSeq to handle.
  single_request_ = SingleRetransmitRequest{.is_poll_request = is_poll_request};
}

void Engine::SetRangeRetransmit(bool is_poll_request) {
  BT_ASSERT(!single_request_.has_value());
  BT_ASSERT(!range_request_.has_value());
  // Store REJ state for UpdateAckSeq to handle.
  range_request_ = RangeRetransmitRequest{.is_poll_request = is_poll_request};
}

void Engine::MaybeSendQueuedData() {
  if (remote_is_busy_ || monitor_task_.is_pending()) {
    return;
  }

  // Find the first PDU that has not already been transmitted (if any).
  // * This is not necessarily the first PDU, because that may have been
  //   transmited already, and is just pending acknowledgement.
  // * This is not necessarily the last PDU, because earlier PDUs may have been
  //   queued without having been sent over-the-air (due, e.g., to tx_window
  //   constraints).
  //
  // TODO(quiche): Consider if there's a way to do this that isn't O(n).
  auto it = std::find_if(
      pending_pdus_.begin(), pending_pdus_.end(), [](const auto& pending_pdu) {
        return pending_pdu.tx_count == 0;
      });

  while (it != pending_pdus_.end() &&
         NumUnackedFrames() < n_frames_in_tx_window_) {
    BT_DEBUG_ASSERT(it->tx_count == 0);
    SendPdu(&*it);
    last_tx_seq_ = it->buf.To<SimpleInformationFrameHeader>().tx_seq();
    ++it;
  }
}

Engine::UpdateAckSeqAction Engine::ProcessSingleRetransmitRequest(
    uint8_t new_seq, bool is_poll_response) {
  const auto single_request = std::exchange(single_request_, std::nullopt);
  BT_ASSERT(!(single_request.has_value() && remote_is_busy_));
  if (!single_request.has_value()) {
    return UpdateAckSeqAction::kDiscardAcknowledged;
  }

  // This implements the logic for SrejActioned=TRUE in the Recv SREJ (P=0)
  // (F=1) event for all of the receiver states (Core Spec v5.0 Vol 3, Part A,
  // Sec 8.6.5.9–11).
  if (is_poll_response && retransmitted_single_during_poll_.has_value() &&
      new_seq == *retransmitted_single_during_poll_) {
    // Only a SREJ (F=1) with a matching AckSeq can clear this state, so if this
    // duplicate suppression isn't performed it sticks around even when we sent
    // the next poll request. AckSeq is modulo kMaxSeqNum + 1 so it'll roll over
    // and potentially suppress retransmission of a different (non-duplicate)
    // packet in a later poll request-response cycle. Unfortunately this isn't
    // fixed as of Core Spec v5.2 so it's implemented as written.
    retransmitted_single_during_poll_.reset();

    // Return early to suppress duplicate retransmission.
    return UpdateAckSeqAction::kConsumeAckSeq;
  }

  // This implements the logic for PbitOutstanding in the Recv SREJ (P=0) (F=0)
  // and Recv SREJ(P=1) events for all of the receiver states (Core Spec v5.0
  // Vol 3, Part A, Sec 8.6.5.9–11).
  if (!is_poll_response && monitor_task_.is_pending()) {
    retransmitted_single_during_poll_ = new_seq;
  }

  if (!RetransmitUnackedData(new_seq, single_request->is_poll_request)) {
    return UpdateAckSeqAction::kConsumeAckSeq;
  }

  // Only "single requests" that are poll requests acknowledge previous I-Frames
  // and cause initial transmission of queued SDUs, per Core Spec v5.0, Vol 3,
  // Part A, Sec 8.6.1.4.
  if (single_request->is_poll_request) {
    return UpdateAckSeqAction::kDiscardAcknowledged;
  }
  return UpdateAckSeqAction::kConsumeAckSeq;
}

void Engine::StartReceiverReadyPollTimer() {
  BT_DEBUG_ASSERT(!monitor_task_.is_pending());
  n_receiver_ready_polls_sent_ = 0;
  receiver_ready_poll_task_.Cancel();
  receiver_ready_poll_task_.PostAfter(kErtmReceiverReadyPollTimerDuration);
}

void Engine::StartMonitorTimer() {
  BT_DEBUG_ASSERT(!receiver_ready_poll_task_.is_pending());
  monitor_task_.Cancel();
  monitor_task_.PostAfter(kErtmMonitorTimerDuration);
}

void Engine::SendReceiverReadyPoll() {
  SimpleReceiverReadyFrame frame;
  frame.set_receive_seq_num(req_seqnum_);
  frame.set_is_poll_request();
  ++n_receiver_ready_polls_sent_;
  BT_ASSERT_MSG(max_transmissions_ == 0 ||
                    n_receiver_ready_polls_sent_ <= max_transmissions_,
                "(n_receiver_ready_polls_sent_ = %u, "
                "max_transmissions = %u)",
                n_receiver_ready_polls_sent_,
                max_transmissions_);
  channel().SendFrame(
      std::make_unique<DynamicByteBuffer>(BufferView(&frame, sizeof(frame))));
}

uint8_t Engine::GetNextTxSeq() {
  auto ret = next_tx_seq_;
  ++next_tx_seq_;
  if (next_tx_seq_ > EnhancedControlField::kMaxSeqNum) {
    next_tx_seq_ = 0;
  }
  return ret;
}

uint8_t Engine::NumUnackedFrames() {
  if (pending_pdus_.empty()) {
    // Initially, |ack_seqnum_ == last_tx_seq_ == 0|, but the number of
    // unacknowledged frames is 0, not 1.
    return 0;
  } else if (pending_pdus_.front().tx_count == 0) {
    // While we have some data queued, none of that data has been sent
    // over-the-air. This might happen, e.g., transiently in QueueSdu().
    return 0;
  } else {
    // Having ascertained that some data _is_ in flight, the number of frames in
    // flight is given by the expression below.
    return NumFramesBetween(
        expected_ack_seq_,
        last_tx_seq_ + 1  // Include frame with |last_tx_seq_| in count
    );
  }
}

void Engine::SendPdu(PendingPdu* pdu) {
  BT_DEBUG_ASSERT(pdu);
  pdu->buf.AsMutable<SimpleInformationFrameHeader>()->set_receive_seq_num(
      req_seqnum_);

  // Prevent tx_count from overflowing to zero, as that would be
  // indistinguishable from "never transmitted." This is only possible when
  // configured for infinite retransmissions, so there is no benefit to having
  // an accurate tx_count after each frame's initial transmission.
  if (pdu->tx_count != std::numeric_limits<decltype(pdu->tx_count)>::max()) {
    pdu->tx_count++;
  }
  StartReceiverReadyPollTimer();
  channel().SendFrame(std::make_unique<DynamicByteBuffer>(pdu->buf));
}

bool Engine::RetransmitUnackedData(std::optional<uint8_t> only_with_seq,
                                   bool set_is_poll_response) {
  // The receive engine should have cleared the remote busy condition before
  // calling any method that would cause us (the transmit engine) to retransmit
  // unacked data. See, e.g., Core Spec v5.0, Volume 3, Part A, Table 8.6, row
  // "Recv REJ (F=0)".
  BT_DEBUG_ASSERT(!remote_is_busy_);

  // Any peer actions that cause retransmission indicate the peer is alive. This
  // is in conflict with Core Spec v5.0, Vol 3, Part A, Sec 8.6.5.8, which only
  // stops the MonitorTimer when a poll response is received and omits what to
  // do when REJ or SREJ cause retransmission. However, this behavior of
  // canceling the MonitorTimer "early" is in line with Sequence Diagram
  // Fig 4.94, L2CAP Test Spec v5.0.2 Section 4.9.7.24, among others that show
  // REJ or SREJ causing the receiver to cancel its MonitorTimer. We follow the
  // latter behavior because (1) it's less ambiguous (2) it makes sense and (3)
  // we need to pass those tests.
  monitor_task_.Cancel();

  const auto n_to_send = NumUnackedFrames();
  BT_ASSERT(n_to_send <= n_frames_in_tx_window_);
  BT_DEBUG_ASSERT(n_to_send <= pending_pdus_.size());

  auto cur_frame = pending_pdus_.begin();
  auto last_frame = std::next(cur_frame, n_to_send);
  for (; cur_frame != last_frame; cur_frame++) {
    BT_DEBUG_ASSERT(cur_frame != pending_pdus_.end());

    const auto control_field =
        cur_frame->buf.To<SimpleInformationFrameHeader>();
    if (only_with_seq.has_value() && control_field.tx_seq() != *only_with_seq) {
      continue;
    }

    // Core Spec v5.0, Vol 3, Part A, Sec 5.4: "In Enhanced Retransmission mode
    // a value of zero for MaxTransmit means infinite retransmissions."
    if (max_transmissions_ != 0 && cur_frame->tx_count >= max_transmissions_) {
      BT_ASSERT_MSG(cur_frame->tx_count == max_transmissions_,
                    "%hhu != %hhu",
                    cur_frame->tx_count,
                    max_transmissions_);
      connection_failure_callback_();
      return false;
    }

    if (set_is_poll_response) {
      cur_frame->buf.AsMutable<EnhancedControlField>()->set_is_poll_response();

      // Per "Retransmit-I-frames" of Core Spec v5.0 Vol 3, Part A, Sec 8.6.5.6,
      // "the F-bit of all other [than the first] unacknowledged I-frames sent
      // shall be 0," so clear this for subsequent iterations.
      set_is_poll_response = false;
    }

    // TODO(https://fxbug.dev/42087625): If the task is already running, we
    // should not restart it.
    SendPdu(&*cur_frame);
    *cur_frame->buf.AsMutable<EnhancedControlField>() = control_field;
  }

  return true;
}

}  // namespace bt::l2cap::internal