pw_bluetooth_proxy: ProxyHost supports multiple sends

Expand ProxyHost H4 packet storage to multiple buffers to support
multiple in-flight ACL sends. Add default constructor to
H4PacketWithH4 to allow the construction of std::arrays of those
packets in tests.

I've manually confirmed that tests pass with a variety of values
for kNumH4Buffs, including 1, 2, 20, & 100.

Fixed: 348680331
Bug: 326499764
Change-Id: I2e7133e8fa571183b30c4b2c0179ac144170504d
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/220573
Lint: Lint 🤖 <[email protected]>
Commit-Queue: Ali Saeed <[email protected]>
Reviewed-by: David Rees <[email protected]>

pw_bluetooth_proxy/proxy_host.cc

@@ -23,14 +23,25 @@
 
 namespace pw::bluetooth::proxy {
 
+std::array<containers::Pair<uint8_t*, bool>, ProxyHost::kNumH4Buffs>
+ProxyHost::InitOccupiedMap() {
+ std::array<containers::Pair<uint8_t*, bool>, kNumH4Buffs> arr;
+ acl_send_mutex_.lock();
+ for (size_t i = 0; i < kNumH4Buffs; ++i) {
+ arr[i] = {h4_buffs_[i].data(), false};
+ }
+ acl_send_mutex_.unlock();
+ return arr;
+}
+
 ProxyHost::ProxyHost(
  pw::Function<void(H4PacketWithHci&& packet)>&& send_to_host_fn,
  pw::Function<void(H4PacketWithH4&& packet)>&& send_to_controller_fn,
  uint16_t le_acl_credits_to_reserve)
  : hci_transport_(std::move(send_to_host_fn),
  std::move(send_to_controller_fn)),
  acl_data_channel_(hci_transport_, le_acl_credits_to_reserve),
- acl_send_pending_(false) {}
+ h4_buff_occupied_(InitOccupiedMap()) {}
 
 void ProxyHost::HandleH4HciFromHost(H4PacketWithH4&& h4_packet) {
  hci_transport_.SendToController(std::move(h4_packet));
@@ -139,10 +150,22 @@ void ProxyHost::HandleH4HciFromController(H4PacketWithHci&& h4_packet) {
 void ProxyHost::Reset() {
  acl_send_mutex_.lock();
  acl_data_channel_.Reset();
- acl_send_pending_ = false;
+ for (const auto& [buff, _] : h4_buff_occupied_) {
+ h4_buff_occupied_.at(buff) = false;
+ }
  acl_send_mutex_.unlock();
 }
 
+std::optional<pw::span<uint8_t>> ProxyHost::ReserveH4Buff() {
+ for (const auto& [buff, occupied] : h4_buff_occupied_) {
+ if (!occupied) {
+ h4_buff_occupied_.at(buff) = true;
+ return {{buff, kH4BuffSize}};
+ }
+ }
+ return std::nullopt;
+}
+
 pw::Status ProxyHost::SendGattNotify(uint16_t connection_handle,
  uint16_t attribute_handle,
  pw::span<const uint8_t> attribute_value) {
@@ -165,22 +188,28 @@ pw::Status ProxyHost::SendGattNotify(uint16_t connection_handle,
  return pw::Status::InvalidArgument();
  }
 
- H4PacketWithH4 h4_att_notify({});
+ H4PacketWithH4 h4_att_notify;
  {
  acl_send_mutex_.lock();
- // TODO: https://pwbug.dev/348680331 - Currently ProxyHost only supports 1
- // in-flight ACL send, increase this to support multiple.
- if (acl_send_pending_) {
+ std::optional<span<uint8_t>> h4_buff = ReserveH4Buff();
+ if (!h4_buff) {
  acl_send_mutex_.unlock();
  return pw::Status::Unavailable();
  }
- acl_send_pending_ = true;
 
  size_t acl_packet_size =
  emboss::AttNotifyOverAcl::MinSizeInBytes() + attribute_value.size();
+ size_t h4_packet_size = sizeof(emboss::H4PacketType) + acl_packet_size;
+
+ if (h4_packet_size > h4_buff->size()) {
+ PW_LOG_ERROR("Buffer is too small for H4 packet. So will not send.");
+ acl_send_mutex_.unlock();
+ return pw::Status::InvalidArgument();
+ }
+
  emboss::AttNotifyOverAclWriter att_notify =
  emboss::MakeAttNotifyOverAclView(attribute_value.size(),
- H4HciSubspan(h4_buff_).data(),
+ H4HciSubspan(*h4_buff).data(),
  acl_packet_size);
  if (!att_notify.IsComplete()) {
  PW_LOG_ERROR("Buffer is too small for ATT Notify. So will not send.");
@@ -191,27 +220,25 @@ pw::Status ProxyHost::SendGattNotify(uint16_t connection_handle,
  BuildAttNotify(
  att_notify, connection_handle, attribute_handle, attribute_value);
 
- size_t h4_packet_size = 1 + acl_packet_size;
- H4PacketWithH4 h4_temp(pw::span(h4_buff_.data(), h4_packet_size),
- [this](const uint8_t* buffer) {
- acl_send_mutex_.lock();
- PW_CHECK_PTR_EQ(
- buffer,
- h4_buff_.data(),
- "Received release callback for buffer that "
- "doesn't match our buffer.");
- PW_LOG_DEBUG("H4 packet release fn called.");
- acl_send_pending_ = false;
- acl_send_mutex_.unlock();
- });
+ H4PacketWithH4 h4_temp(
+ span(h4_buff->data(), h4_packet_size),
+ /*release_fn=*/[this](const uint8_t* buffer) {
+ acl_send_mutex_.lock();
+ PW_CHECK(h4_buff_occupied_.contains(const_cast<uint8_t*>(buffer)),
+ "Received release callback for invalid buffer address.");
+ PW_LOG_DEBUG("H4 packet release fn called.");
+ h4_buff_occupied_.at(const_cast<uint8_t*>(buffer)) = false;
+ acl_send_mutex_.unlock();
+ });
  h4_temp.SetH4Type(emboss::H4PacketType::ACL_DATA);
  h4_att_notify = std::move(h4_temp);
  acl_send_mutex_.unlock();
  }
 
  // H4 packet is hereby moved. Either ACL data channel will move packet to
  // controller or will be unable to send packet. In either case, packet will be
- // destructed, so its release function will clear the `acl_send_pending` flag.
+ // destructed, so its release function will clear the corresponding flag in
+ // `h4_buff_occupied_`.
  if (!acl_data_channel_.SendAcl(std::move(h4_att_notify))) {
  return pw::Status::Unavailable();
  }

pw_bluetooth_proxy/proxy_host_test.cc

@@ -960,42 +960,6 @@ TEST(GattNotifyTest, SendGattNotify2ByteAttribute) {
  EXPECT_EQ(capture.sends_called, 1);
 }
 
-TEST(GattNotifyTest, SendGattNotifyUnavailableWhenPending) {
- struct {
- int sends_called = 0;
- H4PacketWithH4 released_packet{{}};
- } capture;
-
- pw::Function<void(H4PacketWithHci && packet)>&& send_to_host_fn(
- []([[maybe_unused]] H4PacketWithHci&& packet) {});
- pw::Function<void(H4PacketWithH4 && packet)> send_to_controller_fn(
- [&capture](H4PacketWithH4&& packet) {
- ++capture.sends_called;
- capture.released_packet = std::move(packet);
- });
-
- ProxyHost proxy = ProxyHost(
- std::move(send_to_host_fn), std::move(send_to_controller_fn), 2);
- // Allow proxy to reserve 2 credit.
- SendReadBufferResponseFromController(proxy, 2);
-
- std::array<uint8_t, 2> attribute_value = {0xAB, 0xCD};
- EXPECT_TRUE(proxy.SendGattNotify(123, 345, pw::span(attribute_value)).ok());
- // Only one send is allowed at a time, so PW_STATUS_UNAVAILABLE will be
- // returned until the pending packet is destructed.
- EXPECT_EQ(proxy.SendGattNotify(123, 345, pw::span(attribute_value)),
- PW_STATUS_UNAVAILABLE);
- capture.released_packet.~H4PacketWithH4();
- EXPECT_TRUE(proxy.SendGattNotify(123, 345, pw::span(attribute_value)).ok());
- EXPECT_EQ(proxy.SendGattNotify(123, 345, pw::span(attribute_value)),
- PW_STATUS_UNAVAILABLE);
- EXPECT_EQ(capture.sends_called, 2);
-
- // If captured packet is not reset here, it may destruct after the proxy and
- // lead to a crash when it tries to lock the proxy's destructed mutex.
- capture.released_packet.ResetAndReturnReleaseFn();
-}
-
 TEST(GattNotifyTest, SendGattNotifyReturnsErrorForInvalidArgs) {
  pw::Function<void(H4PacketWithHci && packet)>&& send_to_host_fn(
  []([[maybe_unused]] H4PacketWithHci&& packet) {});
@@ -1561,11 +1525,10 @@ TEST(DisconnectionCompleteTest, CanReuseConnectionHandleAfterDisconnection) {
 
 // ########## ResetTest
 
-TEST(ResetTest, ResetClearsPendingSendAndActiveConnections) {
+TEST(ResetTest, ResetClearsActiveConnections) {
  struct {
  int sends_called = 0;
  uint16_t connection_handle = 0x123;
- H4PacketWithH4 released_packet{{}};
  } capture;
 
  pw::Function<void(H4PacketWithHci && packet)> send_to_host_fn(
@@ -1592,12 +1555,7 @@ TEST(ResetTest, ResetClearsPendingSendAndActiveConnections) {
  EXPECT_EQ(view.nocp_data()[0].num_completed_packets().Read(), 1);
  });
  pw::Function<void(H4PacketWithH4 && packet)> send_to_controller_fn(
- [&capture](H4PacketWithH4&& packet) {
- ++capture.sends_called;
- // Capture the packet so proxy's buffer remains marked as occupied when
- // reset is called.
- capture.released_packet = std::move(packet);
- });
+ [&capture](H4PacketWithH4&& packet) { ++capture.sends_called; });
 
  ProxyHost proxy = ProxyHost(
  std::move(send_to_host_fn), std::move(send_to_controller_fn), 2);
@@ -1608,37 +1566,28 @@ TEST(ResetTest, ResetClearsPendingSendAndActiveConnections) {
  .SendGattNotify(
  capture.connection_handle, 1, pw::span(attribute_value))
  .ok());
- // Proxy still has a free credit, but send should fail, as buffer is occupied.
- EXPECT_EQ(proxy.SendGattNotify(1, 1, pw::span(attribute_value)),
- PW_STATUS_UNAVAILABLE);
 
  proxy.Reset();
 
  EXPECT_EQ(proxy.GetNumFreeLeAclPackets(), 0);
  // Reset should not have cleared `le_acl_credits_to_reserve`, so proxy should
  // still indicate the capability.
  EXPECT_TRUE(proxy.HasSendAclCapability());
- // Reset marks buffer as unoccupied, but also resets credits, so send should
- // still fail.
+ // Reset clears credit reservation, so send should fail.
  EXPECT_EQ(proxy.SendGattNotify(1, 1, pw::span(attribute_value)),
  PW_STATUS_UNAVAILABLE);
 
  // Re-initialize AclDataChannel with 2 credits.
  SendReadBufferResponseFromController(proxy, 2);
 
- // Send ACL on random handle to expend one credit. The success of this send
- // indicates that the reset has properly cleared `acl_send_pending_`.
+ // Send ACL on random handle to expend one credit.
  EXPECT_TRUE(proxy.SendGattNotify(1, 1, pw::span(attribute_value)).ok());
  // This should have no effect, as the reset has cleared our active connection
  // on this handle.
  SendNumberOfCompletedPackets(
  proxy,
  FlatMap<uint16_t, uint16_t, 1>({{{capture.connection_handle, 1}}}));
  EXPECT_EQ(proxy.GetNumFreeLeAclPackets(), 1);
-
- // If captured packet is not reset here, it may destruct after the proxy and
- // lead to a crash when it tries to lock the proxy's destructed mutex.
- capture.released_packet.ResetAndReturnReleaseFn();
 }
 
 TEST(ResetTest, ProxyHandlesMultipleResets) {
@@ -1678,5 +1627,144 @@ TEST(ResetTest, ProxyHandlesMultipleResets) {
  EXPECT_EQ(sends_called, 2);
 }
 
+// ########## MultiSendTest
+
+TEST(MultiSendTest, CanOccupyAllThenReuseEachBuffer) {
+ constexpr size_t kMaxSends = ProxyHost::GetNumSimultaneousAclSendsSupported();
+ struct {
+ size_t sends_called = 0;
+ std::array<H4PacketWithH4, 2 * kMaxSends> released_packets;
+ } capture;
+
+ pw::Function<void(H4PacketWithHci && packet)>&& send_to_host_fn(
+ []([[maybe_unused]] H4PacketWithHci&& packet) {});
+ pw::Function<void(H4PacketWithH4 && packet)> send_to_controller_fn(
+ [&capture](H4PacketWithH4&& packet) {
+ // Capture all packets to prevent their destruction.
+ capture.released_packets[capture.sends_called++] = std::move(packet);
+ });
+
+ ProxyHost proxy = ProxyHost(std::move(send_to_host_fn),
+ std::move(send_to_controller_fn),
+ 2 * kMaxSends);
+ // Allow proxy to reserve enough credits to send twice the number of
+ // simultaneous sends supported by proxy.
+ SendReadBufferResponseFromController(proxy, 2 * kMaxSends);
+
+ std::array<uint8_t, 1> attribute_value = {0xF};
+ // Occupy all send buffers.
+ for (size_t i = 0; i < kMaxSends; ++i) {
+ EXPECT_TRUE(proxy.SendGattNotify(123, 345, pw::span(attribute_value)).ok());
+ }
+ EXPECT_EQ(proxy.GetNumFreeLeAclPackets(), kMaxSends);
+ EXPECT_EQ(proxy.SendGattNotify(123, 345, pw::span(attribute_value)),
+ PW_STATUS_UNAVAILABLE);
+
+ // Confirm we can release and reoccupy each buffer slot.
+ for (size_t i = 0; i < kMaxSends; ++i) {
+ capture.released_packets[i].~H4PacketWithH4();
+ EXPECT_TRUE(proxy.SendGattNotify(123, 345, pw::span(attribute_value)).ok());
+ EXPECT_EQ(proxy.SendGattNotify(123, 345, pw::span(attribute_value)),
+ PW_STATUS_UNAVAILABLE);
+ }
+ EXPECT_EQ(capture.sends_called, 2 * kMaxSends);
+
+ // If captured packets are not reset here, they may destruct after the proxy
+ // and lead to a crash when trying to lock the proxy's destructed mutex.
+ for (auto& packet : capture.released_packets) {
+ packet.ResetAndReturnReleaseFn();
+ }
+}
+
+TEST(MultiSendTest, CanRepeatedlyReuseOneBuffer) {
+ constexpr size_t kMaxSends = ProxyHost::GetNumSimultaneousAclSendsSupported();
+ struct {
+ size_t sends_called = 0;
+ std::array<H4PacketWithH4, kMaxSends> released_packets;
+ } capture;
+
+ pw::Function<void(H4PacketWithHci && packet)>&& send_to_host_fn(
+ []([[maybe_unused]] H4PacketWithHci&& packet) {});
+ pw::Function<void(H4PacketWithH4 && packet)> send_to_controller_fn(
+ [&capture](H4PacketWithH4&& packet) {
+ // Capture first kMaxSends packets linearly.
+ if (capture.sends_called < capture.released_packets.size()) {
+ capture.released_packets[capture.sends_called] = std::move(packet);
+ } else {
+ // Reuse only first packet slot after kMaxSends.
+ capture.released_packets[0] = std::move(packet);
+ }
+ ++capture.sends_called;
+ });
+
+ ProxyHost proxy = ProxyHost(std::move(send_to_host_fn),
+ std::move(send_to_controller_fn),
+ 2 * kMaxSends);
+ SendReadBufferResponseFromController(proxy, 2 * kMaxSends);
+
+ std::array<uint8_t, 1> attribute_value = {0xF};
+ // Occupy all send buffers.
+ for (size_t i = 0; i < kMaxSends; ++i) {
+ EXPECT_TRUE(proxy.SendGattNotify(123, 345, pw::span(attribute_value)).ok());
+ }
+
+ // Repeatedly free and reoccupy first buffer.
+ for (size_t i = 0; i < kMaxSends; ++i) {
+ capture.released_packets[0].~H4PacketWithH4();
+ EXPECT_TRUE(proxy.SendGattNotify(123, 345, pw::span(attribute_value)).ok());
+ EXPECT_EQ(proxy.SendGattNotify(123, 345, pw::span(attribute_value)),
+ PW_STATUS_UNAVAILABLE);
+ }
+ EXPECT_EQ(capture.sends_called, 2 * kMaxSends);
+
+ // If captured packets are not reset here, they may destruct after the proxy
+ // and lead to a crash when trying to lock the proxy's destructed mutex.
+ for (auto& packet : capture.released_packets) {
+ packet.ResetAndReturnReleaseFn();
+ }
+}
+
+TEST(MultiSendTest, ResetClearsBuffOccupiedFlags) {
+ constexpr size_t kMaxSends = ProxyHost::GetNumSimultaneousAclSendsSupported();
+ struct {
+ size_t sends_called = 0;
+ std::array<H4PacketWithH4, 2 * kMaxSends> released_packets;
+ } capture;
+
+ pw::Function<void(H4PacketWithHci && packet)>&& send_to_host_fn(
+ []([[maybe_unused]] H4PacketWithHci&& packet) {});
+ pw::Function<void(H4PacketWithH4 && packet)> send_to_controller_fn(
+ [&capture](H4PacketWithH4&& packet) {
+ // Capture all packets to prevent their destruction.
+ capture.released_packets[capture.sends_called++] = std::move(packet);
+ });
+
+ ProxyHost proxy = ProxyHost(
+ std::move(send_to_host_fn), std::move(send_to_controller_fn), kMaxSends);
+ SendReadBufferResponseFromController(proxy, kMaxSends);
+
+ std::array<uint8_t, 1> attribute_value = {0xF};
+ // Occupy all send buffers.
+ for (size_t i = 0; i < kMaxSends; ++i) {
+ EXPECT_TRUE(proxy.SendGattNotify(123, 345, pw::span(attribute_value)).ok());
+ }
+
+ proxy.Reset();
+ SendReadBufferResponseFromController(proxy, kMaxSends);
+
+ // Although sent packets have not been released, proxy.Reset() should have
+ // marked all buffers as unoccupied.
+ for (size_t i = 0; i < kMaxSends; ++i) {
+ EXPECT_TRUE(proxy.SendGattNotify(123, 345, pw::span(attribute_value)).ok());
+ }
+ EXPECT_EQ(capture.sends_called, 2 * kMaxSends);
+
+ // If captured packets are not reset here, they may destruct after the proxy
+ // and lead to a crash when trying to lock the proxy's destructed mutex.
+ for (auto& packet : capture.released_packets) {
+ packet.ResetAndReturnReleaseFn();
+ }
+}
+
 } // namespace
 } // namespace pw::bluetooth::proxy

pw_bluetooth_proxy/public/pw_bluetooth_proxy/h4_packet.h

@@ -85,6 +85,7 @@ class H4PacketWithHci final : public H4PacketInterface {
 /// H4PacketWithHci is an H4Packet backed by an H4 buffer.
 class H4PacketWithH4 final : public H4PacketInterface {
  public:
+ H4PacketWithH4() = default;
  H4PacketWithH4(pw::span<uint8_t> h4_span) : h4_span_(h4_span) {}
 
  /// release_fn (if callable) will be called when H4PacketWithH4 is destructed.