Avoid locking on submitted tasks to multi curl handle by keeping easy handle sdk wrappers in a container

Author: SergeyRyabinin

src/aws-cpp-sdk-core/include/aws/core/http/curl-multi/CurlEasyHandleContext.h

@@ -0,0 +1,99 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#pragma once
+
+#include <aws/core/utils/DateTime.h>
+#include <aws/core/utils/ResourceManager.h>
+#include <aws/core/http/Version.h>
+
+#include <utility>
+#include <curl/curl.h>
+
+namespace Aws
+{
+namespace Utils
+{
+namespace RateLimits
+{
+    class RateLimiterInterface;
+}
+}
+
+namespace Http
+{
+    class CurlMultiHttpClient;
+    class HttpRequest;
+    class HttpResponse;
+
+    namespace Curl
+    {
+        enum class ExecutionPolicy
+        {
+            BLOCKING,
+            ASYNC
+        };
+
+        /* A composition wrapper on curl easy handle to have curl handle and SDK context together*/
+        struct CurlEasyHandleContext final
+        {
+            struct WriteContext final
+            {
+                WriteContext() = default;
+
+                bool m_HasBody = false;
+                HttpRequest *m_request = nullptr;
+                std::shared_ptr<HttpResponse> m_response;
+                Aws::Utils::RateLimits::RateLimiterInterface *m_rateLimiter = nullptr;
+                int64_t m_numBytesResponseReceived = 0;
+            };
+
+            struct ReadContext final
+            {
+                ReadContext() = default;
+
+                Aws::Utils::RateLimits::RateLimiterInterface *m_rateLimiter = nullptr;
+                HttpRequest *m_request = nullptr;
+                bool m_chunkEnd = false;
+            };
+
+            CurlEasyHandleContext() = default;
+
+            /* Curl side */
+            CURL* m_curlEasyHandle = nullptr;
+            // headers set on curl handle, to be cleaned up by SDK after the execution
+            curl_slist* m_curlHandleHeaders = nullptr;
+
+            /* SDK side */
+            const CurlMultiHttpClient* m_client = nullptr;
+            ExecutionPolicy m_execPolicy;
+            std::function<void()> m_onCurlDoneFn;
+            Aws::Utils::DateTime startTransmissionTime;
+
+            /* Curl calls the SDK back */
+            WriteContext writeContext;
+            ReadContext readContext;
+
+            /* SDK polls the curl result */
+            CURLcode curlResult;
+            // ptr acquired by curl_multi_info_read, free-ed by curl in easy handle cleanup, do not free in SDK
+            CURLMsg* curlResultMsg = nullptr;
+
+            /* callbacks set on easy handle */
+            static size_t WriteData(char* ptr, size_t size, size_t nmemb, void* userdata);
+            static size_t WriteHeader(char* ptr, size_t size, size_t nmemb, void* userdata);
+            static size_t ReadBody(char* ptr, size_t size, size_t nmemb, void* userdata, bool isStreaming);
+            static size_t ReadBodyStreaming(char* ptr, size_t size, size_t nmemb, void* userdata);
+            static size_t ReadBodyFunc(char* ptr, size_t size, size_t nmemb, void* userdata);
+            static size_t SeekBody(void* userdata, curl_off_t offset, int origin);
+#if LIBCURL_VERSION_NUM >= 0x072000 // 7.32.0
+            static int CurlProgressCallback(void *userdata, curl_off_t, curl_off_t, curl_off_t, curl_off_t);
+#else
+            static int CurlProgressCallback(void *userdata, double, double, double, double);
+#endif
+        };
+    } // namespace Curl
+} // namespace Http
+} // namespace Aws

src/aws-cpp-sdk-core/include/aws/core/http/curl-multi/CurlMultiHandleContainer.h

