Use curl_multi_handle

Author: SergeyRyabinin

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

@@ -0,0 +1,85 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#pragma once
+
+#include <aws/core/utils/ResourceManager.h>
+#include <aws/core/http/Version.h>
+
+#include <utility>
+#include <curl/curl.h>
+
+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();
+
+    /**
+      * 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);
+
+    /**
+     * When the handle has bad DNS entries, problematic live connections, we need to destroy the handle from pool.
+     */
+    void DestroyCurlHandle(CURL* handle);
+
+    inline CURLM* AccessCurlMultiHandle()
+    {
+        return m_curlMultiHandle;
+    }
+
+private:
+    CurlMultiHandleContainer(const CurlMultiHandleContainer&) = delete;
+    const CurlMultiHandleContainer& operator = (const CurlMultiHandleContainer&) = delete;
+    CurlMultiHandleContainer(const CurlMultiHandleContainer&&) = delete;
+    const CurlMultiHandleContainer& operator = (const CurlMultiHandleContainer&&) = delete;
+
+    CURL* CreateCurlHandleInPool();
+    bool CheckAndGrowPool();
+    void SetDefaultOptionsOnHandle(CURL* handle);
+    static long ConvertHttpVersion(Version version);
+
+    Aws::Utils::ExclusiveOwnershipResourceManager<CURL*> m_handleContainer;
+    CURLM* m_curlMultiHandle = nullptr;
+
+    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 Http
+} // namespace Aws
+

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

@@ -0,0 +1,112 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+
+#pragma once
+
+#include <aws/core/Core_EXPORTS.h>
+#include <aws/core/http/HttpClient.h>
+#include <aws/core/http/curl-multi/CurlMultiHandleContainer.h>
+#include <aws/core/client/ClientConfiguration.h>
+#include <aws/core/utils/memory/stl/AWSString.h>
+#include <aws/core/utils/memory/stl/AWSQueue.h>
+#include <atomic>
+#include <thread>
+#include <queue>
+
+namespace Aws
+{
+namespace Http
+{
+namespace Standard
+{
+    class StandardHttpResponse;
+}
+
+//Curl implementation of an http client. Right now it is only synchronous.
+class AWS_CORE_API CurlMultiHttpClient: public HttpClient
+{
+public:
+    struct CurlMultiHttpClientConfig
+    {
+        struct ProxyConfig
+        {
+            bool isEnabled = false;
+            Aws::String userName;
+            Aws::String password;
+            Aws::String scheme;
+            Aws::String host;
+            Aws::String sslCertPath;
+            Aws::String sslCertType;
+            Aws::String sslKeyPath;
+            Aws::String sslKeyType;
+            Aws::String keyPasswd;
+            unsigned port = 0;
+            Aws::String nonProxyHosts;
+        };
+
+        struct SslConfig
+        {
+            bool verifySSL = true;
+            Aws::String caPath;
+            Aws::String caFile;
+        };
+
+        ProxyConfig proxyConfig;
+        SslConfig sslConfig;
+
+        bool disableExpectHeader = false;
+        bool allowRedirects = false;
+    };
+    struct CurlEasyHandleContext;
+
+    using Base = HttpClient;
+
+    //Creates client, initializes curl handle if it hasn't been created already.
+    CurlMultiHttpClient(const Aws::Client::ClientConfiguration& clientConfig);
+    virtual ~CurlMultiHttpClient();
+
+    //Makes request and receives response synchronously
+    std::shared_ptr<HttpResponse> MakeRequest(const std::shared_ptr<HttpRequest>& request,
+        Aws::Utils::RateLimits::RateLimiterInterface* readLimiter = nullptr,
+        Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter = nullptr) const override;
+
+    static void InitGlobalState();
+    static void CleanupGlobalState();
+
+protected:
+    /**
+     * Override any configuration on CURL handle for each request before sending.
+     * The usage is to have a subclass of CurlMultiHttpClient and have your own implementation of this function to configure whatever you want on CURL handle.
+     */
+    virtual void OverrideOptionsOnConnectionHandle(CURL*) const {}
+
+private:
+    void SubmitTask(std::shared_ptr<CurlEasyHandleContext> pEasyHandleCtx) const;
+
+    static std::shared_ptr<HttpResponse> HandleCurlResponse(std::shared_ptr<CurlEasyHandleContext> pEasyHandleCtx);
+    static void CurlMultiPerformThread(CurlMultiHttpClient* pClient);
+
+    std::thread m_multiHandleThread;
+    std::atomic<bool> m_isRunning;
+    mutable std::mutex m_signalMutex;
+    mutable std::condition_variable m_signalRunning;
+
+    // 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;
+
+    CurlMultiHttpClientConfig m_config;
+
+    mutable CurlMultiHandleContainer m_curlMultiHandleContainer;
+
+    static std::atomic<bool> isGlobalStateInit;
+    std::shared_ptr<smithy::components::tracing::TelemetryProvider> m_telemetryProvider;
+};
+
+} // namespace Http
+} // namespace Aws
+

src/aws-cpp-sdk-core/include/aws/core/utils/ResourceManager.h

@@ -15,7 +15,7 @@ namespace Aws
     namespace Utils
     {
         /**
-         * Generic resource manager with Acquire/Release semantics. Acquire will block waiting on a an available resource. Release will
+         * Generic resource manager with Acquire/Release semantics. Acquire will block waiting on an available resource. Release will
          * cause one blocked acquisition to unblock.
          *
          * You must call ShutdownAndWait() when finished with this container, this unblocks the listening thread and gives you a chance to

src/aws-cpp-sdk-core/source/http/HttpClientFactory.cpp

@@ -12,6 +12,7 @@
 #endif
 #if ENABLE_CURL_CLIENT
 #include <aws/core/http/curl/CurlHttpClient.h>
+#include <aws/core/http/curl-multi/CurlMultiHttpClient.h>
 #include <signal.h>
 
 #elif ENABLE_WINDOWS_CLIENT
@@ -93,7 +94,7 @@ namespace Aws
                 }
 #endif // ENABLE_WINDOWS_IXML_HTTP_REQUEST_2_CLIENT
 #elif ENABLE_CURL_CLIENT
-                return Aws::MakeShared<CurlHttpClient>(HTTP_CLIENT_FACTORY_ALLOCATION_TAG, clientConfiguration);
+                return Aws::MakeShared<CurlMultiHttpClient>(HTTP_CLIENT_FACTORY_ALLOCATION_TAG, clientConfiguration);
 #else
                 // When neither of these clients is enabled, gcc gives a warning (converted
                 // to error by -Werror) about the unused clientConfiguration parameter. We