[promises] Fork mpsc

Ahead of landing a new lock free implementation, fork the existing implementation and keep it around for a while for the production version of chaotic-good (we don't want to change that yet)

PiperOrigin-RevId: 762585546

Author: ctiller

build_autogenerated.yaml

@@ -1346,6 +1346,28 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "lock_based_mpsc",
+    hdrs = [
+        "lib/promise/lock_based_mpsc.h",
+    ],
+    external_deps = [
+        "absl/base:core_headers",
+        "absl/log:check",
+    ],
+    deps = [
+        "activity",
+        "dump_args",
+        "poll",
+        "ref_counted",
+        "status_flag",
+        "sync",
+        "wait_set",
+        "//:gpr",
+        "//:ref_counted_ptr",
+    ],
+)
+
 grpc_cc_library(
     name = "observable",
     hdrs = [
@@ -8874,9 +8896,9 @@ grpc_cc_library(
         "event_engine_context",
         "event_engine_tcp_socket_utils",
         "grpc_promise_endpoint",
+        "lock_based_mpsc",
         "loop",
         "match_promise",
-        "mpsc",
         "seq",
         "try_join",
         "try_seq",
@@ -8919,12 +8941,12 @@ grpc_cc_library(
         "grpc_promise_endpoint",
         "if",
         "inter_activity_pipe",
+        "lock_based_mpsc",
         "loop",
         "map",
         "memory_quota",
         "metadata_batch",
         "metrics",
-        "mpsc",
         "pipe",
         "poll",
         "resource_quota",
@@ -8980,11 +9002,11 @@ grpc_cc_library(
         "if",
         "inter_activity_latch",
         "inter_activity_pipe",
+        "lock_based_mpsc",
         "loop",
         "memory_quota",
         "metadata_batch",
         "metrics",
-        "mpsc",
         "pipe",
         "poll",
         "resource_quota",

src/core/BUILD

@@ -31,9 +31,9 @@
 #include "src/core/lib/debug/trace.h"
 #include "src/core/lib/event_engine/event_engine_context.h"
 #include "src/core/lib/event_engine/tcp_socket_utils.h"
+#include "src/core/lib/promise/lock_based_mpsc.h"
 #include "src/core/lib/promise/loop.h"
 #include "src/core/lib/promise/match_promise.h"
-#include "src/core/lib/promise/mpsc.h"
 #include "src/core/lib/promise/seq.h"
 #include "src/core/lib/promise/try_join.h"
 #include "src/core/lib/promise/try_seq.h"
@@ -154,7 +154,7 @@ class ChaoticGoodTransport : public RefCounted<ChaoticGoodTransport> {
   // Common outbound loop for both client and server (these vary only over the
   // frame type).
   template <typename Frame>
-  auto TransportWriteLoop(MpscReceiver<Frame>& outgoing_frames) {
+  auto TransportWriteLoop(LockBasedMpscReceiver<Frame>& outgoing_frames) {
     return Loop([self = Ref(), &outgoing_frames] {
       return TrySeq(
           // Get next outgoing frame.

src/core/ext/transport/chaotic_good_legacy/chaotic_good_transport.h

@@ -48,8 +48,8 @@
 #include "src/core/lib/promise/for_each.h"
 #include "src/core/lib/promise/if.h"
 #include "src/core/lib/promise/inter_activity_pipe.h"
+#include "src/core/lib/promise/lock_based_mpsc.h"
 #include "src/core/lib/promise/loop.h"
-#include "src/core/lib/promise/mpsc.h"
 #include "src/core/lib/promise/pipe.h"
 #include "src/core/lib/promise/poll.h"
 #include "src/core/lib/promise/try_join.h"
@@ -114,7 +114,7 @@ class ChaoticGoodClientTransport final : public ClientTransport {
   grpc_event_engine::experimental::MemoryAllocator allocator_;
   // Max buffer is set to 4, so that for stream writes each time it will queue
   // at most 2 frames.
-  MpscReceiver<ClientFrame> outgoing_frames_;
+  LockBasedMpscReceiver<ClientFrame> outgoing_frames_;
   Mutex mu_;
   uint32_t next_stream_id_ ABSL_GUARDED_BY(mu_) = 1;
   // Map of stream incoming server frames, key is stream_id.

src/core/ext/transport/chaotic_good_legacy/client_transport.h

@@ -132,7 +132,7 @@ auto BooleanSuccessToTransportErrorCapturingInitiator(CallInitiator initiator) {
 }  // namespace
 
 auto ChaoticGoodServerTransport::SendFrame(
-    ServerFrame frame, MpscSender<ServerFrame> outgoing_frames,
+    ServerFrame frame, LockBasedMpscSender<ServerFrame> outgoing_frames,
     CallInitiator call_initiator) {
   // Capture the call_initiator to ensure the underlying call spine is alive
   // until the outgoing_frames.Send promise completes.
@@ -142,7 +142,7 @@ auto ChaoticGoodServerTransport::SendFrame(
 }
 
 auto ChaoticGoodServerTransport::SendFrameAcked(
-    ServerFrame frame, MpscSender<ServerFrame> outgoing_frames,
+    ServerFrame frame, LockBasedMpscSender<ServerFrame> outgoing_frames,
     CallInitiator call_initiator) {
   // Capture the call_initiator to ensure the underlying call spine is alive
   // until the outgoing_frames.Send promise completes.
@@ -152,7 +152,7 @@ auto ChaoticGoodServerTransport::SendFrameAcked(
 }
 
 auto ChaoticGoodServerTransport::SendCallBody(
-    uint32_t stream_id, MpscSender<ServerFrame> outgoing_frames,
+    uint32_t stream_id, LockBasedMpscSender<ServerFrame> outgoing_frames,
     CallInitiator call_initiator) {
   // Continuously send client frame with client to server messages.
   return ForEach(MessagesFrom(call_initiator),
@@ -166,7 +166,7 @@ auto ChaoticGoodServerTransport::SendCallBody(
 }
 
 auto ChaoticGoodServerTransport::SendCallInitialMetadataAndBody(
-    uint32_t stream_id, MpscSender<ServerFrame> outgoing_frames,
+    uint32_t stream_id, LockBasedMpscSender<ServerFrame> outgoing_frames,
     CallInitiator call_initiator) {
   return TrySeq(
       // Wait for initial metadata then send it out.

src/core/ext/transport/chaotic_good_legacy/server_transport.cc

@@ -54,8 +54,8 @@
 #include "src/core/lib/promise/if.h"
 #include "src/core/lib/promise/inter_activity_latch.h"
 #include "src/core/lib/promise/inter_activity_pipe.h"
+#include "src/core/lib/promise/lock_based_mpsc.h"
 #include "src/core/lib/promise/loop.h"
-#include "src/core/lib/promise/mpsc.h"
 #include "src/core/lib/promise/party.h"
 #include "src/core/lib/promise/pipe.h"
 #include "src/core/lib/promise/poll.h"
@@ -108,10 +108,11 @@ class ChaoticGoodServerTransport final : public ServerTransport {
   absl::Status NewStream(uint32_t stream_id, CallInitiator call_initiator);
   RefCountedPtr<Stream> LookupStream(uint32_t stream_id);
   RefCountedPtr<Stream> ExtractStream(uint32_t stream_id);
-  auto SendCallInitialMetadataAndBody(uint32_t stream_id,
-                                      MpscSender<ServerFrame> outgoing_frames,
-                                      CallInitiator call_initiator);
-  auto SendCallBody(uint32_t stream_id, MpscSender<ServerFrame> outgoing_frames,
+  auto SendCallInitialMetadataAndBody(
+      uint32_t stream_id, LockBasedMpscSender<ServerFrame> outgoing_frames,
+      CallInitiator call_initiator);
+  auto SendCallBody(uint32_t stream_id,
+                    LockBasedMpscSender<ServerFrame> outgoing_frames,
                     CallInitiator call_initiator);
   auto CallOutboundLoop(uint32_t stream_id, CallInitiator call_initiator);
   auto OnTransportActivityDone(absl::string_view activity);
@@ -132,18 +133,19 @@ class ChaoticGoodServerTransport final : public ServerTransport {
   auto PushFrameIntoCall(RefCountedPtr<Stream> stream, ClientEndOfStream frame);
   auto PushFrameIntoCall(RefCountedPtr<Stream> stream, BeginMessageFrame frame);
   auto PushFrameIntoCall(RefCountedPtr<Stream> stream, MessageChunkFrame frame);
-  auto SendFrame(ServerFrame frame, MpscSender<ServerFrame> outgoing_frames,
+  auto SendFrame(ServerFrame frame,
+                 LockBasedMpscSender<ServerFrame> outgoing_frames,
                  CallInitiator call_initiator);
   auto SendFrameAcked(ServerFrame frame,
-                      MpscSender<ServerFrame> outgoing_frames,
+                      LockBasedMpscSender<ServerFrame> outgoing_frames,
                       CallInitiator call_initiator);
 
   RefCountedPtr<UnstartedCallDestination> call_destination_;
   const RefCountedPtr<CallArenaAllocator> call_arena_allocator_;
   const std::shared_ptr<grpc_event_engine::experimental::EventEngine>
       event_engine_;
   InterActivityLatch<void> got_acceptor_;
-  MpscReceiver<ServerFrame> outgoing_frames_;
+  LockBasedMpscReceiver<ServerFrame> outgoing_frames_;
   Mutex mu_;
   // Map of stream incoming server frames, key is stream_id.
   StreamMap stream_map_ ABSL_GUARDED_BY(mu_);