diff --git a/google/cloud/functions/integration_tests/basic_integration_test.cc b/google/cloud/functions/integration_tests/basic_integration_test.cc index 2b3253d6..2cfbcd30 100644 --- a/google/cloud/functions/integration_tests/basic_integration_test.cc +++ b/google/cloud/functions/integration_tests/basic_integration_test.cc @@ -20,14 +20,18 @@ #include #include #include +#include #include +#include #include 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", "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/sleep/1000"); + return std::chrono::duration_cast( + std::chrono::steady_clock::now() - start); + }; + std::vector> tasks(kTaskCount); + std::generate(tasks.begin(), tasks.end(), + [&] { return std::async(std::launch::async, task); }); + std::vector 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(); diff --git a/google/cloud/functions/integration_tests/cloud_event_conformance.cc b/google/cloud/functions/integration_tests/cloud_event_conformance.cc index 44adab92..ee9e9472 100644 --- a/google/cloud/functions/integration_tests/cloud_event_conformance.cc +++ b/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 +#include "google/cloud/functions/internal/framework_impl.h" +#include // NOLINT(modernize-deprecated-headers) #include +#include #include namespace functions = ::google::cloud::functions; + +std::atomic 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*/) {}); } diff --git a/google/cloud/functions/integration_tests/cloud_event_handler.cc b/google/cloud/functions/integration_tests/cloud_event_handler.cc index e67948d6..9d0a18ab 100644 --- a/google/cloud/functions/integration_tests/cloud_event_handler.cc +++ b/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 +#include "google/cloud/functions/internal/framework_impl.h" +#include #include -namespace functions = google::cloud::functions; using ::google::cloud::functions::CloudEvent; +std::atomic 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*/) {}); } diff --git a/google/cloud/functions/integration_tests/cloud_event_integration_test.cc b/google/cloud/functions/integration_tests/cloud_event_integration_test.cc index a0e6041f..5d7abdc3 100644 --- a/google/cloud/functions/integration_tests/cloud_event_integration_test.cc +++ b/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(); diff --git a/google/cloud/functions/integration_tests/echo_server.cc b/google/cloud/functions/integration_tests/echo_server.cc index f48fd727..09389232 100644 --- a/google/cloud/functions/integration_tests/echo_server.cc +++ b/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 +#include "google/cloud/functions/internal/framework_impl.h" +#include #include #include #include @@ -22,9 +22,16 @@ namespace functions = ::google::cloud::functions; using functions::HttpRequest; using functions::HttpResponse; +std::atomic 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*/) {}); } diff --git a/google/cloud/functions/integration_tests/http_conformance.cc b/google/cloud/functions/integration_tests/http_conformance.cc index 84d9cdd6..2694d53e 100644 --- a/google/cloud/functions/integration_tests/http_conformance.cc +++ b/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 #include namespace functions = ::google::cloud::functions; + +std::atomic 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*/) {}); } diff --git a/google/cloud/functions/internal/framework_impl.cc b/google/cloud/functions/internal/framework_impl.cc index 2855d5a9..c7ceaeaf 100644 --- a/google/cloud/functions/internal/framework_impl.cc +++ b/google/cloud/functions/internal/framework_impl.cc @@ -75,9 +75,9 @@ int RunForTestImpl(int argc, char const* const argv[], UserFunction&& function, auto address = asio::ip::make_address(vm["address"].as()); auto port = vm["port"].as(); - // 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(port)}}; + acceptor.listen(boost::asio::socket_base::max_connections); actual_port(acceptor.local_endpoint().port()); auto handle_session = @@ -85,11 +85,41 @@ int RunForTestImpl(int argc, char const* const argv[], UserFunction&& function, HandleSession(std::move(socket), h); }; + auto cleanup = [](std::vector> sessions, auto wait) { + std::vector> 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> 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; }