Add queueing to curl multi

SergeyRyabinin · Sergey Ryabinin · commit 4798c5fd2ede · 2025-01-24T21:21:35.000Z
diff --git a/src/aws-cpp-sdk-core/include/aws/core/http/curl-multi/CurlMultiHandleContainer.h b/src/aws-cpp-sdk-core/include/aws/core/http/curl-multi/CurlMultiHandleContainer.h
@@ -41,6 +41,12 @@ namespace Aws
                                          Version version = Version::HTTP_VERSION_2TLS);
                 ~CurlMultiHandleContainer();
 
+
+                /**
+                  * Gets an available curl handle from the pool, or returns nullptr if no handle is available
+                  */
+                CurlEasyHandleContext* TryAcquireCurlHandle();
+
                 /**
                   * Blocks until a curl handle from the pool is available for use.
                   */
@@ -56,6 +62,11 @@ namespace Aws
                  */
                 void DestroyCurlHandle(CurlEasyHandleContext* handleCtx);
 
+                /**
+                 * Resets handle to an initial state and returns it; or Destroys and creates a new handle if the curl code is bad.
+                 */
+                CurlEasyHandleContext* ResetCurlHandle(CurlEasyHandleContext* handleCtx, const CURLcode code);
+
                 inline CURLM* AccessCurlMultiHandle()
                 {
                   return m_curlMultiHandle;
@@ -67,7 +78,7 @@ namespace Aws
                 CurlMultiHandleContainer(const CurlMultiHandleContainer&&) = delete;
                 const CurlMultiHandleContainer& operator = (const CurlMultiHandleContainer&&) = delete;
 
-                CurlEasyHandleContext* CreateCurlHandleInPool();
+                CurlEasyHandleContext* CreateCurlHandleInPool(bool release = true);
                 bool CheckAndGrowPool();
                 void SetDefaultOptionsOnHandle(CurlEasyHandleContext& handleCtx);
                 static long ConvertHttpVersion(Version version);
diff --git a/src/aws-cpp-sdk-core/include/aws/core/http/curl-multi/CurlMultiHttpClient.h b/src/aws-cpp-sdk-core/include/aws/core/http/curl-multi/CurlMultiHttpClient.h
@@ -103,6 +103,26 @@ class AWS_CORE_API CurlMultiHttpClient: public HttpClient
     virtual void OverrideOptionsOnConnectionHandle(CURL*) const {}
 
 private:
+    struct CurlMultiHttpClientTask
+    {
+       std::shared_ptr<HttpRequest> request;
+       std::shared_ptr<Aws::Utils::Threading::Executor> pExecutor;
+       HttpClient::HttpAsyncOnDoneHandler onDoneHandler;
+       Aws::Utils::RateLimits::RateLimiterInterface* readLimiter = nullptr;
+       Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter = nullptr;
+    };
+
+    mutable Aws::Queue<Aws::UniquePtr<CurlMultiHttpClientTask>> m_MultiCurlTasks;
+    mutable std::mutex m_MultiCurlTasksLock;
+
+    /* Actually submits a request for processing */
+    virtual void SubmitAsyncRequest(Curl::CurlEasyHandleContext* easyHandleContext,
+                                    const std::shared_ptr<HttpRequest>& request,
+                                    std::shared_ptr<Aws::Utils::Threading::Executor> pExecutor,
+                                    HttpClient::HttpAsyncOnDoneHandler onDoneHandler,
+                                    Aws::Utils::RateLimits::RateLimiterInterface* readLimiter,
+                                    Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter) const;
+
     bool SubmitTask(Curl::CurlEasyHandleContext* pEasyHandleCtx) const;
 
     static std::shared_ptr<HttpResponse> HandleCurlResponse(Curl::CurlEasyHandleContext* pEasyHandleCtx);
diff --git a/src/aws-cpp-sdk-core/include/aws/core/utils/ResourceManager.h b/src/aws-cpp-sdk-core/include/aws/core/utils/ResourceManager.h
@@ -28,6 +28,31 @@ namespace Aws
         public:
             ExclusiveOwnershipResourceManager() : m_shutdown(false) {}
 
+            /**
+             * Returns a resource with exclusive ownership. You must call Release on the resource when you are finished or other
+             * threads will block waiting to acquire it.
+             *
+             * @return instance of RESOURCE_TYPE
+             */
+            bool TryAcquire(RESOURCE_TYPE& result)
+            {
+                std::unique_lock<std::mutex> locker(m_queueLock);
+                if (m_shutdown.load() || m_resources.size() == 0)
+                {
+                  result = {};
+                  return false;
+                }
+                else
+                {
+                  assert(!m_shutdown.load());
+
+                  result = std::move(m_resources.back());
+                  m_resources.pop_back();
+
+                  return true;
+                }
+            }
+
             /**
              * Returns a resource with exclusive ownership. You must call Release on the resource when you are finished or other
              * threads will block waiting to acquire it.
diff --git a/src/aws-cpp-sdk-core/source/client/AWSClient.cpp b/src/aws-cpp-sdk-core/source/client/AWSClient.cpp
@@ -797,7 +797,7 @@ void AWSClient::HandleExhaustiveAsyncReply(std::shared_ptr<Aws::Client::AwsClien
               }
             }
             return false;
-        } (); // <- IIFE (immideately invoked lambda)
+        } (); // <- IIFE (immediately invoked lambda)
 
         long sleepMillis = TracingUtils::MakeCallWithTiming<long>(
             [&]() -> long {
diff --git a/src/aws-cpp-sdk-core/source/http/curl-multi/CurlMultiHandleContainer.cpp b/src/aws-cpp-sdk-core/source/http/curl-multi/CurlMultiHandleContainer.cpp
@@ -64,6 +64,27 @@ CurlMultiHandleContainer::~CurlMultiHandleContainer()
     }
 }
 
+CurlEasyHandleContext* CurlMultiHandleContainer::TryAcquireCurlHandle()
+{
+    AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Attempting to acquire curl easy handle.");
+
+    if(!m_handleContainer.HasResourcesAvailable())
+    {
+        AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "No current connections available in pool. Attempting to create new connections.");
+        CheckAndGrowPool();
+    }
+
+    CurlEasyHandleContext* handle = nullptr;
+    if(m_handleContainer.TryAcquire(handle))
+    {
+      AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Returning connection handle " << handle);
+    } else {
+      AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "No connection handle available");
+    }
+
+    return handle;
+}
+
 CurlEasyHandleContext* CurlMultiHandleContainer::AcquireCurlHandle()
 {
     AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Attempting to acquire curl easy handle.");
@@ -121,8 +142,45 @@ void CurlMultiHandleContainer::DestroyCurlHandle(CurlEasyHandleContext* handleCt
     }
 }
 
+CurlEasyHandleContext* CurlMultiHandleContainer::ResetCurlHandle(CurlEasyHandleContext* handleCtx, const CURLcode code)
+{
+  if (code != CURLE_OK)
+  {
+    if(handleCtx && handleCtx->m_curlEasyHandle)
+    {
+        AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Destroy curl handle: " << handleCtx->m_curlEasyHandle);
+        curl_easy_cleanup(handleCtx->m_curlEasyHandle);
+        handleCtx->m_curlEasyHandle = nullptr;
+        Aws::Delete(handleCtx);
+        handleCtx = nullptr;
+    }
+
+    {
+        std::lock_guard<std::mutex> locker(m_containerLock);
+        handleCtx = CreateCurlHandleInPool(false);
+    }
+    if (handleCtx)
+    {
+        AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Created replacement handle: " << handleCtx->m_curlEasyHandle);
+    }
+  }
+  else
+  {
+    CURL* handle = handleCtx ? handleCtx->m_curlEasyHandle : nullptr;
+    if (handle)
+    {
+#if LIBCURL_VERSION_NUM >= 0x074D00 // 7.77.0
+        curl_easy_setopt(handle, CURLOPT_COOKIEFILE, NULL); // workaround a mem leak on curl
+#endif
+        curl_easy_reset(handle);
+        SetDefaultOptionsOnHandle(*handleCtx);
+        curl_easy_setopt(handle, CURLOPT_PRIVATE, handleCtx);
+    }
+  }
+  return handleCtx;
+}
 
-CurlEasyHandleContext* CurlMultiHandleContainer::CreateCurlHandleInPool()
+CurlEasyHandleContext* CurlMultiHandleContainer::CreateCurlHandleInPool(bool release /* = true */)
 {
     CurlEasyHandleContext* handleCtx = Aws::New<CurlEasyHandleContext>(CURL_HANDLE_CONTAINER_TAG);
     if(!handleCtx)
@@ -138,7 +196,9 @@ CurlEasyHandleContext* CurlMultiHandleContainer::CreateCurlHandleInPool()
     {
         SetDefaultOptionsOnHandle(*handleCtx);
         curl_easy_setopt(handleCtx->m_curlEasyHandle, CURLOPT_PRIVATE, handleCtx);
-        m_handleContainer.Release(handleCtx);
+        if (release) {
+          m_handleContainer.Release(handleCtx);
+        }
     }
     else
     {
diff --git a/src/aws-cpp-sdk-core/source/http/curl-multi/CurlMultiHttpClient.cpp b/src/aws-cpp-sdk-core/source/http/curl-multi/CurlMultiHttpClient.cpp
@@ -427,8 +427,6 @@ CurlMultiHttpClient::~CurlMultiHttpClient()
     m_signalRunning.notify_all();
   }
   m_multiHandleThread.join();
-
-  curl_multi_cleanup(m_curlMultiHandleContainer.AccessCurlMultiHandle());
 }
 
 struct curl_slist* PrepareHeaders(const HttpRequest* request, const bool disableExpectHeader)
