fix

ctiller · Jan 4, 2024 · cfe70da · cfe70da
1 parent 0492ba8
commit cfe70da
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 20 deletions.
diff --git a/src/core/ext/transport/chaotic_good/client_transport.cc b/src/core/ext/transport/chaotic_good/client_transport.cc
@@ -200,9 +200,7 @@ ChaoticGoodClientTransport::~ChaoticGoodClientTransport() {
 void ChaoticGoodClientTransport::AbortWithError() {
   // Mark transport as unavailable when the endpoint write/read failed.
   // Close all the available pipes.
-  if (!outgoing_frames_.IsClosed()) {
-    outgoing_frames_.MarkClosed();
-  }
+  outgoing_frames_.MarkClosed();
   ReleasableMutexLock lock(&mu_);
   StreamMap stream_map = std::move(stream_map_);
   stream_map_.clear();

diff --git a/src/core/ext/transport/chaotic_good/server_transport.cc b/src/core/ext/transport/chaotic_good/server_transport.cc
@@ -296,9 +296,7 @@ ChaoticGoodServerTransport::~ChaoticGoodServerTransport() {
 void ChaoticGoodServerTransport::AbortWithError() {
   // Mark transport as unavailable when the endpoint write/read failed.
   // Close all the available pipes.
-  if (!outgoing_frames_.IsClosed()) {
-    outgoing_frames_.MarkClosed();
-  }
+  outgoing_frames_.MarkClosed();
   ReleasableMutexLock lock(&mu_);
   StreamMap stream_map = std::move(stream_map_);
   stream_map_.clear();

diff --git a/src/core/lib/promise/mpsc.h b/src/core/lib/promise/mpsc.h
@@ -103,14 +103,12 @@ class Center : public RefCounted<Center<T>> {
 
   // Mark that the receiver is closed.
   void ReceiverClosed() {
-    MutexLock lock(&mu_);
+    ReleasableMutexLock lock(&mu_);
+    if (receiver_closed_) return;
     receiver_closed_ = true;
-  }
-
-  // Return whether the receiver is closed.
-  bool IsClosed() {
-    MutexLock lock(&mu_);
-    return receiver_closed_;
+    auto wakeups = send_wakers_.TakeWakeupSet();
+    lock.Release();
+    wakeups.Wakeup();
   }
 
  private:
@@ -173,7 +171,6 @@ class MpscReceiver {
   ~MpscReceiver() {
     if (center_ != nullptr) center_->ReceiverClosed();
   }
-  bool IsClosed() { return center_->IsClosed(); }
   void MarkClosed() {
     if (center_ != nullptr) center_->ReceiverClosed();
   }

diff --git a/test/core/transport/chaotic_good/client_transport_error_test.cc b/test/core/transport/chaotic_good/client_transport_error_test.cc
@@ -277,13 +277,13 @@ TEST_F(ClientTransportTest, AddMultipleStreamWithWriteFailed) {
   auto call2 =
       MakeCall(event_engine().get(), Arena::Create(8192, memory_allocator()));
   transport->StartCall(std::move(call2.handler));
-  call1.initiator.SpawnGuarded("test-send", [initiator =
-                                                 call1.initiator]() mutable {
+  call1.initiator.SpawnGuarded("test-send-1", [initiator =
+                                                   call1.initiator]() mutable {
     return TrySeq(initiator.PushClientInitialMetadata(TestInitialMetadata()),
                   SendClientToServerMessages(initiator, 1));
   });
-  call2.initiator.SpawnGuarded("test-send", [initiator =
-                                                 call2.initiator]() mutable {
+  call2.initiator.SpawnGuarded("test-send-2", [initiator =
+                                                   call2.initiator]() mutable {
     return TrySeq(initiator.PushClientInitialMetadata(TestInitialMetadata()),
                   SendClientToServerMessages(initiator, 1));
   });
@@ -292,7 +292,7 @@ TEST_F(ClientTransportTest, AddMultipleStreamWithWriteFailed) {
   StrictMock<MockFunction<void()>> on_done2;
   EXPECT_CALL(on_done2, Call());
   call1.initiator.SpawnInfallible(
-      "test-read", [&on_done1, initiator = call1.initiator]() mutable {
+      "test-read-1", [&on_done1, initiator = call1.initiator]() mutable {
         return Seq(
             initiator.PullServerInitialMetadata(),
             [](ValueOrFailure<ServerMetadataHandle> md) {
@@ -308,7 +308,7 @@ TEST_F(ClientTransportTest, AddMultipleStreamWithWriteFailed) {
             });
       });
   call2.initiator.SpawnInfallible(
-      "test-read", [&on_done2, initiator = call2.initiator]() mutable {
+      "test-read-2", [&on_done2, initiator = call2.initiator]() mutable {
         return Seq(
             initiator.PullServerInitialMetadata(),
             [](ValueOrFailure<ServerMetadataHandle> md) {