1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
/**
* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
* SPDX-License-Identifier: Apache-2.0.
*/
#include <aws/core/http/curl/CurlHandleContainer.h>
#include <aws/core/utils/logging/LogMacros.h>
#include <algorithm>
using namespace Aws::Utils::Logging;
using namespace Aws::Http;
static const char* CURL_HANDLE_CONTAINER_TAG = "CurlHandleContainer";
CurlHandleContainer::CurlHandleContainer(unsigned maxSize, long httpRequestTimeout, long connectTimeout, bool enableTcpKeepAlive,
unsigned long tcpKeepAliveIntervalMs, long lowSpeedTime, unsigned long lowSpeedLimit) :
m_maxPoolSize(maxSize), m_httpRequestTimeout(httpRequestTimeout), m_connectTimeout(connectTimeout), m_enableTcpKeepAlive(enableTcpKeepAlive),
m_tcpKeepAliveIntervalMs(tcpKeepAliveIntervalMs), m_lowSpeedTime(lowSpeedTime), m_lowSpeedLimit(lowSpeedLimit), m_poolSize(0)
{
AWS_LOGSTREAM_INFO(CURL_HANDLE_CONTAINER_TAG, "Initializing CurlHandleContainer with size " << maxSize);
}
CurlHandleContainer::~CurlHandleContainer()
{
AWS_LOGSTREAM_INFO(CURL_HANDLE_CONTAINER_TAG, "Cleaning up CurlHandleContainer.");
for (CURL* handle : m_handleContainer.ShutdownAndWait(m_poolSize))
{
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Cleaning up " << handle);
curl_easy_cleanup(handle);
}
}
CURL* CurlHandleContainer::AcquireCurlHandle()
{
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Attempting to acquire curl connection.");
if(!m_handleContainer.HasResourcesAvailable())
{
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "No current connections available in pool. Attempting to create new connections.");
CheckAndGrowPool();
}
CURL* handle = m_handleContainer.Acquire();
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Connection has been released. Continuing.");
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Returning connection handle " << handle);
return handle;
}
void CurlHandleContainer::ReleaseCurlHandle(CURL* handle)
{
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(handle);
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Releasing curl handle " << handle);
m_handleContainer.Release(handle);
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Notified waiting threads.");
}
}
void CurlHandleContainer::DestroyCurlHandle(CURL* handle)
{
if (!handle)
{
return;
}
curl_easy_cleanup(handle);
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Destroy curl handle: " << handle);
{
std::lock_guard<std::mutex> locker(m_containerLock);
// Other threads could be blocked and waiting on m_handleContainer.Acquire()
// If the handle is not released back to the pool, it could create a deadlock
// Create a new handle and release that into the pool
handle = CreateCurlHandleInPool();
}
if (handle)
{
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "Created replacement handle and released to pool: " << handle);
}
}
CURL* CurlHandleContainer::CreateCurlHandleInPool()
{
CURL* curlHandle = curl_easy_init();
if (curlHandle)
{
SetDefaultOptionsOnHandle(curlHandle);
m_handleContainer.Release(curlHandle);
}
else
{
AWS_LOGSTREAM_ERROR(CURL_HANDLE_CONTAINER_TAG, "curl_easy_init failed to allocate.");
}
return curlHandle;
}
bool CurlHandleContainer::CheckAndGrowPool()
{
std::lock_guard<std::mutex> locker(m_containerLock);
if (m_poolSize < m_maxPoolSize)
{
unsigned multiplier = m_poolSize > 0 ? m_poolSize : 1;
unsigned amountToAdd = (std::min)(multiplier * 2, m_maxPoolSize - m_poolSize);
AWS_LOGSTREAM_DEBUG(CURL_HANDLE_CONTAINER_TAG, "attempting to grow pool size by " << amountToAdd);
unsigned actuallyAdded = 0;
for (unsigned i = 0; i < amountToAdd; ++i)
{
CURL* curlHandle = CreateCurlHandleInPool();
if (curlHandle)
{
++actuallyAdded;
}
else
{
break;
}
}
AWS_LOGSTREAM_INFO(CURL_HANDLE_CONTAINER_TAG, "Pool grown by " << actuallyAdded);
m_poolSize += actuallyAdded;
return actuallyAdded > 0;
}
AWS_LOGSTREAM_INFO(CURL_HANDLE_CONTAINER_TAG, "Pool cannot be grown any further, already at max size.");
return false;
}
void CurlHandleContainer::SetDefaultOptionsOnHandle(CURL* handle)
{
//for timeouts to work in a multi-threaded context,
//always turn signals off. This also forces dns queries to
//not be included in the timeout calculations.
curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt(handle, CURLOPT_TIMEOUT_MS, m_httpRequestTimeout);
curl_easy_setopt(handle, CURLOPT_CONNECTTIMEOUT_MS, m_connectTimeout);
curl_easy_setopt(handle, CURLOPT_LOW_SPEED_LIMIT, m_lowSpeedLimit);
curl_easy_setopt(handle, CURLOPT_LOW_SPEED_TIME, m_lowSpeedTime < 1000 ? (m_lowSpeedTime == 0 ? 0 : 1) : m_lowSpeedTime / 1000);
curl_easy_setopt(handle, CURLOPT_TCP_KEEPALIVE, m_enableTcpKeepAlive ? 1L : 0L);
curl_easy_setopt(handle, CURLOPT_TCP_KEEPINTVL, m_tcpKeepAliveIntervalMs / 1000);
curl_easy_setopt(handle, CURLOPT_TCP_KEEPIDLE, m_tcpKeepAliveIntervalMs / 1000);
#ifdef CURL_HAS_H2
curl_easy_setopt(handle, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2_0);
#endif
}
|
