pw_bluetooth_proxy: Add write flow control mechanism

Notify on L2capWriteChannel queue space available if a previous write
failed because of no space. Expose this for RfcommChannel and
optionally for L2capCoc and BasicL2capChannel's

Bug: 380299794
Change-Id: I413fade05375dcbd4ef7ca1865ac809288f2cefd
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/251435
Pigweed-Auto-Submit: Austin Foxley <afoxley@google.com>
Reviewed-by: David Rees <drees@google.com>
Docs-Not-Needed: Austin Foxley <afoxley@google.com>
Lint: Lint 🤖 <android-build-ayeaye@system.gserviceaccount.com>
Commit-Queue: Auto-Submit <auto-submit@pigweed-service-accounts.iam.gserviceaccount.com>

pw_bluetooth_proxy/basic_l2cap_channel.cc

@@ -27,8 +27,8 @@ pw::Result<BasicL2capChannel> BasicL2capChannel::Create(
     uint16_t connection_handle,
     uint16_t local_cid,
     uint16_t remote_cid,
-    pw::Function<void(pw::span<uint8_t> payload)>&&
-        payload_from_controller_fn) {
+    Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn,
+    Function<void()>&& queue_space_available_fn) {
   if (!AreValidParameters(/*connection_handle=*/connection_handle,
                           /*local_cid=*/local_cid,
                           /*remote_cid=*/remote_cid)) {
@@ -40,7 +40,8 @@ pw::Result<BasicL2capChannel> BasicL2capChannel::Create(
       /*connection_handle=*/connection_handle,
       /*local_cid=*/local_cid,
       /*remote_cid=*/remote_cid,
-      /*payload_from_controller_fn=*/std::move(payload_from_controller_fn));
+      /*payload_from_controller_fn=*/std::move(payload_from_controller_fn),
+      /*queue_space_available_fn=*/std::move(queue_space_available_fn));
 }
 
 pw::Status BasicL2capChannel::Write(pw::span<const uint8_t> payload) {
@@ -74,14 +75,16 @@ BasicL2capChannel::BasicL2capChannel(
     uint16_t connection_handle,
     uint16_t local_cid,
     uint16_t remote_cid,
-    pw::Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn)
+    Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn,
+    Function<void()>&& queue_space_available_fn)
     : L2capChannel(/*l2cap_channel_manager=*/l2cap_channel_manager,
                    /*connection_handle=*/connection_handle,
                    /*transport=*/AclTransportType::kLe,
                    /*local_cid=*/local_cid,
                    /*remote_cid=*/remote_cid,
                    /*payload_from_controller_fn=*/
-                   std::move(payload_from_controller_fn)) {}
+                   std::move(payload_from_controller_fn),
+                   std::move(queue_space_available_fn)) {}
 
 bool BasicL2capChannel::HandlePduFromController(pw::span<uint8_t> bframe) {
   Result<emboss::BFrameWriter> bframe_view =

pw_bluetooth_proxy/gatt_notify_channel.cc

@@ -91,7 +91,8 @@ GattNotifyChannel::GattNotifyChannel(L2capChannelManager& l2cap_channel_manager,
                    /*transport=*/AclTransportType::kLe,
                    /*local_cid=*/kAttributeProtocolCID,
                    /*remote_cid=*/kAttributeProtocolCID,
-                   /*payload_from_controller_fn=*/nullptr),
+                   /*payload_from_controller_fn=*/nullptr,
+                   /*queue_space_available_fn=*/nullptr),
       attribute_handle_(attribute_handle) {}
 
 }  // namespace pw::bluetooth::proxy

pw_bluetooth_proxy/l2cap_channel.cc

@@ -33,6 +33,7 @@ L2capChannel::L2capChannel(L2capChannel&& other)
       transport_(other.transport()),
       local_cid_(other.local_cid()),
       remote_cid_(other.remote_cid()),
+      queue_space_available_fn_(std::move(other.queue_space_available_fn_)),
       payload_from_controller_fn_(
           std::move(other.payload_from_controller_fn_)) {
   // All L2capChannels share a static mutex, so only one lock needs to be
@@ -41,6 +42,7 @@ L2capChannel::L2capChannel(L2capChannel&& other)
   // static, elide this operator or acquire a lock on both channels' mutexes.
   std::lock_guard lock(global_send_queue_mutex_);
   send_queue_ = std::move(other.send_queue_);
+  notify_on_dequeue_ = other.notify_on_dequeue_;
   l2cap_channel_manager_.ReleaseChannel(other);
   l2cap_channel_manager_.RegisterChannel(*this);
 }
