feat: use thread-local curl handle for multi-concurrency support

include/aws/http/response.h

@@ -34,7 +34,7 @@ class response {
     inline void append_body(const char* p, size_t sz);
     inline bool has_header(char const* header) const;
     inline lambda_runtime::outcome<std::string, bool> get_header(char const* header) const;
-    inline response_code get_response_code() const { return m_response_code; }
+    response_code get_response_code() const { return m_response_code; }
     inline void set_response_code(aws::http::response_code c);
     inline void set_content_type(char const* ct);
     inline std::string const& get_body() const;

include/aws/lambda-runtime/runtime.h

@@ -167,14 +167,13 @@ class runtime {
 
 private:
     void set_curl_next_options();
-    void set_curl_post_result_options();
+    static void set_curl_post_result_options();
     post_outcome do_post(
         std::string const& url,
         std::string const& request_id,
         invocation_response const& handler_response);
     std::string const m_user_agent_header;
     std::array<std::string const, 3> const m_endpoints;
-    CURL* const m_curl_handle;
 };
 
 inline std::chrono::milliseconds invocation_request::get_time_remaining() const

include/aws/logging/logging.h

@@ -15,11 +15,12 @@
  */
 
 #include <cstdarg>
+#include <cstdint>
 
 namespace aws {
 namespace logging {
 
-enum class verbosity {
+enum class verbosity : std::uint8_t {
     error,
     info,
     debug,

src/logging.cpp

@@ -14,15 +14,17 @@
  */
 #include "aws/logging/logging.h"
 #include <array>
+#include <cstdarg>
 #include <cstdio>
 #include <chrono>
 
 #define LAMBDA_RUNTIME_API __attribute__((visibility("default")))
 
 namespace aws {
 namespace logging {
+namespace {
 
-static inline char const* get_prefix(verbosity v)
+inline char const* get_prefix(verbosity v)
 {
     switch (v) {
         case verbosity::error:
@@ -36,6 +38,8 @@ static inline char const* get_prefix(verbosity v)
     }
 }
 
+} // namespace
+
 LAMBDA_RUNTIME_API
 void log(verbosity v, char const* tag, char const* msg, va_list args)
 {

src/runtime.cpp

@@ -42,6 +42,7 @@ static constexpr auto COGNITO_IDENTITY_HEADER = "lambda-runtime-cognito-identity
 static constexpr auto DEADLINE_MS_HEADER = "lambda-runtime-deadline-ms";
 static constexpr auto FUNCTION_ARN_HEADER = "lambda-runtime-invoked-function-arn";
 static constexpr auto TENANT_ID_HEADER = "lambda-runtime-aws-tenant-id";
+thread_local static CURL* m_curl_handle = curl_easy_init();
 
 enum Endpoints {
     INIT,
@@ -155,7 +156,7 @@ static int rt_curl_debug_callback(CURL* handle, curl_infotype type, char* data,
     (void)handle;
     (void)type;
     (void)userdata;
-    std::string s(data, size);
+    const std::string s(data, size);
     logging::log_debug(LOG_TAG, "CURL DBG: %s", s.c_str());
     return 0;
 }
@@ -164,83 +165,78 @@ static int rt_curl_debug_callback(CURL* handle, curl_infotype type, char* data,
 runtime::runtime(std::string const& endpoint) : runtime(endpoint, "AWS_Lambda_Cpp/" + std::string(get_version())) {}
 
 runtime::runtime(std::string const& endpoint, std::string const& user_agent)
-    : m_user_agent_header("User-Agent: " + user_agent),
-      m_endpoints{
-          {endpoint + "/2018-06-01/runtime/init/error",
-           endpoint + "/2018-06-01/runtime/invocation/next",
-           endpoint + "/2018-06-01/runtime/invocation/"}},
-      m_curl_handle(curl_easy_init())
+    : m_user_agent_header("User-Agent: " + user_agent), m_endpoints{
+                                                            {endpoint + "/2018-06-01/runtime/init/error",
+                                                             endpoint + "/2018-06-01/runtime/invocation/next",
+                                                             endpoint + "/2018-06-01/runtime/invocation/"}}
 {
-    if (!m_curl_handle) {
+    if (!lambda_runtime::m_curl_handle) {
         logging::log_error(LOG_TAG, "Failed to acquire curl easy handle for next.");
     }
 }
 
-runtime::~runtime()
-{
-    curl_easy_cleanup(m_curl_handle);
-}
+runtime::~runtime() = default;
 
 void runtime::set_curl_next_options()
 {
     // lambda freezes the container when no further tasks are available. The freezing period could be longer than the
     // request timeout, which causes the following get_next request to fail with a timeout error.
-    curl_easy_reset(m_curl_handle);
-    curl_easy_setopt(m_curl_handle, CURLOPT_TIMEOUT, 0L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_CONNECTTIMEOUT, 1L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_NOSIGNAL, 1L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_TCP_NODELAY, 1L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
+    curl_easy_reset(lambda_runtime::m_curl_handle);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_TIMEOUT, 0L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_CONNECTTIMEOUT, 1L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_NOSIGNAL, 1L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_TCP_NODELAY, 1L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
 
-    curl_easy_setopt(m_curl_handle, CURLOPT_HTTPGET, 1L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_URL, m_endpoints[Endpoints::NEXT].c_str());
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HTTPGET, 1L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_URL, m_endpoints[Endpoints::NEXT].c_str());
 
-    curl_easy_setopt(m_curl_handle, CURLOPT_WRITEFUNCTION, write_data);
-    curl_easy_setopt(m_curl_handle, CURLOPT_HEADERFUNCTION, write_header);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_WRITEFUNCTION, write_data);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HEADERFUNCTION, write_header);
 
-    curl_easy_setopt(m_curl_handle, CURLOPT_PROXY, "");
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_PROXY, "");
 
 #ifndef NDEBUG
-    curl_easy_setopt(m_curl_handle, CURLOPT_VERBOSE, 1);
-    curl_easy_setopt(m_curl_handle, CURLOPT_DEBUGFUNCTION, rt_curl_debug_callback);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_VERBOSE, 1);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_DEBUGFUNCTION, rt_curl_debug_callback);
 #endif
 }
 
 void runtime::set_curl_post_result_options()
 {
-    curl_easy_reset(m_curl_handle);
-    curl_easy_setopt(m_curl_handle, CURLOPT_TIMEOUT, 0L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_CONNECTTIMEOUT, 1L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_NOSIGNAL, 1L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_TCP_NODELAY, 1L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
+    curl_easy_reset(lambda_runtime::m_curl_handle);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_TIMEOUT, 0L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_CONNECTTIMEOUT, 1L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_NOSIGNAL, 1L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_TCP_NODELAY, 1L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
 
-    curl_easy_setopt(m_curl_handle, CURLOPT_POST, 1L);
-    curl_easy_setopt(m_curl_handle, CURLOPT_READFUNCTION, read_data);
-    curl_easy_setopt(m_curl_handle, CURLOPT_WRITEFUNCTION, write_data);
-    curl_easy_setopt(m_curl_handle, CURLOPT_HEADERFUNCTION, write_header);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_POST, 1L);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_READFUNCTION, read_data);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_WRITEFUNCTION, write_data);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HEADERFUNCTION, write_header);
 
-    curl_easy_setopt(m_curl_handle, CURLOPT_PROXY, "");
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_PROXY, "");
 
 #ifndef NDEBUG
-    curl_easy_setopt(m_curl_handle, CURLOPT_VERBOSE, 1);
-    curl_easy_setopt(m_curl_handle, CURLOPT_DEBUGFUNCTION, rt_curl_debug_callback);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_VERBOSE, 1);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_DEBUGFUNCTION, rt_curl_debug_callback);
 #endif
 }
 
 runtime::next_outcome runtime::get_next()
 {
     http::response resp;
     set_curl_next_options();
-    curl_easy_setopt(m_curl_handle, CURLOPT_WRITEDATA, &resp);
-    curl_easy_setopt(m_curl_handle, CURLOPT_HEADERDATA, &resp);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_WRITEDATA, &resp);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HEADERDATA, &resp);
 
     curl_slist* headers = nullptr;
     headers = curl_slist_append(headers, m_user_agent_header.c_str());