@@ -13,73 +13,80 @@
 
 namespace Aws
 {
-namespace Http
-{
-
-/**
-  * Simple Connection pool manager for Curl. It maintains connections in a thread safe manner. You
-  * can call into acquire a handle, then put it back when finished. It is assumed that reusing an already
-  * initialized handle is preferable (especially for synchronous clients). The pool doubles in capacity as
-  * needed up to the maximum amount of connections.
-  */
-class CurlMultiHandleContainer
-{
-public:
-    /**
-      * Initializes an empty stack of CURL handles. If you are only making synchronous calls via your http client
-      * then a small size is best. For async support, a good value would be 6 * number of Processors.   *
-      */
-    CurlMultiHandleContainer(unsigned maxSize = 50, long httpRequestTimeout = 0, long connectTimeout = 1000, bool tcpKeepAlive = true,
-                        unsigned long tcpKeepAliveIntervalMs = 30000, long lowSpeedTime = 3000, unsigned long lowSpeedLimit = 1,
-                        Version version = Version::HTTP_VERSION_2TLS);
-    ~CurlMultiHandleContainer();
+    namespace Http
+    {
+        namespace Curl
+        {
+            struct CurlEasyHandleContext;
 
-    /**
-      * Blocks until a curl handle from the pool is available for use.
-      */
-    CURL* AcquireCurlHandle();
-    /**
-      * Returns a handle to the pool for reuse. It is imperative that this is called
-      * after you are finished with the handle.
-      */
-    void ReleaseCurlHandle(CURL* handle);
+            /**
+            * Simple Connection pool manager for Curl. It maintains connections in a thread safe manner. You
+            * can call into acquire a handle, then put it back when finished. It is assumed that reusing an already
+            * initialized handle is preferable (especially for synchronous clients). The pool doubles in capacity as
+            * needed up to the maximum amount of connections.
+            */
+            class CurlMultiHandleContainer
+            {
+            public:
+                /**
+                  * Initializes an empty stack of CURL handles.
+                  */
+                CurlMultiHandleContainer(unsigned maxSize = 50,
+                                         long httpRequestTimeout = 0,
+                                         long connectTimeout = 1000,
+                                         bool tcpKeepAlive = true,
+                                         unsigned long tcpKeepAliveIntervalMs = 30000,
+                                         long lowSpeedTime = 3000,
+                                         unsigned long lowSpeedLimit = 1,
+                                         Version version = Version::HTTP_VERSION_2TLS);
+                ~CurlMultiHandleContainer();
 
-    /**
-     * When the handle has bad DNS entries, problematic live connections, we need to destroy the handle from pool.
-     */
-    void DestroyCurlHandle(CURL* handle);
+                /**
+                  * Blocks until a curl handle from the pool is available for use.
+                  */
+                CurlEasyHandleContext* AcquireCurlHandle();
+                /**
+                  * Returns a handle to the pool for reuse. It is imperative that this is called
+                  * after you are finished with the handle.
+                  */
+                void ReleaseCurlHandle(CurlEasyHandleContext* handleCtx);
 
-    inline CURLM* AccessCurlMultiHandle()
-    {
-        return m_curlMultiHandle;
-    }
+                /**
+                 * When the handle has bad DNS entries, problematic live connections, we need to destroy the handle from pool.
+                 */
+                void DestroyCurlHandle(CurlEasyHandleContext* handleCtx);
 
-private:
-    CurlMultiHandleContainer(const CurlMultiHandleContainer&) = delete;
-    const CurlMultiHandleContainer& operator = (const CurlMultiHandleContainer&) = delete;
-    CurlMultiHandleContainer(const CurlMultiHandleContainer&&) = delete;
-    const CurlMultiHandleContainer& operator = (const CurlMultiHandleContainer&&) = delete;
+                inline CURLM* AccessCurlMultiHandle()
+                {
+                  return m_curlMultiHandle;
+                }
 
-    CURL* CreateCurlHandleInPool();
-    bool CheckAndGrowPool();
-    void SetDefaultOptionsOnHandle(CURL* handle);
-    static long ConvertHttpVersion(Version version);
+            private:
+                CurlMultiHandleContainer(const CurlMultiHandleContainer&) = delete;
+                const CurlMultiHandleContainer& operator = (const CurlMultiHandleContainer&) = delete;
+                CurlMultiHandleContainer(const CurlMultiHandleContainer&&) = delete;
+                const CurlMultiHandleContainer& operator = (const CurlMultiHandleContainer&&) = delete;
 
-    Aws::Utils::ExclusiveOwnershipResourceManager<CURL*> m_handleContainer;
-    CURLM* m_curlMultiHandle = nullptr;
+                CurlEasyHandleContext* CreateCurlHandleInPool();
+                bool CheckAndGrowPool();
+                void SetDefaultOptionsOnHandle(CurlEasyHandleContext& handleCtx);
+                static long ConvertHttpVersion(Version version);
 
-    unsigned m_maxPoolSize;
-    unsigned long m_httpRequestTimeout;
-    unsigned long m_connectTimeout;
-    bool m_enableTcpKeepAlive;
-    unsigned long m_tcpKeepAliveIntervalMs;
-    unsigned long m_lowSpeedTime;
-    unsigned long m_lowSpeedLimit;
-    unsigned m_poolSize;
-    std::mutex m_containerLock;
-    Version m_version;
-};
+                Aws::Utils::ExclusiveOwnershipResourceManager<CurlEasyHandleContext*> m_handleContainer;
+                CURLM* m_curlMultiHandle = nullptr;
 
-} // namespace Http
-} // namespace Aws
+                unsigned m_maxPoolSize;
+                unsigned long m_httpRequestTimeout;
+                unsigned long m_connectTimeout;
+                bool m_enableTcpKeepAlive;
+                unsigned long m_tcpKeepAliveIntervalMs;
+                unsigned long m_lowSpeedTime;
+                unsigned long m_lowSpeedLimit;
+                unsigned m_poolSize;
+                std::mutex m_containerLock;
+                Version m_version;
+            };
 
+        } // namespace Curl
+    } // namespace Http
+} // namespace Aws