@@ -55,12 +57,14 @@ L2capChannel& L2capChannel::operator=(L2capChannel&& other) {
   local_cid_ = other.local_cid();
   remote_cid_ = other.remote_cid();
   payload_from_controller_fn_ = std::move(other.payload_from_controller_fn_);
+  queue_space_available_fn_ = std::move(other.queue_space_available_fn_);
   // All L2capWriteChannels share a static mutex, so only one lock needs to be
   // acquired here.
   // TODO: https://pwbug.dev/369849508 - Once mutex is no longer static,
   // elide this operator or acquire a lock on both channels' mutexes.
   std::lock_guard lock(global_send_queue_mutex_);
   send_queue_ = std::move(other.send_queue_);
+  notify_on_dequeue_ = other.notify_on_dequeue_;
   l2cap_channel_manager_.ReleaseChannel(other);
   l2cap_channel_manager_.RegisterChannel(*this);
   return *this;
@@ -77,6 +81,7 @@ Status L2capChannel::QueuePacket(H4PacketWithH4&& packet) {
     std::lock_guard lock(global_send_queue_mutex_);
     if (send_queue_.full()) {
       status = Status::Unavailable();
+      notify_on_dequeue_ = true;
     } else {
       send_queue_.push(std::move(packet));
       status = OkStatus();
@@ -87,12 +92,22 @@ Status L2capChannel::QueuePacket(H4PacketWithH4&& packet) {
 }
 
 std::optional<H4PacketWithH4> L2capChannel::DequeuePacket() {
-  std::lock_guard lock(global_send_queue_mutex_);
-  if (send_queue_.empty()) {
-    return std::nullopt;
+  std::optional<H4PacketWithH4> packet;
+  bool should_notify = false;
+  {
+    std::lock_guard lock(global_send_queue_mutex_);
+    if (!send_queue_.empty()) {
+      packet.emplace(std::move(send_queue_.front()));
+      send_queue_.pop();
+      should_notify = notify_on_dequeue_;
+      notify_on_dequeue_ = false;
+    }
   }
-  H4PacketWithH4 packet = std::move(send_queue_.front());
-  send_queue_.pop();
+
+  if (queue_space_available_fn_ && should_notify) {
+    queue_space_available_fn_();
+  }
+
   return packet;
 }
 
@@ -108,12 +123,14 @@ L2capChannel::L2capChannel(
     AclTransportType transport,
     uint16_t local_cid,
     uint16_t remote_cid,
-    pw::Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn)
+    Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn,
+    Function<void()>&& queue_space_available_fn)
     : l2cap_channel_manager_(l2cap_channel_manager),
       connection_handle_(connection_handle),
       transport_(transport),
       local_cid_(local_cid),
       remote_cid_(remote_cid),
+      queue_space_available_fn_(std::move(queue_space_available_fn)),
       payload_from_controller_fn_(std::move(payload_from_controller_fn)) {
   l2cap_channel_manager_.RegisterChannel(*this);
 }

pw_bluetooth_proxy/l2cap_coc.cc

@@ -102,8 +102,9 @@ pw::Result<L2capCoc> L2capCoc::Create(
     uint16_t connection_handle,
     CocConfig rx_config,
     CocConfig tx_config,
-    pw::Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn,
-    pw::Function<void(Event event)>&& event_fn) {
+    Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn,
+    Function<void(Event event)>&& event_fn,
+    Function<void()>&& queue_space_available_fn) {
   if (!AreValidParameters(/*connection_handle=*/connection_handle,
                           /*local_cid=*/rx_config.cid,
                           /*remote_cid=*/tx_config.cid)) {
@@ -125,7 +126,8 @@ pw::Result<L2capCoc> L2capCoc::Create(
       /*rx_config=*/rx_config,
       /*tx_config=*/tx_config,
       /*payload_from_controller_fn=*/std::move(payload_from_controller_fn),
-      /*event_fn=*/std::move(event_fn));
+      /*event_fn=*/std::move(event_fn),
+      /*queue_space_available_fn=*/std::move(queue_space_available_fn));
 }
 
 bool L2capCoc::HandlePduFromController(pw::span<uint8_t> kframe) {
@@ -232,14 +234,16 @@ L2capCoc::L2capCoc(
     uint16_t connection_handle,
     CocConfig rx_config,
     CocConfig tx_config,
-    pw::Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn,
-    pw::Function<void(Event event)>&& event_fn)
+    Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn,
+    Function<void(Event event)>&& event_fn,
+    Function<void()>&& queue_space_available_fn)
     : L2capChannel(l2cap_channel_manager,
                    connection_handle,
                    AclTransportType::kLe,
                    rx_config.cid,
                    tx_config.cid,
-                   std::move(payload_from_controller_fn)),
+                   std::move(payload_from_controller_fn),
+                   std::move(queue_space_available_fn)),
       state_(CocState::kRunning),
       rx_mtu_(rx_config.mtu),
       rx_mps_(rx_config.mps),

pw_bluetooth_proxy/l2cap_signaling_channel.cc

@@ -36,7 +36,8 @@ L2capSignalingChannel::L2capSignalingChannel(
                         /*connection_handle=*/connection_handle,
                         /*local_cid=*/fixed_cid,
                         /*remote_cid=*/fixed_cid,
-                        /*payload_from_controller_fn=*/nullptr),
+                        /*payload_from_controller_fn=*/nullptr,
+                        /*queue_space_available_fn=*/nullptr),
       l2cap_channel_manager_(l2cap_channel_manager) {}
 
 L2capSignalingChannel& L2capSignalingChannel::operator=(

pw_bluetooth_proxy/proxy_host.cc

@@ -397,8 +397,9 @@ pw::Result<L2capCoc> ProxyHost::AcquireL2capCoc(
     uint16_t connection_handle,
     L2capCoc::CocConfig rx_config,
     L2capCoc::CocConfig tx_config,
-    pw::Function<void(pw::span<uint8_t> payload)>&& receive_fn,
-    pw::Function<void(L2capCoc::Event event)>&& event_fn,
+    Function<void(pw::span<uint8_t> payload)>&& receive_fn,
+    Function<void(L2capCoc::Event event)>&& event_fn,
+    Function<void()>&& queue_space_available_fn,
     uint16_t rx_additional_credits) {
   Status status = acl_data_channel_.CreateAclConnection(connection_handle,
                                                         AclTransportType::kLe);
@@ -423,16 +424,17 @@ pw::Result<L2capCoc> ProxyHost::AcquireL2capCoc(
                                   rx_config,
                                   tx_config,
                                   std::move(receive_fn),
-                                  std::move(event_fn));
+                                  std::move(event_fn),
+                                  std::move(queue_space_available_fn));
 }
 
 pw::Result<BasicL2capChannel> ProxyHost::AcquireBasicL2capChannel(
     uint16_t connection_handle,
     uint16_t local_cid,
     uint16_t remote_cid,
     AclTransportType transport,
-    pw::Function<void(pw::span<uint8_t> payload)>&&
-        payload_from_controller_fn) {
+    Function<void(pw::span<uint8_t> payload)>&& payload_from_controller_fn,
+    Function<void()>&& queue_space_available_fn) {
   Status status =
       acl_data_channel_.CreateAclConnection(connection_handle, transport);
   if (status.IsResourceExhausted()) {
@@ -444,7 +446,8 @@ pw::Result<BasicL2capChannel> ProxyHost::AcquireBasicL2capChannel(
       /*connection_handle=*/connection_handle,
       /*local_cid=*/local_cid,
       /*remote_cid=*/remote_cid,
-      /*payload_from_controller_fn=*/std::move(payload_from_controller_fn));
+      /*payload_from_controller_fn=*/std::move(payload_from_controller_fn),
+      /*queue_space_available_fn=*/std::move(queue_space_available_fn));
 }
 
 pw::Status ProxyHost::SendGattNotify(uint16_t connection_handle,
@@ -470,7 +473,8 @@ pw::Result<RfcommChannel> ProxyHost::AcquireRfcommChannel(
     RfcommChannel::Config rx_config,
     RfcommChannel::Config tx_config,
     uint8_t channel_number,
-    pw::Function<void(pw::span<uint8_t> payload)>&& receive_fn) {
+    Function<void(pw::span<uint8_t> payload)>&& receive_fn,
+    Function<void()>&& queue_space_available_fn) {
   Status status = acl_data_channel_.CreateAclConnection(
       connection_handle, AclTransportType::kBrEdr);
   if (status != OkStatus() && status != Status::AlreadyExists()) {
@@ -481,7 +485,8 @@ pw::Result<RfcommChannel> ProxyHost::AcquireRfcommChannel(
                                rx_config,
                                tx_config,
                                channel_number,
-                               std::move(receive_fn));
+                               std::move(receive_fn),
+                               std::move(queue_space_available_fn));
 }
 
 bool ProxyHost::HasSendLeAclCapability() const {