-    curl_easy_setopt(m_curl_handle, CURLOPT_HTTPHEADER, headers);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HTTPHEADER, headers);
 
     logging::log_debug(LOG_TAG, "Making request to %s", m_endpoints[Endpoints::NEXT].c_str());
-    CURLcode curl_code = curl_easy_perform(m_curl_handle);
+    const CURLcode curl_code = curl_easy_perform(lambda_runtime::m_curl_handle);
     logging::log_debug(LOG_TAG, "Completed request to %s", m_endpoints[Endpoints::NEXT].c_str());
     curl_slist_free_all(headers);
 
@@ -255,13 +251,13 @@ runtime::next_outcome runtime::get_next()
 
     {
         long resp_code;
-        curl_easy_getinfo(m_curl_handle, CURLINFO_RESPONSE_CODE, &resp_code);
+        curl_easy_getinfo(lambda_runtime::m_curl_handle, CURLINFO_RESPONSE_CODE, &resp_code);
         resp.set_response_code(static_cast<aws::http::response_code>(resp_code));
     }
 
     {
         char* content_type = nullptr;
-        curl_easy_getinfo(m_curl_handle, CURLINFO_CONTENT_TYPE, &content_type);
+        curl_easy_getinfo(lambda_runtime::m_curl_handle, CURLINFO_CONTENT_TYPE, &content_type);
         resp.set_content_type(content_type);
     }
 