@@ -680,8 +678,6 @@ std::shared_ptr<HttpResponse> CurlMultiHttpClient::HandleCurlResponse(Curl::Curl
                 AWS_LOGSTREAM_ERROR(CURL_HTTP_CLIENT_TAG, "Response body length doesn't match the content-length header.");
             }
         }
-
-        AWS_LOGSTREAM_DEBUG(CURL_HTTP_CLIENT_TAG, "Releasing curl handle " << connectionHandle);
     }
 
     double timep = 0.0;
@@ -748,13 +744,38 @@ std::shared_ptr<HttpResponse> CurlMultiHttpClient::HandleCurlResponse(Curl::Curl
     std::shared_ptr<HttpResponse> res = std::move(pEasyHandleCtx->writeContext.m_response);
     pEasyHandleCtx->writeContext.m_response.reset();
 
-    if (curlResponseCode != CURLE_OK)
+    Aws::UniquePtr<CurlMultiHttpClientTask> newTask;
     {
-        client->m_curlMultiHandleContainer.DestroyCurlHandle(pEasyHandleCtx);
+      std::unique_lock<std::mutex> lockGuard(client->m_MultiCurlTasksLock);
+      if(!client->m_MultiCurlTasks.empty()) {
+        newTask = std::move(client->m_MultiCurlTasks.front());
+        client->m_MultiCurlTasks.pop();
+      }
+    }
+
+    if(newTask)
+    {
+      AWS_LOGSTREAM_DEBUG(CURL_HTTP_CLIENT_TAG, "Submitting a new task from the queue to the curl handle " << connectionHandle);
+      pEasyHandleCtx = client->m_curlMultiHandleContainer.ResetCurlHandle(pEasyHandleCtx, curlResponseCode);
+
+      client->SubmitAsyncRequest(pEasyHandleCtx,
+                                 std::move(newTask->request),
+                                 std::move(newTask->pExecutor),
+                                 std::move(newTask->onDoneHandler),
+                                 newTask->readLimiter,
+                                 newTask->writeLimiter);
     }
     else
     {
+      AWS_LOGSTREAM_DEBUG(CURL_HTTP_CLIENT_TAG, "Releasing curl handle " << connectionHandle);
+      if (curlResponseCode != CURLE_OK)
+      {
+        client->m_curlMultiHandleContainer.DestroyCurlHandle(pEasyHandleCtx);
+      }
+      else
+      {
         client->m_curlMultiHandleContainer.ReleaseCurlHandle(pEasyHandleCtx);
+      }
     }
 
     return res;
@@ -788,10 +809,57 @@ bool CurlMultiHttpClient::SubmitTask(Curl::CurlEasyHandleContext* pEasyHandleCtx
 
 // Async
 void CurlMultiHttpClient::MakeAsyncRequest(const std::shared_ptr<HttpRequest>& request,
-                                                      std::shared_ptr<Aws::Utils::Threading::Executor> pExecutor,
-                                                      HttpClient::HttpAsyncOnDoneHandler onDoneHandler,
-                                                      Aws::Utils::RateLimits::RateLimiterInterface* readLimiter,
-                                                      Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter) const
+                                           std::shared_ptr<Aws::Utils::Threading::Executor> pExecutor,
+                                           HttpClient::HttpAsyncOnDoneHandler onDoneHandler,
+                                           Aws::Utils::RateLimits::RateLimiterInterface* readLimiter,
+                                           Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter) const
+{
+    if(!pExecutor)
+    {
+      assert(pExecutor);
+      AWS_LOGSTREAM_FATAL(CURL_HTTP_CLIENT_TAG, "Failed to submit async http request: executor is a nullptr");
+    }
+
+    Curl::CurlEasyHandleContext* easyHandleContext = m_curlMultiHandleContainer.TryAcquireCurlHandle();
+    if(easyHandleContext)
+    {
+      return SubmitAsyncRequest(easyHandleContext, std::move(request), std::move(pExecutor), std::move(onDoneHandler), readLimiter, writeLimiter);
+    }
+    else
+    {
+      AWS_LOGSTREAM_DEBUG(CURL_HTTP_CLIENT_TAG, "No available curl handle, queueing the request.");
+      Aws::UniquePtr<CurlMultiHttpClientTask> task = Aws::MakeUnique<CurlMultiHttpClientTask>(CURL_HTTP_CLIENT_TAG);
+      if(!task) {
+        AWS_LOGSTREAM_FATAL(CURL_HTTP_CLIENT_TAG, "Failed to allocate a UniquePtr for holding enqueued http client task!");
+        pExecutor->Submit([request, onDoneHandler]()
+                          {
+                              std::shared_ptr<HttpResponse> response = Aws::MakeShared<StandardHttpResponse>(CURL_HTTP_CLIENT_TAG, request);
+                              if(response)
+                              {
+                                response->SetClientErrorType(CoreErrors::MEMORY_ALLOCATION);
+                                response->SetClientErrorMessage("Failed to allocate a UniquePtr for holding enqueued http client task");
+                              }
+                              onDoneHandler(response);
+                          } );
+        return;
+      }
+      task->request = std::move(request);
+      task->pExecutor = std::move(pExecutor);
+      task->onDoneHandler = std::move(onDoneHandler);
+      task->readLimiter = readLimiter;
+      task->writeLimiter = writeLimiter;
+
+      std::unique_lock<std::mutex> lockGuard(m_MultiCurlTasksLock);
+      m_MultiCurlTasks.emplace(std::move(task));
+    }
+}
+
+void CurlMultiHttpClient::SubmitAsyncRequest(Curl::CurlEasyHandleContext* easyHandleContext,
+                                             const std::shared_ptr<HttpRequest>& request,
+                                             std::shared_ptr<Aws::Utils::Threading::Executor> pExecutor,
+                                             HttpClient::HttpAsyncOnDoneHandler onDoneHandler,
+                                             Aws::Utils::RateLimits::RateLimiterInterface* readLimiter,
+                                             Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter) const
 {
     if(!pExecutor)
     {
@@ -810,7 +878,6 @@ void CurlMultiHttpClient::MakeAsyncRequest(const std::shared_ptr<HttpRequest>& r
         return;
     }
 
-    Curl::CurlEasyHandleContext* easyHandleContext = m_curlMultiHandleContainer.AcquireCurlHandle();
     if(!easyHandleContext)
     {
         response->SetClientErrorType(CoreErrors::NETWORK_CONNECTION);

Original file line number	Diff line number	Diff line change
`@@ -797,7 +797,7 @@ void AWSClient::HandleExhaustiveAsyncReply(std::shared_ptr<Aws::Client::AwsClien`
`797`	`797`	`}`
`798`	`798`	`}`
`799`	`799`	`return false;`
`800`		`- } (); // <- IIFE (immideately invoked lambda)`
	`800`	`+ } (); // <- IIFE (immediately invoked lambda)`
`801`	`801`
`802`	`802`	`long sleepMillis = TracingUtils::MakeCallWithTiming<long>(`
`803`	`803`	`[&]() -> long {`