[PH2][Keepalive] Move spawn into the transport

PiperOrigin-RevId: 856971493

Author: ac-patel

src/core/ext/transport/chttp2/transport/http2_client_transport.cc

@@ -1504,7 +1504,7 @@ Http2ClientTransport::Http2ClientTransport(
       KeepAliveInterfaceImpl::Make(this),
       ((args.keepalive_timeout < args.ping_timeout) ? args.keepalive_timeout
                                                     : Duration::Infinity()),
-      args.keepalive_time, general_party_.get());
+      args.keepalive_time);
 
   GRPC_DCHECK(ping_manager_.has_value());
   GRPC_DCHECK(keepalive_manager_.has_value());
@@ -1514,6 +1514,7 @@ Http2ClientTransport::Http2ClientTransport(
 
 void Http2ClientTransport::SpawnTransportLoops() {
   GRPC_HTTP2_CLIENT_DLOG << "Http2ClientTransport::SpawnTransportLoops Begin";
+  MaybeSpawnKeepaliveLoop();
   SpawnGuardedTransportParty(
       "FlowControlPeriodicUpdateLoop",
       UntilTransportClosed(FlowControlPeriodicUpdateLoop()));
@@ -2252,6 +2253,15 @@ void Http2ClientTransport::MaybeSpawnDelayedPing(
   }
 }
 
+void Http2ClientTransport::MaybeSpawnKeepaliveLoop() {
+  if (keepalive_manager_->IsKeepAliveLoopNeeded()) {
+    SpawnGuardedTransportParty(
+        "KeepaliveLoop", [self = RefAsSubclass<Http2ClientTransport>()]() {
+          return self->keepalive_manager_->KeepaliveLoop();
+        });
+  }
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 // Class PingSystemInterfaceImpl
 

src/core/ext/transport/chttp2/transport/http2_client_transport.h

@@ -389,6 +389,7 @@ class Http2ClientTransport final : public ClientTransport,
   std::optional<KeepaliveManager> keepalive_manager_;
   void MaybeSpawnPingTimeout(std::optional<uint64_t> opaque_data);
   void MaybeSpawnDelayedPing(std::optional<Duration> delayed_ping_wait);
+  void MaybeSpawnKeepaliveLoop();
 
   // Flags
   bool keepalive_permit_without_calls_;

src/core/ext/transport/chttp2/transport/keepalive.cc

@@ -23,6 +23,7 @@
 #include "src/core/lib/promise/if.h"
 #include "src/core/lib/promise/loop.h"
 #include "src/core/lib/promise/party.h"
+#include "src/core/lib/promise/promise.h"
 #include "src/core/lib/promise/race.h"
 #include "src/core/lib/promise/sleep.h"
 #include "src/core/lib/promise/try_seq.h"
@@ -33,12 +34,10 @@ namespace grpc_core {
 namespace http2 {
 KeepaliveManager::KeepaliveManager(
     std::unique_ptr<KeepAliveInterface> keep_alive_interface,
-    Duration keepalive_timeout, const Duration keepalive_time, Party* party)
+    Duration keepalive_timeout, const Duration keepalive_time)
     : keep_alive_interface_(std::move(keep_alive_interface)),
       keepalive_timeout_(keepalive_timeout),
-      keepalive_time_(keepalive_time) {
-  MaybeSpawnKeepaliveLoop(party);
-}
+      keepalive_time_(keepalive_time) {}
 
 auto KeepaliveManager::WaitForKeepAliveTimeout() {
   return AssertResultType<absl::Status>(
@@ -90,23 +89,18 @@ auto KeepaliveManager::MaybeSendKeepAlivePing() {
              }));
 }
 
-void KeepaliveManager::MaybeSpawnKeepaliveLoop(Party* party) {
-  if (!IsKeepAliveNeeded()) {
-    GRPC_HTTP2_KEEPALIVE_LOG << "Not spawning keepalive loop.";
-    return;
-  }
-  keep_alive_spawned_ = true;
+bool KeepaliveManager::IsKeepAliveLoopNeeded() {
+  return IsKeepAliveNeeded() && !keep_alive_spawned_;
+}
 
-  party->Spawn("KeepAliveLoop", Loop([this]() {
-                 return TrySeq(
-                     Sleep(keepalive_time_),
-                     [this]() { return MaybeSendKeepAlivePing(); },
-                     []() -> LoopCtl<absl::Status> { return Continue(); });
-               }),
-               [](auto status) {
-                 GRPC_HTTP2_KEEPALIVE_LOG << "KeepAlive end with status: "
-                                          << status;
-               });
+Promise<absl::Status> KeepaliveManager::KeepaliveLoop() {
+  GRPC_HTTP2_KEEPALIVE_LOG << "KeepaliveManager::KeepaliveLoop Spawning.";
+  keep_alive_spawned_ = true;
+  return Loop([this]() {
+    return TrySeq(
+        Sleep(keepalive_time_), [this]() { return MaybeSendKeepAlivePing(); },
+        []() -> LoopCtl<absl::Status> { return Continue(); });
+  });
 }
 }  // namespace http2
 }  // namespace grpc_core

src/core/ext/transport/chttp2/transport/keepalive.h

@@ -18,8 +18,10 @@
 #ifndef GRPC_SRC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_KEEPALIVE_H
 #define GRPC_SRC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_KEEPALIVE_H
 
+#include "src/core/lib/promise/loop.h"
 #include "src/core/lib/promise/party.h"
 #include "src/core/lib/promise/promise.h"
+#include "src/core/lib/promise/sleep.h"
 #include "src/core/util/grpc_check.h"
 #include "absl/status/status.h"
 
@@ -45,8 +47,7 @@ class KeepAliveInterface {
 class KeepaliveManager {
  public:
   KeepaliveManager(std::unique_ptr<KeepAliveInterface> keep_alive_interface,
-                   Duration keepalive_timeout, Duration keepalive_time,
-                   Party* party);
+                   Duration keepalive_timeout, Duration keepalive_time);
 
   // Needs to be called when any data is read from the endpoint.
   void GotData() {
@@ -66,11 +67,13 @@ class KeepaliveManager {
     keepalive_timeout_ = keepalive_timeout;
   }
 
- private:
-  // Spawns the keepalive loop on the given party. This MUST be called at most
-  // once during the lifetime of the keepalive manager.
-  void MaybeSpawnKeepaliveLoop(Party* party);
+  bool IsKeepAliveLoopNeeded();
+
+  // Returns a promise that processes keepalive pings. This MUST be called at
+  // most once during the lifetime of the keepalive manager.
+  Promise<absl::Status> KeepaliveLoop();
 
+ private:
   // Returns a promise that sleeps for the keepalive_timeout_ and triggers the
   // keepalive timeout unless data is read within the keepalive timeout.
   auto WaitForKeepAliveTimeout();

test/core/transport/chttp2/keepalive_test.cc

@@ -88,6 +88,17 @@ class MockKeepAliveInterface : public KeepAliveInterface {
   }
 };
 
+void MaybeSpawnKeepaliveLoop(KeepaliveManager& keepalive_manager,
+                             Party* party) {
+  if (keepalive_manager.IsKeepAliveLoopNeeded()) {
+    party->Spawn(
+        "KeepaliveLoop", [&]() { return keepalive_manager.KeepaliveLoop(); },
+        [](absl::Status status) {
+          LOG(INFO) << "KeepaliveLoop end with status: " << status;
+        });
+  }
+}
+
 class KeepaliveManagerTest : public YodelTest {
  protected:
   using YodelTest::YodelTest;
@@ -122,17 +133,17 @@ YODEL_TEST(KeepaliveManagerTest, TestKeepAlive) {
                                                       /*return_value=*/true);
 
   KeepaliveManager keep_alive_system(std::move(keep_alive_interface),
-                                     keepalive_timeout, keepalive_interval,
-                                     GetParty());
+                                     keepalive_timeout, keepalive_interval);
+  MaybeSpawnKeepaliveLoop(keep_alive_system, GetParty());
 
   WaitForAllPendingWork();
   event_engine()->TickUntilIdle();
   event_engine()->UnsetGlobalHooks();
 }
 
 YODEL_TEST(KeepaliveManagerTest, TestKeepAliveTimeout) {
-  // Simple test to simulate sending a keepalive ping and not receiving any data
-  // within the keepalive timeout. The test asserts that:
+  // Simple test to simulate sending a keepalive ping and not receiving any
+  // data within the keepalive timeout. The test asserts that:
   // 1. The keepalive timeout is triggered.
   // 2. The keepalive ping is sent.
   InitParty();
@@ -148,8 +159,8 @@ YODEL_TEST(KeepaliveManagerTest, TestKeepAliveTimeout) {
                                                       /*return_value=*/true);
 
   KeepaliveManager keep_alive_system(std::move(keep_alive_interface),
-                                     keepalive_timeout, keepalive_interval,
-                                     GetParty());
+                                     keepalive_timeout, keepalive_interval);
+  MaybeSpawnKeepaliveLoop(keep_alive_system, GetParty());
 
   WaitForAllPendingWork();
   event_engine()->TickUntilIdle();
@@ -159,7 +170,8 @@ YODEL_TEST(KeepaliveManagerTest, TestKeepAliveTimeout) {
 YODEL_TEST(KeepaliveManagerTest, TestKeepAliveWithData) {
   // Test to simulate reading of data at certain intervals. The test asserts
   // that:
-  // 1. The keepalive ping is not sent as long as there is data read within the
+  // 1. The keepalive ping is not sent as long as there is data read within
+  // the
   //    keepalive interval.
   InitParty();
   int end_after = 1;
@@ -175,8 +187,8 @@ YODEL_TEST(KeepaliveManagerTest, TestKeepAliveWithData) {
                                                       /*return_value=*/true);
 
   KeepaliveManager keep_alive_system(std::move(keep_alive_interface),
-                                     keepalive_timeout, keepalive_interval,
-                                     GetParty());
+                                     keepalive_timeout, keepalive_interval);
+  MaybeSpawnKeepaliveLoop(keep_alive_system, GetParty());
 
   GetParty()->Spawn(
       "ReadData", Loop([&read_loop_end_after, &keep_alive_system]() {
@@ -200,7 +212,8 @@ YODEL_TEST(KeepaliveManagerTest, TestKeepAliveWithData) {
 YODEL_TEST(KeepaliveManagerTest, TestKeepAliveTimeoutWithData) {
   // Test to simulate reading of data at certain intervals. The test asserts
   // that:
-  // 1. The keepalive ping is not sent as long as there is data read within the
+  // 1. The keepalive ping is not sent as long as there is data read within
+  // the
   //    keepalive interval.
   // 2. Keepalive timeout is triggered once no data is read within the
   //    keepalive timeout.
@@ -219,8 +232,8 @@ YODEL_TEST(KeepaliveManagerTest, TestKeepAliveTimeoutWithData) {
                                                       /*return_value=*/true);
 
   KeepaliveManager keep_alive_system(std::move(keep_alive_interface),
-                                     keepalive_timeout, keepalive_interval,
-                                     GetParty());
+                                     keepalive_timeout, keepalive_interval);
+  MaybeSpawnKeepaliveLoop(keep_alive_system, GetParty());
 
   GetParty()->Spawn(
       "ReadData", Loop([&read_loop_end_after, &keep_alive_system]() {