fix: actually handle sessions in parallel

google/cloud/functions/integration_tests/basic_integration_test.cc

@@ -20,14 +20,18 @@
 #include <boost/process.hpp>
 #include <gmock/gmock.h>
 #include <nlohmann/json.hpp>
+#include <algorithm>
 #include <chrono>
+#include <future>
 #include <string>
 
 namespace google::cloud::functions_internal {
 inline namespace FUNCTIONS_FRAMEWORK_CPP_NS {
 namespace {
 
+using ::testing::Each;
 using ::testing::HasSubstr;
+
 namespace bp = boost::process;
 // Even with C++17, we have to use the Boost version because Boost.Process
 // expects it.
@@ -120,6 +124,7 @@ TEST(RunIntegrationTest, Basic) {
 
   try {
     (void)HttpGet("localhost", "8010", "/quit/program/0");
+    (void)HttpGet("localhost", "8010", "/quit/program/0");
   } catch (...) {
   }
   server.wait();
@@ -148,6 +153,7 @@ TEST(RunIntegrationTest, ExceptionLogsToStderr) {
 
   try {
     (void)HttpGet("localhost", "8010", "/quit/program/0");
+    (void)HttpGet("localhost", "8010", "/quit/program/0");
   } catch (...) {
   }
   server.wait();
@@ -179,6 +185,7 @@ TEST(RunIntegrationTest, OutputIsFlushed) {
 
   try {
     (void)HttpGet("localhost", "8010", "/quit/program/0");
+    (void)HttpGet("localhost", "8010", "/quit/program/0");
   } catch (...) {
   }
   server.wait();
@@ -192,6 +199,37 @@ TEST(RunIntegrationTest, ConformanceSmokeTest) {
 
   try {
     (void)HttpGet("localhost", "8010", "/quit/program/0");
+    (void)HttpGet("localhost", "8010", "/quit/program/0");
+  } catch (...) {
+  }
+  server.wait();
+  EXPECT_EQ(server.exit_code(), 0);
+}
+
+TEST(RunIntegrationTest, SomeParallelism) {
+  auto server = bp::child(ExePath(kServer), "--port=8010");
+  auto result = WaitForServerReady("localhost", "8010");
+  ASSERT_EQ(result, 0);
+
+  auto constexpr kTaskCount = 16;
+  auto task = [] {
+    auto start = std::chrono::steady_clock::now();
+    (void)HttpGet("localhost", "8010", "/sleep/1000");
+    return std::chrono::duration_cast<std::chrono::milliseconds>(
+        std::chrono::steady_clock::now() - start);
+  };
+  std::vector<std::future<std::chrono::milliseconds>> tasks(kTaskCount);
+  std::generate(tasks.begin(), tasks.end(),
+                [&] { return std::async(std::launch::async, task); });
+  std::vector<std::chrono::milliseconds> elapsed(tasks.size());
+  std::transform(tasks.begin(), tasks.end(), elapsed.begin(),
+                 [](auto& f) { return f.get(); });
+  // We ask for a 1s delay, but we tolerate up to 5.
+  EXPECT_THAT(elapsed, Each(::testing::Le(std::chrono::seconds(5))));
+
+  try {
+    (void)HttpGet("localhost", "8010", "/quit/program/0");
+    (void)HttpGet("localhost", "8010", "/quit/program/0");
   } catch (...) {
   }
   server.wait();

google/cloud/functions/integration_tests/cloud_event_conformance.cc

@@ -12,12 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "google/cloud/functions/framework.h"
-#include <absl/time/time.h>
+#include "google/cloud/functions/internal/framework_impl.h"
+#include <absl/time/time.h>  // NOLINT(modernize-deprecated-headers)
 #include <nlohmann/json.hpp>
+#include <atomic>
 #include <fstream>
 
 namespace functions = ::google::cloud::functions;
+
+std::atomic<bool> shutdown_server{false};
+
 void CloudEventConformance(functions::CloudEvent const& ev) {
   // This output name is hard requirement from the conformance testing
   // framework.
@@ -54,9 +58,11 @@ void CloudEventConformance(functions::CloudEvent const& ev) {
   std::ofstream(kOutputFilename) << event.dump() << "\n";
 
   // This is here just to gracefully shutdown and collect coverage data.
-  if (ev.subject() == "/quit/program/0") std::exit(0);
+  if (ev.subject() == "/quit/program/0") shutdown_server.store(true);
 }
 
 int main(int argc, char* argv[]) {
-  return functions::Run(argc, argv, CloudEventConformance);
+  return google::cloud::functions_internal::RunForTest(
+      argc, argv, CloudEventConformance, [] { return shutdown_server.load(); },
+      [](int /*port*/) {});
 }

google/cloud/functions/integration_tests/cloud_event_handler.cc

@@ -12,16 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "google/cloud/functions/framework.h"
-#include <cstdlib>
+#include "google/cloud/functions/internal/framework_impl.h"
+#include <atomic>
 #include <iostream>
 
-namespace functions = google::cloud::functions;
 using ::google::cloud::functions::CloudEvent;
 
+std::atomic<bool> shutdown_server{false};
+
 void CloudEventHandler(CloudEvent const& event) {
   auto const& subject = event.subject().value_or("");
-  if (subject == "/quit/program/0") std::exit(0);
+  if (subject == "/quit/program/0") {
+    shutdown_server.store(true);
+    return;
+  }
   if (subject.rfind("/exception/", 0) == 0) throw std::runtime_error(subject);
   if (subject.rfind("/unknown-exception/", 0) == 0) throw "uh-oh";
   if (subject.rfind("/buffered-stdout/", 0) == 0) {
@@ -33,5 +37,7 @@ void CloudEventHandler(CloudEvent const& event) {
 }
 
 int main(int argc, char* argv[]) {
-  return functions::Run(argc, argv, CloudEventHandler);
+  return google::cloud::functions_internal::RunForTest(
+      argc, argv, CloudEventHandler, [] { return shutdown_server.load(); },
+      [](int /*port*/) {});
 }

google/cloud/functions/integration_tests/cloud_event_integration_test.cc

@@ -200,6 +200,7 @@ TEST(RunIntegrationTest, Basic) {
 
   try {
     (void)HttpPost("localhost", "8020", "/quit/program/0");
+    (void)HttpPost("localhost", "8020", "/quit/program/0");
   } catch (...) {
   }
   server.wait();
@@ -215,6 +216,7 @@ TEST(RunIntegrationTest, Batch) {
   EXPECT_EQ(actual.result_int(), functions::HttpResponse::kOkay);
   try {
     (void)HttpPost("localhost", "8020", "/quit/program/0");
+    (void)HttpPost("localhost", "8020", "/quit/program/0");
   } catch (...) {
   }
   server.wait();
@@ -230,6 +232,7 @@ TEST(RunIntegrationTest, Binary) {
   EXPECT_EQ(actual.result_int(), functions::HttpResponse::kOkay);
   try {
     (void)HttpPost("localhost", "8020", "/quit/program/0");
+    (void)HttpPost("localhost", "8020", "/quit/program/0");
   } catch (...) {
   }
   server.wait();
@@ -258,6 +261,7 @@ TEST(RunIntegrationTest, ExceptionLogsToStderr) {
 
   try {
     (void)HttpPost("localhost", "8020", "/quit/program/0");
+    (void)HttpPost("localhost", "8020", "/quit/program/0");
   } catch (...) {
   }
   server.wait();
@@ -289,6 +293,7 @@ TEST(RunIntegrationTest, OutputIsFlushed) {
 
   try {
     (void)HttpPost("localhost", "8020", "/quit/program/0");
+    (void)HttpPost("localhost", "8020", "/quit/program/0");
   } catch (...) {
   }
   server.wait();
@@ -302,6 +307,7 @@ TEST(RunIntegrationTest, ConformanceSmokeTest) {
 
   try {
     (void)HttpPost("localhost", "8020", "/quit/program/0");
+    (void)HttpPost("localhost", "8020", "/quit/program/0");
   } catch (...) {
   }
   server.wait();

google/cloud/functions/integration_tests/echo_server.cc

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "google/cloud/functions/framework.h"
-#include <cstdlib>
+#include "google/cloud/functions/internal/framework_impl.h"
+#include <atomic>
 #include <cstring>
 #include <iostream>
 #include <sstream>
@@ -22,9 +22,16 @@ namespace functions = ::google::cloud::functions;
 using functions::HttpRequest;
 using functions::HttpResponse;
 
+std::atomic<bool> shutdown_server{false};
+
 HttpResponse EchoServer(HttpRequest const& request) {
   auto const& target = request.target();
-  if (target == "/quit/program/0") std::exit(0);
+  if (target == "/quit/program/0") {
+    shutdown_server = true;
+    return HttpResponse{}
+        .set_header("Content-Type", "text/plain")
+        .set_payload("OK");
+  }
   if (target.rfind("/exception/", 0) == 0) throw std::runtime_error(target);
   if (target.rfind("/unknown-exception/", 0) == 0) throw "uh-oh";
 
@@ -63,5 +70,7 @@ HttpResponse EchoServer(HttpRequest const& request) {
 }
 
 int main(int argc, char* argv[]) {
-  return functions::Run(argc, argv, EchoServer);
+  return google::cloud::functions_internal::RunForTest(
+      argc, argv, EchoServer, [] { return shutdown_server.load(); },
+      [](int /*port*/) {});
 }

google/cloud/functions/integration_tests/http_conformance.cc

@@ -12,18 +12,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "google/cloud/functions/framework.h"
+#include "google/cloud/functions/internal/framework_impl.h"
+#include <atomic>
 #include <fstream>
 
 namespace functions = ::google::cloud::functions;
+
+std::atomic<bool> shutdown_server{false};
+
 functions::HttpResponse HttpConformance(functions::HttpRequest const& request) {
   auto constexpr kOutputFilename = "function_output.json";
   std::ofstream(kOutputFilename) << request.payload() << "\n";
   // This is here just to gracefully shutdown and collect coverage data.
-  if (request.target() == "/quit/program/0") std::exit(0);
+  if (request.target() == "/quit/program/0") shutdown_server = true;
   return functions::HttpResponse{};
 }
 
 int main(int argc, char* argv[]) {
-  return functions::Run(argc, argv, HttpConformance);
+  return google::cloud::functions_internal::RunForTest(
+      argc, argv, HttpConformance, [] { return shutdown_server.load(); },
+      [](int /*port*/) {});
 }

google/cloud/functions/internal/framework_impl.cc

@@ -75,21 +75,51 @@ int RunForTestImpl(int argc, char const* const argv[], UserFunction&& function,
   auto address = asio::ip::make_address(vm["address"].as<std::string>());
   auto port = vm["port"].as<int>();
 
-  // TODO(#35) - maybe replace with Boost.Log
-  asio::io_context ioc{/*concurrency_hint=*/1};
+  asio::io_context ioc{1};
   tcp::acceptor acceptor{ioc, {address, static_cast<std::uint16_t>(port)}};
+  acceptor.listen(boost::asio::socket_base::max_connections);
   actual_port(acceptor.local_endpoint().port());
 
   auto handle_session =
       [h = std::forward<UserFunction>(function)](tcp::socket socket) {
         HandleSession(std::move(socket), h);
       };
 
+  auto cleanup = [](std::vector<std::future<void>> sessions, auto wait) {
+    std::vector<std::future<void>> running;
+    for (auto& s : sessions) {
+      if (s.wait_for(wait) == std::future_status::timeout) {
+        running.push_back(std::move(s));
+        continue;
+      }
+      s.get();
+    }
+    return running;
+  };
+
+  // Reaching this number of sessions triggers a cleanup of any sessions that
+  // may have finished already. We allow up to 80 sessions to start without
+  // blocking, this should be enough for Cloud Run and Cloud Functions:
+  //     https://cloud.google.com/run/docs/about-concurrency#concurrency_values
+  auto constexpr kSessionCleanupThreshold = 80;
+
+  // If we reach this number of sessions we block until we are below the cleanup
+  // threshold.
+  auto constexpr kMaximumSessions = 160;
+
+  std::vector<std::future<void>> sessions;
   while (!shutdown()) {
+    if (sessions.size() >= kSessionCleanupThreshold) {
+      sessions = cleanup(std::move(sessions), std::chrono::milliseconds(0));
+    }
+    while (sessions.size() >= kMaximumSessions) {
+      sessions = cleanup(std::move(sessions), std::chrono::seconds(1));
+    }
     auto socket = acceptor.accept(ioc);
     if (!socket.is_open()) break;
     // Run a thread per-session, transferring ownership of the socket
-    (void)std::async(std::launch::async, handle_session, std::move(socket));
+    sessions.push_back(
+        std::async(std::launch::async, handle_session, std::move(socket)));
   }
   return 0;
 }