@@ -312,7 +308,7 @@ runtime::next_outcome runtime::get_next()
     if (out.is_success()) {
         auto const& deadline_string = std::move(out).get_result();
         constexpr int base = 10;
-        unsigned long ms = strtoul(deadline_string.c_str(), nullptr, base);
+        const unsigned long ms = strtoul(deadline_string.c_str(), nullptr, base);
         assert(ms > 0);
         assert(ms < ULONG_MAX);
         req.deadline += std::chrono::milliseconds(ms);
@@ -343,7 +339,7 @@ runtime::post_outcome runtime::do_post(
     invocation_response const& handler_response)
 {
     set_curl_post_result_options();
-    curl_easy_setopt(m_curl_handle, CURLOPT_URL, url.c_str());
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_URL, url.c_str());
     logging::log_info(LOG_TAG, "Making request to %s", url.c_str());
 
     curl_slist* headers = nullptr;
@@ -364,11 +360,11 @@ runtime::post_outcome runtime::do_post(
 
     std::pair<std::string const&, size_t> ctx{payload, 0};
     aws::http::response resp;
-    curl_easy_setopt(m_curl_handle, CURLOPT_WRITEDATA, &resp);
-    curl_easy_setopt(m_curl_handle, CURLOPT_HEADERDATA, &resp);
-    curl_easy_setopt(m_curl_handle, CURLOPT_READDATA, &ctx);
-    curl_easy_setopt(m_curl_handle, CURLOPT_HTTPHEADER, headers);
-    CURLcode curl_code = curl_easy_perform(m_curl_handle);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_WRITEDATA, &resp);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HEADERDATA, &resp);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_READDATA, &ctx);
+    curl_easy_setopt(lambda_runtime::m_curl_handle, CURLOPT_HTTPHEADER, headers);
+    const CURLcode curl_code = curl_easy_perform(lambda_runtime::m_curl_handle);
     curl_slist_free_all(headers);
 
     if (curl_code != CURLE_OK) {
@@ -382,7 +378,7 @@ runtime::post_outcome runtime::do_post(
     }
 
     long http_response_code;
-    curl_easy_getinfo(m_curl_handle, CURLINFO_RESPONSE_CODE, &http_response_code);
+    curl_easy_getinfo(lambda_runtime::m_curl_handle, CURLINFO_RESPONSE_CODE, &http_response_code);
 
     if (!is_success(aws::http::response_code(http_response_code))) {
         logging::log_error(
@@ -448,7 +444,7 @@ void run_handler(std::function<invocation_response(invocation_request const&)> c
 
         auto const req = std::move(next_outcome).get_result();
         logging::log_info(LOG_TAG, "Invoking user handler");
-        invocation_response res = handler(req);
+        const invocation_response res = handler(req);
         logging::log_info(LOG_TAG, "Invoking user handler completed.");
 
         if (res.is_success()) {
@@ -476,7 +472,7 @@ static std::string json_escape(std::string const& in)
     constexpr char last_non_printable_character = 31;
     std::string out;
     out.reserve(in.length()); // most strings will end up identical
-    for (char ch : in) {
+    for (const char ch : in) {
         if (ch > last_non_printable_character && ch != '\"' && ch != '\\') {
             out.append(1, ch);
         }

tests/CMakeLists.txt

@@ -15,7 +15,8 @@ if(DEFINED ENV{GITHUB_ACTIONS})
     FetchContent_MakeAvailable(gtest)
 
     add_executable(unit_tests
-        unit/no_op_test.cpp)
+        unit/thread_local_curl_test.cpp
+        unit/unit_tests.cpp)
     target_link_libraries(unit_tests PRIVATE gtest_main aws-lambda-runtime)
 
     # Register unit tests
@@ -25,7 +26,22 @@ if(DEFINED ENV{GITHUB_ACTIONS})
         LABELS "unit"
         DISCOVERY_TIMEOUT 10)
 else()
-    message(STATUS "Unit tests skipped: Not in GitHub Actions environment")
+    # Build unit tests using the bundled gtest
+    add_executable(unit_tests
+        unit/thread_local_curl_test.cpp
+        unit/unit_tests.cpp
+        gtest/gtest-all.cc)
+    target_include_directories(unit_tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
+    target_link_libraries(unit_tests PRIVATE aws-lambda-runtime pthread)
+
+    # Provide a main() for gtest
+    target_sources(unit_tests PRIVATE unit/main.cpp)
+
+    include(GoogleTest)
+    gtest_discover_tests(unit_tests
+        PROPERTIES
+        LABELS "unit"
+        DISCOVERY_TIMEOUT 10)
 endif()
 
 

tests/unit/main.cpp

@@ -0,0 +1,7 @@
+#include "gtest/gtest.h"
+
+int main(int argc, char** argv)
+{
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}

tests/unit/thread_local_curl_test.cpp

@@ -0,0 +1,62 @@
+#include <gtest/gtest.h>
+#include <aws/lambda-runtime/runtime.h>
+#include <thread>
+#include <atomic>
+#include <vector>
+
+using namespace aws::lambda_runtime;
+
+TEST(ThreadLocalCurl, runtime_construction_succeeds)
+{
+    runtime rt("http://127.0.0.1:9001");
+    auto outcome = rt.get_next();
+    // We expect a connection failure since there's no server, but the runtime itself should construct fine
+    ASSERT_FALSE(outcome.is_success());
+}
+
+TEST(ThreadLocalCurl, multiple_runtimes_on_same_thread_do_not_crash)
+{
+    {
+        runtime rt1("http://127.0.0.1:9001");
+        auto o1 = rt1.get_next();
+        ASSERT_FALSE(o1.is_success());
+    }
+    {
+        runtime rt2("http://127.0.0.1:9001");
+        auto o2 = rt2.get_next();
+        ASSERT_FALSE(o2.is_success());
+    }
+}
+
+TEST(ThreadLocalCurl, concurrent_runtimes_on_different_threads)
+{
+    constexpr int num_threads = 4;
+    std::vector<std::thread> threads;
+    std::atomic<int> successes{0};
+
+    for (int i = 0; i < num_threads; ++i) {
+        threads.emplace_back([&successes]() {
+            runtime rt("http://127.0.0.1:9001");
+            auto outcome = rt.get_next();
+            // Connection will fail but runtime should not crash or corrupt state
+            if (!outcome.is_success()) {
+                successes.fetch_add(1);
+            }
+        });
+    }
+
+    for (auto& t : threads) {
+        t.join();
+    }
+
+    ASSERT_EQ(successes.load(), num_threads);
+}
+
+TEST(ThreadLocalCurl, sequential_requests_on_same_runtime)
+{
+    runtime rt("http://127.0.0.1:9001");
+    for (int i = 0; i < 5; ++i) {
+        auto outcome = rt.get_next();
+        ASSERT_FALSE(outcome.is_success());
+    }
+}