src/aws-cpp-sdk-core/include/aws/core/http/curl-multi/CurlMultiHttpClient.h

@@ -20,6 +20,10 @@ namespace Aws
 {
 namespace Http
 {
+namespace Curl
+{
+    struct CurlEasyHandleContext;
+}
 namespace Standard
 {
     class StandardHttpResponse;
@@ -60,8 +64,6 @@ class AWS_CORE_API CurlMultiHttpClient: public HttpClient
         bool disableExpectHeader = false;
         bool allowRedirects = false;
     };
-    struct CurlEasyHandleContext;
-
     using Base = HttpClient;
 
     //Creates client, initializes curl handle if it hasn't been created already.
@@ -84,9 +86,9 @@ class AWS_CORE_API CurlMultiHttpClient: public HttpClient
     virtual void OverrideOptionsOnConnectionHandle(CURL*) const {}
 
 private:
-    void SubmitTask(std::shared_ptr<CurlEasyHandleContext> pEasyHandleCtx) const;
+    void SubmitTask(Curl::CurlEasyHandleContext* pEasyHandleCtx) const;
 
-    static std::shared_ptr<HttpResponse> HandleCurlResponse(std::shared_ptr<CurlEasyHandleContext> pEasyHandleCtx);
+    static std::shared_ptr<HttpResponse> HandleCurlResponse(Curl::CurlEasyHandleContext* pEasyHandleCtx);
     static void CurlMultiPerformThread(CurlMultiHttpClient* pClient);
 
     std::thread m_multiHandleThread;
@@ -97,11 +99,11 @@ class AWS_CORE_API CurlMultiHttpClient: public HttpClient
     // mutable std::mutex m_tasksMutex;
     mutable std::atomic<size_t> m_tasksQueued;
     mutable std::mutex m_tasksMutex;
-    mutable Aws::UnorderedMap<CURL*, std::shared_ptr<CurlEasyHandleContext>> m_tasks;
+    mutable Aws::UnorderedMap<CURL*, std::shared_ptr<Curl::CurlEasyHandleContext>> m_tasks;
 
     CurlMultiHttpClientConfig m_config;
 
-    mutable CurlMultiHandleContainer m_curlMultiHandleContainer;
+    mutable Curl::CurlMultiHandleContainer m_curlMultiHandleContainer;
 
     static std::atomic<bool> isGlobalStateInit;
     std::shared_ptr<smithy::components::tracing::TelemetryProvider> m_telemetryProvider;