Singularity-Viewer/SingularityViewer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HTTPTimeout work in progress.

Browse Source 
* Moved DoutCurlEasy and DoutCurlEasyEntering from aicurl.cpp
  to aicurl.h and renamed them to DoutCurl and DoutCurlEntering
  respectively.
* Moved the callback functions from aicurl.cpp to aicurlthread.cpp.
* In CurlEasyRequest, renamed timeout_timings to print_curl_timings
  and mTimeoutLowercaseHostname to mLowercaseHostname.
* Put all remaining CurlEasyRequest::mTimeout* variables and
  timeout_* methods in curlthread::HTTPTimeout, stripping them
  of said prefix, and moved the definition to aicurlprivate.h.
  Added a ThreadSafeCurlEasyRequest* member and a get_lockobj()
  method so to that class so we can still use DoutCurl /
  DoutCurlEntering. timeout_add_easy_request was removed completely
  and reimplemented as the constructor of HTTPTimeout.
  timeout_has_stalled was renamed to HTTPTimeout::has_stalled,
  but also reimplemented as CurlEasyRequest::has_stalled.
* CurlEasyRequest::mRequestFinalized was removed and it's
  functionality taken over by CurlEasyRequest::mTimeoutPolicy.
* Fixed the indentation of struct Stats, class CurlEasyHandle
  and class CurlEasyRequest.
* Added CurlEasyRequest::set_timeout_opts
* Added CurlSocketInfo::mTimeout (LLPointer<HTTPTimeout>).
* mUploadFinished is now reset in HTTPTimeout::data_received,
  this was needed because "HEAD /something" header-only
  messages triggered upload_finished (unlike "GET ..."),
  and in combination with redirection that caused an assert.
This commit is contained in:
Aleric Inglewood
2012-10-15 03:34:32 +02:00
parent 9fba54ad5d
commit 76fc30e460
4 changed files with 961 additions and 921 deletions
Download Patch File Download Diff File Expand all files Collapse all files

689
indra/aistatemachine/aicurl.cpp
View File

@@ -59,38 +59,6 @@
#include "aihttptimeoutpolicy.h"
#include "aicurleasyrequeststatemachine.h"
// Some pretty printing for curl easy handle related things:
// Print the lock object related to the current easy handle in every debug output.
#ifdef CWDEBUG
#include <libcwd/buf2str.h>
#include <sstream>
#define DoutCurlEasy(x) do { \
using namespace libcwd; \
std::ostringstream marker; \
marker << (void*)this->get_lockobj(); \
libcw_do.push_marker(); \
libcw_do.marker().assign(marker.str().data(), marker.str().size()); \
libcw_do.inc_indent(2); \
Dout(dc::curl, x); \
libcw_do.dec_indent(2); \
libcw_do.pop_marker(); \
} while(0)
#define DoutCurlEasyEntering(x) do { \
using namespace libcwd; \
std::ostringstream marker; \
marker << (void*)this->get_lockobj(); \
libcw_do.push_marker(); \
libcw_do.marker().assign(marker.str().data(), marker.str().size()); \
libcw_do.inc_indent(2); \
DoutEntering(dc::curl, x); \
libcw_do.dec_indent(2); \
libcw_do.pop_marker(); \
} while(0)
#else // !CWDEBUG
#define DoutCurlEasy(x) Dout(dc::curl, x << " [" << (void*)this->get_lockobj() << ']')
#define DoutCurlEasyEntering(x) DoutEntering(dc::curl, x << " [" << (void*)this->get_lockobj() << ']')
#endif // CWDEBUG
//==================================================================================
// Local variables.
//
@@ -571,6 +539,11 @@ void intrusive_ptr_release(Responder* responder)
namespace AICurlPrivate {
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
// CURLOPT_DEBUGFUNCTION function.
extern int debug_callback(CURL*, curl_infotype infotype, char* buf, size_t size, void* user_ptr);
#endif
//static
LLAtomicU32 Stats::easy_calls;
LLAtomicU32 Stats::easy_errors;
@@ -841,7 +814,7 @@ void CurlEasyRequest::setPost(AIPostFieldPtr const& postdata, U32 size)
{
llassert_always(postdata->data());
DoutCurlEasy("POST size is " << size << " bytes: \"" << libcwd::buf2str(postdata->data(), size) << "\".");
DoutCurl("POST size is " << size << " bytes: \"" << libcwd::buf2str(postdata->data(), size) << "\".");
setPostField(postdata); // Make sure the data stays around until we don't need it anymore.
setPost_raw(size, postdata->data());
@@ -852,7 +825,7 @@ void CurlEasyRequest::setPost_raw(U32 size, char const* data)
if (!data)
{
// data == NULL when we're going to read the data using CURLOPT_READFUNCTION.
DoutCurlEasy("POST size is " << size << " bytes.");
DoutCurl("POST size is " << size << " bytes.");
}
// Accept everything (send an Accept-Encoding header containing all encodings we support (zlib and gzip)).
@@ -1015,7 +988,8 @@ void CurlEasyRequest::resetState(void)
reset();
curl_slist_free_all(mHeaders);
mHeaders = NULL;
mRequestFinalized = false;
mTimeoutPolicy = NULL;
mTimeout = NULL;
mEventsTarget = NULL;
mResult = CURLE_FAILED_INIT;
applyDefaultOptions();
@@ -1023,125 +997,16 @@ void CurlEasyRequest::resetState(void)
void CurlEasyRequest::addHeader(char const* header)
{
llassert(!mRequestFinalized);
llassert(!mTimeoutPolicy); // Cannot add a header after calling finalizeRequest.
mHeaders = curl_slist_append(mHeaders, header);
}
void CurlEasyRequest::addHeaders(AIHTTPHeaders const& headers)
{
llassert(!mRequestFinalized);
llassert(!mTimeoutPolicy); // Cannot add headers after calling finalizeRequest.
headers.append_to(mHeaders);
}
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
static int curl_debug_cb(CURL*, curl_infotype infotype, char* buf, size_t size, void* user_ptr)
{
#ifdef CWDEBUG
using namespace ::libcwd;
CurlEasyRequest* request = (CurlEasyRequest*)user_ptr;
std::ostringstream marker;
marker << (void*)request->get_lockobj();
libcw_do.push_marker();
libcw_do.marker().assign(marker.str().data(), marker.str().size());
if (!debug::channels::dc::curlio.is_on())
debug::channels::dc::curlio.on();
LibcwDoutScopeBegin(LIBCWD_DEBUGCHANNELS, libcw_do, dc::curlio|cond_nonewline_cf(infotype == CURLINFO_TEXT))
#else
if (infotype == CURLINFO_TEXT)
{
while (size > 0 && (buf[size - 1] == '\r' || buf[size - 1] == '\n'))
--size;
}
LibcwDoutScopeBegin(LIBCWD_DEBUGCHANNELS, libcw_do, dc::curlio)
#endif
switch (infotype)
{
case CURLINFO_TEXT:
LibcwDoutStream << "* ";
break;
case CURLINFO_HEADER_IN:
LibcwDoutStream << "H> ";
break;
case CURLINFO_HEADER_OUT:
LibcwDoutStream << "H< ";
if (size >= 4 && strncmp(buf, "GET ", 4) == 0)
request->mDebugIsGetMethod = true;
break;
case CURLINFO_DATA_IN:
LibcwDoutStream << "D> ";
break;
case CURLINFO_DATA_OUT:
LibcwDoutStream << "D< ";
break;
case CURLINFO_SSL_DATA_IN:
LibcwDoutStream << "S> ";
break;
case CURLINFO_SSL_DATA_OUT:
LibcwDoutStream << "S< ";
break;
default:
LibcwDoutStream << "?? ";
}
if (infotype == CURLINFO_TEXT)
LibcwDoutStream.write(buf, size);
else if (infotype == CURLINFO_HEADER_IN || infotype == CURLINFO_HEADER_OUT)
LibcwDoutStream << libcwd::buf2str(buf, size);
else if (infotype == CURLINFO_DATA_IN)
{
LibcwDoutStream << size << " bytes";
bool finished = false;
size_t i = 0;
while (i < size)
{
char c = buf[i];
if (!('0' <= c && c <= '9') && !('a' <= c && c <= 'f'))
{
if (0 < i && i + 1 < size && buf[i] == '\r' && buf[i + 1] == '\n')
{
// Binary output: "[0-9a-f]*\r\n ...binary data..."
LibcwDoutStream << ": \"" << libcwd::buf2str(buf, i + 2) << "\"...";
finished = true;
}
break;
}
++i;
}
if (!finished && size > 9 && buf[0] == '<')
{
// Human readable output: html, xml or llsd.
if (!strncmp(buf, "<!DOCTYPE", 9) || !strncmp(buf, "<?xml", 5) || !strncmp(buf, "<llsd>", 6))
{
LibcwDoutStream << ": \"" << libcwd::buf2str(buf, size) << '"';
finished = true;
}
}
if (!finished)
{
// Unknown format. Only print the first and last 20 characters.
if (size > 40UL)
{
LibcwDoutStream << ": \"" << libcwd::buf2str(buf, 20) << "\"...\"" << libcwd::buf2str(&buf[size - 20], 20) << '"';
}
else
{
LibcwDoutStream << ": \"" << libcwd::buf2str(buf, size) << '"';
}
}
}
else if (infotype == CURLINFO_DATA_OUT)
LibcwDoutStream << size << " bytes: \"" << libcwd::buf2str(buf, size) << '"';
else
LibcwDoutStream << size << " bytes";
LibcwDoutScopeEnd;
#ifdef CWDEBUG
libcw_do.pop_marker();
#endif
return 0;
}
#endif
void CurlEasyRequest::applyProxySettings(void)
{
LLProxy& proxy = *LLProxy::getInstance();
@@ -1232,7 +1097,7 @@ void CurlEasyRequest::applyDefaultOptions(void)
if (dc::curlio.is_on())
{
setopt(CURLOPT_VERBOSE, 1);
setopt(CURLOPT_DEBUGFUNCTION, &curl_debug_cb);
setopt(CURLOPT_DEBUGFUNCTION, &debug_callback);
setopt(CURLOPT_DEBUGDATA, this);
}
);
@@ -1290,8 +1155,8 @@ static std::string extract_canonical_hostname(std::string const& url)
void CurlEasyRequest::finalizeRequest(std::string const& url, AIHTTPTimeoutPolicy const& policy, AICurlEasyRequestStateMachine* state_machine)
{
DoutCurlEasyEntering("CurlEasyRequest::finalizeRequest(\"" << url << "\", " << policy.name() << ", " << (void*)state_machine << ")");
llassert(!mRequestFinalized);
DoutCurlEntering("CurlEasyRequest::finalizeRequest(\"" << url << "\", " << policy.name() << ", " << (void*)state_machine << ")");
llassert(!mTimeoutPolicy); // May only call finalizeRequest once!
mResult = CURLE_FAILED_INIT; // General error code; the final result code is stored here by MultiHandle::check_run_count when msg is CURLMSG_DONE.
#ifdef SHOW_ASSERT
// Do a sanity check on the headers.
@@ -1308,10 +1173,9 @@ void CurlEasyRequest::finalizeRequest(std::string const& url, AIHTTPTimeoutPolic
llwarns << "Requesting: \"" << url << "\": " << content_type_count << " Content-Type: headers!" << llendl;
}
#endif
mRequestFinalized = true;
setopt(CURLOPT_HTTPHEADER, mHeaders);
setoptString(CURLOPT_URL, url);
mTimeoutLowercaseHostname = extract_canonical_hostname(url);
mLowercaseHostname = extract_canonical_hostname(url);
mTimeoutPolicy = &policy;
state_machine->setTotalDelayTimeout(policy.getTotalDelay());
// The following line is a bit tricky: we store a pointer to the object without increasing its reference count.
@@ -1331,305 +1195,31 @@ void CurlEasyRequest::finalizeRequest(std::string const& url, AIHTTPTimeoutPolic
setopt(CURLOPT_PRIVATE, get_lockobj());
}
//.............................................................................
// HTTP Timeout stuff
// AIFIXME: Doing this only when it is actually being added assures that the first curl easy handle that is
// // being added for a particular host will be the one getting extra 'DNS lookup' connect time.
// // However, if another curl easy handle for the same host is added immediately after, it will
// // get less connect time, while it still (also) has to wait for this DNS lookup.
void CurlEasyRequest::set_timeout_opts(void)
{
setopt(CURLOPT_CONNECTTIMEOUT, mTimeoutPolicy->getConnectTimeout(mLowercaseHostname));
setopt(CURLOPT_TIMEOUT, mTimeoutPolicy->getCurlTransaction());
}
//static
F64 const CurlEasyRequest::sTimeoutClockWidth = 1.0 / calc_clock_frequency(); // Time between two clock ticks, in seconds.
U64 CurlEasyRequest::sTimeoutClockCount; // Clock count, set once per select() exit.
void CurlEasyRequest::timeout_timings(void)
void CurlEasyRequest::print_curl_timings(void) const
{
double t;
getinfo(CURLINFO_NAMELOOKUP_TIME, &t);
DoutCurlEasy("CURLINFO_NAMELOOKUP_TIME = " << t);
DoutCurl("CURLINFO_NAMELOOKUP_TIME = " << t);
getinfo(CURLINFO_CONNECT_TIME, &t);
DoutCurlEasy("CURLINFO_CONNECT_TIME = " << t);
DoutCurl("CURLINFO_CONNECT_TIME = " << t);
getinfo(CURLINFO_APPCONNECT_TIME, &t);
DoutCurlEasy("CURLINFO_APPCONNECT_TIME = " << t);
DoutCurl("CURLINFO_APPCONNECT_TIME = " << t);
getinfo(CURLINFO_PRETRANSFER_TIME, &t);
DoutCurlEasy("CURLINFO_PRETRANSFER_TIME = " << t);
DoutCurl("CURLINFO_PRETRANSFER_TIME = " << t);
getinfo(CURLINFO_STARTTRANSFER_TIME, &t);
DoutCurlEasy("CURLINFO_STARTTRANSFER_TIME = " << t);
DoutCurl("CURLINFO_STARTTRANSFER_TIME = " << t);
}
// CURL-THREAD
// This is called when the easy handle is actually being added to the multi handle (thus after being queued).
// AIFIXME: Doing this only when it is actually being added assures that the first curl easy handle that is
// being added for a particular host will be the one getting extra 'DNS lookup' connect time.
// However, if another curl easy handle for the same host is added immediately after, it will
// get less connect time, while it still (also) has to wait for this DNS lookup.
//
// <-----------------------------mTimeoutNothingReceivedYet-------------------------->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^
// |
void CurlEasyRequest::timeout_add_easy_request(void)
{
setopt(CURLOPT_CONNECTTIMEOUT, mTimeoutPolicy->getConnectTimeout(mTimeoutLowercaseHostname));
setopt(CURLOPT_TIMEOUT, mTimeoutPolicy->getCurlTransaction());
// This boolean is valid (only) if we get a time out event from libcurl.
mTimeoutLowSpeedOn = false;
mTimeoutNothingReceivedYet = true;
mTimeoutStalled = (U64)-1;
DoutCurlEasy("timeout_add_easy_request: mTimeoutStalled set to -1");
mTimeoutUploadFinished = false;
}
// CURL-THREAD
// This is called when body data was sent to the server socket.
// <--mTimeoutLowSpeedOn-->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^ ^ ^ ^ ^ ^
// | | | | | |
bool CurlEasyRequest::timeout_data_sent(size_t n)
{
// Generate events.
if (!mTimeoutLowSpeedOn)
{
// If we can send data (for the first time) then that's our only way to know we connected.
timeout_reset_lowspeed();
}
// Detect low speed.
return timeout_lowspeed(n);
}
// CURL-THREAD
// This is called when the 'low speed' timer should be started.
// <--mTimeoutLowSpeedOn--> <----mTimeoutLowSpeedOn---->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^ ^
// | |
void CurlEasyRequest::timeout_reset_lowspeed(void)
{
mTimeoutLowSpeedClock = sTimeoutClockCount;
mTimeoutLowSpeedOn = true;
mTimeoutLastSecond = -1; // This causes timeout_lowspeed to initialize the rest.
mTimeoutStalled = (U64)-1; // Stop reply delay timer.
DoutCurlEasy("timeout_reset_lowspeed: mTimeoutLowSpeedClock = " << mTimeoutLowSpeedClock << "; mTimeoutStalled = -1");
}
// CURL-THREAD
// This is called when everything we had to send to the server has been sent.
// <--mTimeoutLowSpeedOn-->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^
// |
void CurlEasyRequest::timeout_upload_finished(void)
{
if (1 || mTimeoutUploadFinished)
timeout_timings();
llassert(!mTimeoutUploadFinished); // If we get here twice, then the 'upload finished' detection failed.
mTimeoutUploadFinished = true;
// We finished uploading (if there was a body to upload at all), so not more transfer rate timeouts.
mTimeoutLowSpeedOn = false;
// Timeout if the server doesn't reply quick enough.
mTimeoutStalled = sTimeoutClockCount + mTimeoutPolicy->getReplyDelay() / sTimeoutClockWidth;
DoutCurlEasy("timeout_upload_finished: mTimeoutStalled set to sTimeoutClockCount (" << sTimeoutClockCount << ") + " << (mTimeoutStalled - sTimeoutClockCount) << " (" << mTimeoutPolicy->getReplyDelay() << " seconds)");
}
// CURL-THREAD
// This is called when data was received from the server.
//
// <-----------------------------mTimeoutNothingReceivedYet--------------------------><----mTimeoutLowSpeedOn---->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^ ^ ^ ^ ^ ^ ^ ^
// | | | | | | | |
bool CurlEasyRequest::timeout_data_received(size_t n)
{
// The HTTP header of the reply is the first thing we receive.
if (mTimeoutNothingReceivedYet && n > 0)
{
mTimeoutNothingReceivedYet = false;
if (!mTimeoutUploadFinished)
{
// mTimeoutUploadFinished not being set this point should only happen for GET requests (in fact, then it is normal),
// because in that case it is impossible to detect the difference between connecting and waiting for a reply without
// using CURLOPT_DEBUGFUNCTION. Note that mDebugIsGetMethod is only valid when the debug channel 'curlio' is on,
// because it is set in the debug callback function.
Debug(llassert(mDebugIsGetMethod || !dc::curlio.is_on()));
// 'Upload finished' detection failed, generate it now.
timeout_upload_finished();
}
// We received something; switch to getLowSpeedLimit()/getLowSpeedTime().
timeout_reset_lowspeed();
}
return mTimeoutLowSpeedOn ? timeout_lowspeed(n) : false;
}
// CURL_THREAD
// bytes is the number of bytes we just sent or received (including headers).
// Returns true if the transfer should be aborted.
//
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^
// | | | | | | | | | | | | | |
bool CurlEasyRequest::timeout_lowspeed(size_t bytes)
{
DoutCurlEasyEntering("CurlEasyRequest::timeout_lowspeed(" << bytes << ")");
// The algorithm to determine if we timed out if different from how libcurls CURLOPT_LOW_SPEED_TIME works.
//
// libcurl determines the transfer rate since the last call to an equivalent 'lowspeed' function, and then
// triggers a timeout if CURLOPT_LOW_SPEED_TIME long such a transfer value is less than CURLOPT_LOW_SPEED_LIMIT.
// That doesn't work right because once there IS data it can happen that this function is called a few
// times (with less than a milisecond in between) causing seemingly VERY high "transfer rate" spikes.
// The only correct way to determine the transfer rate is to actually average over CURLOPT_LOW_SPEED_TIME
// seconds.
//
// We do this as follows: we create low_speed_time (in seconds) buckets and fill them with the number
// of bytes received during that second. We also keep track of the sum of all bytes received between 'now'
// and 'now - llmax(starttime, low_speed_time)'. Then if that period reaches at least low_speed_time
// seconds, and the transfer rate (sum / low_speed_time) is less than low_speed_limit, we abort.
// When are we?
S32 second = (sTimeoutClockCount - mTimeoutLowSpeedClock) * sTimeoutClockWidth;
llassert(sTimeoutClockWidth > 0.0);
// This REALLY should never happen, but due to another bug it did happened
// and caused something so evil and hard to find that... NEVER AGAIN!
llassert(second >= 0);
// If this is the same second as last time, just add the number of bytes to the current bucket.
if (second == mTimeoutLastSecond)
{
mTimeoutTotalBytes += bytes;
mTimeoutBuckets[mTimeoutBucket] += bytes;
return false;
}
// We arrived at a new second.
// The below is at most executed once per second, even though for
// every currently connected transfer, CPU is not a big issue.
// Determine the number of buckets needed and increase the number of buckets if needed.
U16 const low_speed_time = mTimeoutPolicy->getLowSpeedTime();
if (low_speed_time > mTimeoutBuckets.size())
{
mTimeoutBuckets.resize(low_speed_time, 0);
}
S32 s = mTimeoutLastSecond;
mTimeoutLastSecond = second;
// If this is the first time this function is called, we need to do some initialization.
if (s == -1)
{
mTimeoutBucket = 0; // It doesn't really matter where we start.
mTimeoutTotalBytes = bytes;
mTimeoutBuckets[mTimeoutBucket] = bytes;
return false;
}
// Update all administration.
U16 bucket = mTimeoutBucket;
while(1) // Run over all the seconds that were skipped.
{
if (++bucket == low_speed_time)
bucket = 0;
if (++s == second)
break;
mTimeoutTotalBytes -= mTimeoutBuckets[bucket];
mTimeoutBuckets[bucket] = 0;
}
mTimeoutBucket = bucket;
mTimeoutTotalBytes -= mTimeoutBuckets[mTimeoutBucket];
mTimeoutTotalBytes += bytes;
mTimeoutBuckets[mTimeoutBucket] = bytes;
// Check if we timed out.
U32 const low_speed_limit = mTimeoutPolicy->getLowSpeedLimit();
U32 mintotalbytes = low_speed_limit * low_speed_time;
DoutCurlEasy("Transfered " << mTimeoutTotalBytes << " bytes in " << llmin(second, (S32)low_speed_time) << " seconds after " << second << " second" << ((second == 1) ? "" : "s") << ".");
if (second >= low_speed_time)
{
DoutCurlEasy("Average transfer rate is " << (mTimeoutTotalBytes / low_speed_time) << " bytes/s (low speed limit is " << low_speed_limit << " bytes/s)");
if (mTimeoutTotalBytes < mintotalbytes)
{
// The average transfer rate over the passed low_speed_time seconds is too low. Abort the transfer.
llwarns <<
#ifdef CWDEBUG
(void*)get_lockobj() << ": "
#endif
"aborting slow connection (average transfer rate below " << low_speed_limit <<
" for more than " << low_speed_time << " second" << ((low_speed_time == 1) ? "" : "s") << ")." << llendl;
return true;
}
}
// Calculate how long the data transfer may stall until we should timeout.
llassert_always(mintotalbytes > 0);
S32 max_stall_time = 0;
U32 dropped_bytes = 0;
while(1)
{
if (++bucket == low_speed_time) // The next second the next bucket will be emptied.
bucket = 0;
++max_stall_time;
dropped_bytes += mTimeoutBuckets[bucket];
// Note how, when max_stall_time == low_speed_time, dropped_bytes has
// to be equal to mTimeoutTotalBytes, the sum of all vector elements.
llassert_always(max_stall_time < low_speed_time || dropped_bytes == mTimeoutTotalBytes);
// And thus the following will certainly abort.
if (second + max_stall_time >= low_speed_time && mTimeoutTotalBytes - dropped_bytes < mintotalbytes)
break;
}
// If this function isn't called again within max_stall_time seconds, we stalled.
mTimeoutStalled = sTimeoutClockCount + max_stall_time / sTimeoutClockWidth;
DoutCurlEasy("mTimeoutStalled set to sTimeoutClockCount (" << sTimeoutClockCount << ") + " << (mTimeoutStalled - sTimeoutClockCount) << " (" << max_stall_time << " seconds)");
return false;
}
// CURL-THREAD
// This is called immediately before done() after curl finished, with code.
// <----mTimeoutLowSpeedOn---->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^
// |
void CurlEasyRequest::timeout_done(CURLcode code)
{
llassert(mTimeoutUploadFinished || mTimeoutNothingReceivedYet); // If this is false then the 'upload finished' detection failed.
if (code == CURLE_OPERATION_TIMEDOUT || code == CURLE_COULDNT_RESOLVE_HOST)
{
bool dns_problem = false;
if (code == CURLE_COULDNT_RESOLVE_HOST)
{
// Note that CURLINFO_OS_ERRNO returns 0; we don't know any more than this.
llwarns << "Failed to resolve hostname " << mTimeoutLowercaseHostname << llendl;
dns_problem = true;
}
else if (mTimeoutNothingReceivedYet)
{
// Only consider this to possibly be related to a DNS lookup if we didn't
// resolved the host yet, which can be detected by asking for
// CURLINFO_NAMELOOKUP_TIME which is set when libcurl initiates the
// actual connect and thus knows the IP# (possibly from it's DNS cache).
double namelookup_time;
getinfo(CURLINFO_NAMELOOKUP_TIME, &namelookup_time);
dns_problem = (namelookup_time == 0);
}
if (dns_problem)
{
// Inform policy object that there might be problems with resolving this host.
// This will increase the connect timeout the next time we try to connect to this host.
AIHTTPTimeoutPolicy::connect_timed_out(mTimeoutLowercaseHostname);
// AIFIXME: use return value to change priority
}
}
// Make sure no timeout will happen anymore.
mTimeoutLowSpeedOn = false;
mTimeoutStalled = (U64)-1;
DoutCurlEasy("timeout_done: mTimeoutStalled set to -1");
}
void CurlEasyRequest::timeout_print_diagnostics(AIHTTPTimeoutPolicy const& policy)
{
llwarns << "Request to " << mTimeoutLowercaseHostname << " timed out for " << policy.name() << llendl;
}
// End of HTTP Timeout stuff.
//.............................................................................
void CurlEasyRequest::getTransferInfo(AICurlInterface::TransferInfo* info)
{
// Curl explicitly demands a double for these info's.
@@ -1780,225 +1370,6 @@ void CurlResponderBuffer::prepRequest(AICurlEasyRequest_wat& curl_easy_request_w
curl_easy_request_w->addHeaders(headers);
}
//static
size_t CurlResponderBuffer::curlWriteCallback(char* data, size_t size, size_t nmemb, void* user_data)
{
ThreadSafeBufferedCurlEasyRequest* lockobj = static_cast<ThreadSafeBufferedCurlEasyRequest*>(user_data);
// We need to lock the curl easy request object too, because that lock is used
// to make sure that callbacks and destruction aren't done simultaneously.
AICurlEasyRequest_wat buffered_easy_request_w(*lockobj);
S32 bytes = size * nmemb; // The amount to write.
AICurlResponderBuffer_wat buffer_w(*lockobj);
// CurlResponderBuffer::setBodyLimit is never called, so buffer_w->mBodyLimit is infinite.
//S32 bytes = llmin(size * nmemb, buffer_w->mBodyLimit); buffer_w->mBodyLimit -= bytes;
buffer_w->getOutput()->append(sChannels.in(), (U8 const*)data, bytes);
buffer_w->mResponseTransferedBytes += bytes; // Accumulate data received from the server.
if (buffered_easy_request_w->timeout_data_received(bytes)) // Update timeout administration.
{
// Transfer timed out. Return 0 which will abort with error CURLE_WRITE_ERROR.
return 0;
}
return bytes;
}
//static
size_t CurlResponderBuffer::curlReadCallback(char* data, size_t size, size_t nmemb, void* user_data)
{
ThreadSafeBufferedCurlEasyRequest* lockobj = static_cast<ThreadSafeBufferedCurlEasyRequest*>(user_data);
// We need to lock the curl easy request object too, because that lock is used
// to make sure that callbacks and destruction aren't done simultaneously.
AICurlEasyRequest_wat buffered_easy_request_w(*lockobj);
S32 bytes = size * nmemb; // The maximum amount to read.
AICurlResponderBuffer_wat buffer_w(*lockobj);
buffer_w->mLastRead = buffer_w->getInput()->readAfter(sChannels.out(), buffer_w->mLastRead, (U8*)data, bytes);
buffer_w->mRequestTransferedBytes += bytes; // Accumulate data sent to the server.
if (buffered_easy_request_w->timeout_data_sent(bytes)) // Timeout administration.
{
// Transfer timed out. Return CURL_READFUNC_ABORT which will abort with error CURLE_ABORTED_BY_CALLBACK.
return CURL_READFUNC_ABORT;
}
return bytes; // Return the amount actually read (might be lowered by readAfter()).
}
//static
size_t CurlResponderBuffer::curlHeaderCallback(char* data, size_t size, size_t nmemb, void* user_data)
{
ThreadSafeBufferedCurlEasyRequest* lockobj = static_cast<ThreadSafeBufferedCurlEasyRequest*>(user_data);
// We need to lock the curl easy request object, because that lock is used
// to make sure that callbacks and destruction aren't done simultaneously.
AICurlEasyRequest_wat buffered_easy_request_w(*lockobj);
// This used to be headerCallback() in llurlrequest.cpp.
char const* const header_line = static_cast<char const*>(data);
size_t const header_len = size * nmemb;
if (buffered_easy_request_w->timeout_data_received(header_len)) // Update timeout administration.
{
// Transfer timed out. Return 0 which will abort with error CURLE_WRITE_ERROR.
return 0;
}
if (!header_len)
{
return header_len;
}
std::string header(header_line, header_len);
if (!LLStringUtil::_isASCII(header))
{
return header_len;
}
// Per HTTP spec the first header line must be the status line.
if (header.substr(0, 5) == "HTTP/")
{
std::string::iterator const begin = header.begin();
std::string::iterator const end = header.end();
std::string::iterator pos1 = std::find(begin, end, ' ');
if (pos1 != end) ++pos1;
std::string::iterator pos2 = std::find(pos1, end, ' ');
if (pos2 != end) ++pos2;
std::string::iterator pos3 = std::find(pos2, end, '\r');
U32 status;
std::string reason;
if (pos3 != end && std::isdigit(*pos1))
{
status = atoi(&header_line[pos1 - begin]);
reason.assign(pos2, pos3);
}
else
{
status = HTTP_INTERNAL_ERROR;
reason = "Header parse error.";
llwarns << "Received broken header line from server: \"" << header << "\"" << llendl;
}
{
AICurlResponderBuffer_wat curl_responder_buffer_w(*lockobj);
curl_responder_buffer_w->received_HTTP_header();
curl_responder_buffer_w->setStatusAndReason(status, reason);
}
return header_len;
}
std::string::iterator sep = std::find(header.begin(), header.end(), ':');
if (sep != header.end())
{
std::string key(header.begin(), sep);
std::string value(sep + 1, header.end());
key = utf8str_tolower(utf8str_trim(key));
value = utf8str_trim(value);
AICurlResponderBuffer_wat(*lockobj)->received_header(key, value);
}
else
{
LLStringUtil::trim(header);
if (!header.empty())
{
llwarns << "Unable to parse header: " << header << llendl;
}
}
return header_len;
}
void CurlResponderBuffer::setStatusAndReason(U32 status, std::string const& reason)
{
mStatus = status;
mReason = reason;
}
void CurlResponderBuffer::added_to_multi_handle(AICurlEasyRequest_wat& curl_easy_request_w)
{
llerrs << "Unexpected call to added_to_multi_handle()." << llendl;
}
void CurlResponderBuffer::finished(AICurlEasyRequest_wat& curl_easy_request_w)
{
llerrs << "Unexpected call to finished()." << llendl;
}
void CurlResponderBuffer::removed_from_multi_handle(AICurlEasyRequest_wat& curl_easy_request_w)
{
DoutCurlEasy("Calling CurlResponderBuffer::removed_from_multi_handle(@" << (void*)&*curl_easy_request_w << ") for this = " << (void*)this);
// Lock self.
ThreadSafeBufferedCurlEasyRequest* lockobj = get_lockobj();
llassert(dynamic_cast<ThreadSafeBufferedCurlEasyRequest*>(static_cast<ThreadSafeCurlEasyRequest*>(ThreadSafeCurlEasyRequest::wrapper_cast(&*curl_easy_request_w))) == lockobj);
AICurlResponderBuffer_wat buffer_w(*lockobj);
llassert(&*buffer_w == this);
processOutput(curl_easy_request_w);
}
void CurlResponderBuffer::processOutput(AICurlEasyRequest_wat& curl_easy_request_w)
{
U32 responseCode = 0;
std::string responseReason;
CURLcode code;
curl_easy_request_w->getResult(&code);
if (code == CURLE_OK)
{
curl_easy_request_w->getinfo(CURLINFO_RESPONSE_CODE, &responseCode);
// If getResult code is CURLE_OK then we should have decoded the first header line ourselves.
llassert(responseCode == mStatus);
if (responseCode == mStatus)
responseReason = mReason;
else
responseReason = "Unknown reason.";
}
else
{
responseCode = 499;
responseReason = AICurlInterface::strerror(code);
curl_easy_request_w->setopt(CURLOPT_FRESH_CONNECT, TRUE);
}
if (mResponder)
{
if (code == CURLE_OPERATION_TIMEDOUT)
{
curl_easy_request_w->timeout_print_diagnostics(mResponder->getHTTPTimeoutPolicy());
}
if (mEventsTarget)
{
// Only the responder registers for these events.
llassert(mEventsTarget == mResponder.get());
// Allow clients to parse headers before we attempt to parse
// the body and provide completed/result/error calls.
mEventsTarget->completed_headers(responseCode, responseReason);
}
mResponder->completedRaw(responseCode, responseReason, sChannels, mOutput);
mResponder = NULL;
}
resetState(curl_easy_request_w);
}
void CurlResponderBuffer::received_HTTP_header(void)
{
if (mEventsTarget)
mEventsTarget->received_HTTP_header();
}
void CurlResponderBuffer::received_header(std::string const& key, std::string const& value)
{
if (mEventsTarget)
mEventsTarget->received_header(key, value);
}
void CurlResponderBuffer::completed_headers(U32 status, std::string const& reason)
{
if (mEventsTarget)
mEventsTarget->completed_headers(status, reason);
}
//-----------------------------------------------------------------------------
// CurlMultiHandle

34
indra/aistatemachine/aicurl.h
View File

@@ -49,7 +49,7 @@
#undef CURLOPT_DNS_USE_GLOBAL_CACHE
#define CURLOPT_DNS_USE_GLOBAL_CACHE do_not_use_CURLOPT_DNS_USE_GLOBAL_CACHE
#include "stdtypes.h" // U32
#include "stdtypes.h" // U16, S32, U32, F64
#include "llatomic.h" // LLAtomicU32
#include "aithreadsafe.h"
#include "aihttpheaders.h"
@@ -59,6 +59,38 @@ class LLBufferArray;
class LLChannelDescriptors;
class AIHTTPTimeoutPolicy;
// Some pretty printing for curl easy handle related things:
// Print the lock object related to the current easy handle in every debug output.
#ifdef CWDEBUG
#include <libcwd/buf2str.h>
#include <sstream>
#define DoutCurl(x) do { \
using namespace libcwd; \
std::ostringstream marker; \
marker << (void*)this->get_lockobj(); \
libcw_do.push_marker(); \
libcw_do.marker().assign(marker.str().data(), marker.str().size()); \
libcw_do.inc_indent(2); \
Dout(dc::curl, x); \
libcw_do.dec_indent(2); \
libcw_do.pop_marker(); \
} while(0)
#define DoutCurlEntering(x) do { \
using namespace libcwd; \
std::ostringstream marker; \
marker << (void*)this->get_lockobj(); \
libcw_do.push_marker(); \
libcw_do.marker().assign(marker.str().data(), marker.str().size()); \
libcw_do.inc_indent(2); \
DoutEntering(dc::curl, x); \
libcw_do.dec_indent(2); \
libcw_do.pop_marker(); \
} while(0)
#else // !CWDEBUG
#define DoutCurl(x) Dout(dc::curl, x << " [" << (void*)this->get_lockobj() << ']')
#define DoutCurlEntering(x) DoutEntering(dc::curl, x << " [" << (void*)this->get_lockobj() << ']')
#endif // CWDEBUG
//-----------------------------------------------------------------------------
// Exceptions.
//

544
indra/aistatemachine/aicurlprivate.h
View File

@@ -33,323 +33,358 @@
#include <sstream>
#include "llatomic.h"
#include "llrefcount.h"
class AIHTTPHeaders;
class AIHTTPTimeoutPolicy;
class AICurlEasyRequestStateMachine;
namespace AICurlPrivate {
namespace curlthread { class MultiHandle; }
struct Stats {
static LLAtomicU32 easy_calls;
static LLAtomicU32 easy_errors;
static LLAtomicU32 easy_init_calls;
static LLAtomicU32 easy_init_errors;
static LLAtomicU32 easy_cleanup_calls;
static LLAtomicU32 multi_calls;
static LLAtomicU32 multi_errors;
class ThreadSafeCurlEasyRequest;
static void print(void);
};
namespace curlthread {
class MultiHandle;
void handle_multi_error(CURLMcode code);
inline CURLMcode check_multi_code(CURLMcode code) { Stats::multi_calls++; if (code != CURLM_OK) handle_multi_error(code); return code; }
// A class that keeps track of timeout administration per connection.
class HTTPTimeout : public LLRefCount {
private:
AIHTTPTimeoutPolicy const* mPolicy; // A pointer to the used timeout policy.
std::vector<U32> mBuckets; // An array with the number of bytes transfered in each second.
U16 mBucket; // The bucket corresponding to mLastSecond.
bool mNothingReceivedYet; // Set when created, reset when the HTML reply header from the server is received.
bool mLowSpeedOn; // Set while uploading or downloading data.
bool mUploadFinished; // Used to keep track of whether upload_finished was called yet.
S32 mLastSecond; // The time at which lowspeed() was last called, in seconds since mLowSpeedClock.
U32 mTotalBytes; // The sum of all bytes in mBuckets.
U64 mLowSpeedClock; // Clock count at which low speed detection (re)started.
U64 mStalled; // The clock count at which this transaction is considered to be stalling if nothing is transfered anymore.
public:
static F64 const sClockWidth; // Time between two clock ticks in seconds.
static U64 sClockCount; // Clock count used as 'now' during one loop of the main loop.
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
ThreadSafeCurlEasyRequest* mLockObj;
#endif
bool curlThreadIsRunning(void);
void wakeUpCurlThread(void);
void stopCurlThread(void);
public:
HTTPTimeout(AIHTTPTimeoutPolicy const* policy, ThreadSafeCurlEasyRequest* lock_obj) :
mPolicy(policy), mNothingReceivedYet(true), mLowSpeedOn(false), mUploadFinished(false), mStalled((U64)-1)
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
, mLockObj(lock_obj)
#endif
{ }
class ThreadSafeCurlEasyRequest;
class ThreadSafeBufferedCurlEasyRequest;
// Called after sending all headers, when body data is written the first time.
void connected(void);
// Called when everything we had to send to the server has been sent.
void upload_finished(void);
// Called when data is sent. Returns true if transfer timed out.
bool data_sent(size_t n);
// Called when data is received. Returns true if transfer timed out.
bool data_received(size_t n);
// Called immediately before done() after curl finished, with code.
void done(AICurlEasyRequest_wat const& curlEasyRequest_w, CURLcode code);
// Accessor.
bool has_stalled(void) const { return mStalled < sClockCount; }
// Called from CurlResponderBuffer::processOutput if a timeout occurred.
void print_diagnostics(AICurlEasyRequest_wat const& curlEasyRequest_w);
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
void* get_lockobj(void) const { return mLockObj; }
#endif
private:
// (Re)start low speed transer rate detection.
void reset_lowspeed(void);
// Common low speed detection, Called from data_sent or data_received.
bool lowspeed(size_t bytes);
};
} // namespace curlthread
struct Stats {
static LLAtomicU32 easy_calls;
static LLAtomicU32 easy_errors;
static LLAtomicU32 easy_init_calls;
static LLAtomicU32 easy_init_errors;
static LLAtomicU32 easy_cleanup_calls;
static LLAtomicU32 multi_calls;
static LLAtomicU32 multi_errors;
static void print(void);
};
void handle_multi_error(CURLMcode code);
inline CURLMcode check_multi_code(CURLMcode code) { Stats::multi_calls++; if (code != CURLM_OK) handle_multi_error(code); return code; }
bool curlThreadIsRunning(void);
void wakeUpCurlThread(void);
void stopCurlThread(void);
class ThreadSafeCurlEasyRequest;
class ThreadSafeBufferedCurlEasyRequest;
#define DECLARE_SETOPT(param_type) \
CURLcode setopt(CURLoption option, param_type parameter)
// This class wraps CURL*'s.
// It guarantees that a pointer is cleaned up when no longer needed, as required by libcurl.
class CurlEasyHandle : public boost::noncopyable, protected AICurlEasyHandleEvents {
public:
CurlEasyHandle(void);
~CurlEasyHandle();
// This class wraps CURL*'s.
// It guarantees that a pointer is cleaned up when no longer needed, as required by libcurl.
class CurlEasyHandle : public boost::noncopyable, protected AICurlEasyHandleEvents {
public:
CurlEasyHandle(void);
~CurlEasyHandle();
private:
// Disallow assignment.
CurlEasyHandle& operator=(CurlEasyHandle const*);
private:
// Disallow assignment.
CurlEasyHandle& operator=(CurlEasyHandle const*);
public:
// Reset all options of a libcurl session handle.
void reset(void) { llassert(!mActiveMultiHandle); curl_easy_reset(mEasyHandle); }
public:
// Reset all options of a libcurl session handle.
void reset(void) { llassert(!mActiveMultiHandle); curl_easy_reset(mEasyHandle); }
// Set options for a curl easy handle.
DECLARE_SETOPT(long);
DECLARE_SETOPT(long long);
DECLARE_SETOPT(void const*);
DECLARE_SETOPT(curl_debug_callback);
DECLARE_SETOPT(curl_write_callback);
//DECLARE_SETOPT(curl_read_callback); Same type as curl_write_callback
DECLARE_SETOPT(curl_ssl_ctx_callback);
DECLARE_SETOPT(curl_conv_callback);
// Set options for a curl easy handle.
DECLARE_SETOPT(long);
DECLARE_SETOPT(long long);
DECLARE_SETOPT(void const*);
DECLARE_SETOPT(curl_debug_callback);
DECLARE_SETOPT(curl_write_callback);
//DECLARE_SETOPT(curl_read_callback); Same type as curl_write_callback
DECLARE_SETOPT(curl_ssl_ctx_callback);
DECLARE_SETOPT(curl_conv_callback);
#if 0 // Not used by the viewer.
DECLARE_SETOPT(curl_progress_callback);
DECLARE_SETOPT(curl_seek_callback);
DECLARE_SETOPT(curl_ioctl_callback);
DECLARE_SETOPT(curl_sockopt_callback);
DECLARE_SETOPT(curl_opensocket_callback);
DECLARE_SETOPT(curl_closesocket_callback);
DECLARE_SETOPT(curl_sshkeycallback);
DECLARE_SETOPT(curl_chunk_bgn_callback);
DECLARE_SETOPT(curl_chunk_end_callback);
DECLARE_SETOPT(curl_fnmatch_callback);
DECLARE_SETOPT(curl_progress_callback);
DECLARE_SETOPT(curl_seek_callback);
DECLARE_SETOPT(curl_ioctl_callback);
DECLARE_SETOPT(curl_sockopt_callback);
DECLARE_SETOPT(curl_opensocket_callback);
DECLARE_SETOPT(curl_closesocket_callback);
DECLARE_SETOPT(curl_sshkeycallback);
DECLARE_SETOPT(curl_chunk_bgn_callback);
DECLARE_SETOPT(curl_chunk_end_callback);
DECLARE_SETOPT(curl_fnmatch_callback);
#endif
// Automatically cast int types to a long. Note that U32/S32 are int and
// that you can overload int and long even if they have the same size.
CURLcode setopt(CURLoption option, U32 parameter) { return setopt(option, (long)parameter); }
CURLcode setopt(CURLoption option, S32 parameter) { return setopt(option, (long)parameter); }
// Automatically cast int types to a long. Note that U32/S32 are int and
// that you can overload int and long even if they have the same size.
CURLcode setopt(CURLoption option, U32 parameter) { return setopt(option, (long)parameter); }
CURLcode setopt(CURLoption option, S32 parameter) { return setopt(option, (long)parameter); }
// Clone a libcurl session handle using all the options previously set.
//CurlEasyHandle(CurlEasyHandle const& orig);
// Clone a libcurl session handle using all the options previously set.
//CurlEasyHandle(CurlEasyHandle const& orig);
// URL encode/decode the given string.
char* escape(char* url, int length);
char* unescape(char* url, int inlength , int* outlength);
// URL encode/decode the given string.
char* escape(char* url, int length);
char* unescape(char* url, int inlength , int* outlength);
// Extract information from a curl handle.
private:
CURLcode getinfo_priv(CURLINFO info, void* data) const;
public:
// The rest are inlines to provide some type-safety.
CURLcode getinfo(CURLINFO info, char** data) const { return getinfo_priv(info, data); }
CURLcode getinfo(CURLINFO info, curl_slist** data) const { return getinfo_priv(info, data); }
CURLcode getinfo(CURLINFO info, double* data) const { return getinfo_priv(info, data); }
CURLcode getinfo(CURLINFO info, long* data) const { return getinfo_priv(info, data); }
// Extract information from a curl handle.
private:
CURLcode getinfo_priv(CURLINFO info, void* data) const;
public:
// The rest are inlines to provide some type-safety.
CURLcode getinfo(CURLINFO info, char** data) const { return getinfo_priv(info, data); }
CURLcode getinfo(CURLINFO info, curl_slist** data) const { return getinfo_priv(info, data); }
CURLcode getinfo(CURLINFO info, double* data) const { return getinfo_priv(info, data); }
CURLcode getinfo(CURLINFO info, long* data) const { return getinfo_priv(info, data); }
#ifdef __LP64__ // sizeof(long) > sizeof(int) ?
// Overload for integer types that are too small (libcurl demands a long).
CURLcode getinfo(CURLINFO info, S32* data) const { long ldata; CURLcode res = getinfo_priv(info, &ldata); *data = static_cast<S32>(ldata); return res; }
CURLcode getinfo(CURLINFO info, U32* data) const { long ldata; CURLcode res = getinfo_priv(info, &ldata); *data = static_cast<U32>(ldata); return res; }
// Overload for integer types that are too small (libcurl demands a long).
CURLcode getinfo(CURLINFO info, S32* data) const { long ldata; CURLcode res = getinfo_priv(info, &ldata); *data = static_cast<S32>(ldata); return res; }
CURLcode getinfo(CURLINFO info, U32* data) const { long ldata; CURLcode res = getinfo_priv(info, &ldata); *data = static_cast<U32>(ldata); return res; }
#else // sizeof(long) == sizeof(int)
CURLcode getinfo(CURLINFO info, S32* data) const { return getinfo_priv(info, reinterpret_cast<long*>(data)); }
CURLcode getinfo(CURLINFO info, U32* data) const { return getinfo_priv(info, reinterpret_cast<long*>(data)); }
CURLcode getinfo(CURLINFO info, S32* data) const { return getinfo_priv(info, reinterpret_cast<long*>(data)); }
CURLcode getinfo(CURLINFO info, U32* data) const { return getinfo_priv(info, reinterpret_cast<long*>(data)); }
#endif
// Perform a file transfer (blocking).
CURLcode perform(void);
// Pause and unpause a connection.
CURLcode pause(int bitmask);
// Perform a file transfer (blocking).
CURLcode perform(void);
// Pause and unpause a connection.
CURLcode pause(int bitmask);
// Called when a request is queued for removal. In that case a race between the actual removal
// and revoking of the callbacks is harmless (and happens for the raw non-statemachine version).
void remove_queued(void) { mQueuedForRemoval = true; }
// In case it's added after being removed.
void add_queued(void) { mQueuedForRemoval = false; }
// Called when a request is queued for removal. In that case a race between the actual removal
// and revoking of the callbacks is harmless (and happens for the raw non-statemachine version).
void remove_queued(void) { mQueuedForRemoval = true; }
// In case it's added after being removed.
void add_queued(void) { mQueuedForRemoval = false; }
private:
CURL* mEasyHandle;
CURLM* mActiveMultiHandle;
mutable char* mErrorBuffer;
AIPostFieldPtr mPostField; // This keeps the POSTFIELD data alive for as long as the easy handle exists.
bool mQueuedForRemoval; // Set if the easy handle is (probably) added to the multi handle, but is queued for removal.
private:
CURL* mEasyHandle;
CURLM* mActiveMultiHandle;
mutable char* mErrorBuffer;
AIPostFieldPtr mPostField; // This keeps the POSTFIELD data alive for as long as the easy handle exists.
bool mQueuedForRemoval; // Set if the easy handle is (probably) added to the multi handle, but is queued for removal.
#ifdef SHOW_ASSERT
public:
bool mRemovedPerCommand; // Set if mActiveMultiHandle was reset as per command from the main thread.
public:
bool mRemovedPerCommand; // Set if mActiveMultiHandle was reset as per command from the main thread.
#endif
private:
// This should only be called from MultiHandle; add/remove an easy handle to/from a multi handle.
friend class curlthread::MultiHandle;
CURLMcode add_handle_to_multi(AICurlEasyRequest_wat& curl_easy_request_w, CURLM* multi_handle);
CURLMcode remove_handle_from_multi(AICurlEasyRequest_wat& curl_easy_request_w, CURLM* multi_handle);
private:
// This should only be called from MultiHandle; add/remove an easy handle to/from a multi handle.
friend class curlthread::MultiHandle;
CURLMcode add_handle_to_multi(AICurlEasyRequest_wat& curl_easy_request_w, CURLM* multi_handle);
CURLMcode remove_handle_from_multi(AICurlEasyRequest_wat& curl_easy_request_w, CURLM* multi_handle);
public:
// Returns true if this easy handle was added to a curl multi handle.
bool active(void) const { return mActiveMultiHandle; }
public:
// Returns true if this easy handle was added to a curl multi handle.
bool active(void) const { return mActiveMultiHandle; }
// Returns true when it is expected that the parent will revoke callbacks before the curl
// easy handle is removed from the multi handle; that usually happens when an external
// error demands termination of the request (ie, an expiration).
bool no_warning(void) const { return mQueuedForRemoval || LLApp::isExiting(); }
// Returns true when it is expected that the parent will revoke callbacks before the curl
// easy handle is removed from the multi handle; that usually happens when an external
// error demands termination of the request (ie, an expiration).
bool no_warning(void) const { return mQueuedForRemoval || LLApp::isExiting(); }
// Used for debugging purposes.
bool operator==(CURL* easy_handle) const { return mEasyHandle == easy_handle; }
// Used for debugging purposes.
bool operator==(CURL* easy_handle) const { return mEasyHandle == easy_handle; }
private:
// Call this prior to every curl_easy function whose return value is passed to check_easy_code.
void setErrorBuffer(void) const;
private:
// Call this prior to every curl_easy function whose return value is passed to check_easy_code.
void setErrorBuffer(void) const;
static void handle_easy_error(CURLcode code);
static void handle_easy_error(CURLcode code);
// Always first call setErrorBuffer()!
static inline CURLcode check_easy_code(CURLcode code)
{
Stats::easy_calls++;
if (code != CURLE_OK)
handle_easy_error(code);
return code;
}
// Always first call setErrorBuffer()!
static inline CURLcode check_easy_code(CURLcode code)
{
Stats::easy_calls++;
if (code != CURLE_OK)
handle_easy_error(code);
return code;
}
protected:
// Return the underlying curl easy handle.
CURL* getEasyHandle(void) const { return mEasyHandle; }
protected:
// Return the underlying curl easy handle.
CURL* getEasyHandle(void) const { return mEasyHandle; }
// Keep POSTFIELD data alive.
void setPostField(AIPostFieldPtr const& post_field_ptr) { mPostField = post_field_ptr; }
// Keep POSTFIELD data alive.
void setPostField(AIPostFieldPtr const& post_field_ptr) { mPostField = post_field_ptr; }
private:
// Return, and possibly create, the curl (easy) error buffer used by the current thread.
static char* getTLErrorBuffer(void);
};
private:
// Return, and possibly create, the curl (easy) error buffer used by the current thread.
static char* getTLErrorBuffer(void);
};
// CurlEasyRequest adds a slightly more powerful interface that can be used
// to set the options on a curl easy handle.
//
// Calling sendRequest() will then connect to the given URL and perform
// the data exchange. Use getResult() to determine if an error occurred.
//
// Note that the life cycle of a CurlEasyRequest is controlled by AICurlEasyRequest:
// a CurlEasyRequest is only ever created as base class of a ThreadSafeCurlEasyRequest,
// which is only created by creating a AICurlEasyRequest. When the last copy of such
// AICurlEasyRequest is deleted, then also the ThreadSafeCurlEasyRequest is deleted
// and the CurlEasyRequest destructed.
class CurlEasyRequest : public CurlEasyHandle {
private:
void setPost_raw(U32 size, char const* data);
public:
void setPost(U32 size) { setPost_raw(size, NULL); }
void setPost(AIPostFieldPtr const& postdata, U32 size);
void setPost(char const* data, U32 size) { setPost(new AIPostField(data), size); }
void setoptString(CURLoption option, std::string const& value);
void addHeader(char const* str);
void addHeaders(AIHTTPHeaders const& headers);
// CurlEasyRequest adds a slightly more powerful interface that can be used
// to set the options on a curl easy handle.
//
// Calling sendRequest() will then connect to the given URL and perform
// the data exchange. Use getResult() to determine if an error occurred.
//
// Note that the life cycle of a CurlEasyRequest is controlled by AICurlEasyRequest:
// a CurlEasyRequest is only ever created as base class of a ThreadSafeCurlEasyRequest,
// which is only created by creating a AICurlEasyRequest. When the last copy of such
// AICurlEasyRequest is deleted, then also the ThreadSafeCurlEasyRequest is deleted
// and the CurlEasyRequest destructed.
class CurlEasyRequest : public CurlEasyHandle {
private:
void setPost_raw(U32 size, char const* data);
public:
void setPost(U32 size) { setPost_raw(size, NULL); }
void setPost(AIPostFieldPtr const& postdata, U32 size);
void setPost(char const* data, U32 size) { setPost(new AIPostField(data), size); }
void setoptString(CURLoption option, std::string const& value);
void addHeader(char const* str);
void addHeaders(AIHTTPHeaders const& headers);
private:
// Callback stubs.
static size_t headerCallback(char* ptr, size_t size, size_t nmemb, void* userdata);
static size_t writeCallback(char* ptr, size_t size, size_t nmemb, void* userdata);
static size_t readCallback(char* ptr, size_t size, size_t nmemb, void* userdata);
static CURLcode SSLCtxCallback(CURL* curl, void* sslctx, void* userdata);
private:
// Callback stubs.
static size_t headerCallback(char* ptr, size_t size, size_t nmemb, void* userdata);
static size_t writeCallback(char* ptr, size_t size, size_t nmemb, void* userdata);
static size_t readCallback(char* ptr, size_t size, size_t nmemb, void* userdata);
static CURLcode SSLCtxCallback(CURL* curl, void* sslctx, void* userdata);
curl_write_callback mHeaderCallback;
void* mHeaderCallbackUserData;
curl_write_callback mWriteCallback;
void* mWriteCallbackUserData;
curl_read_callback mReadCallback;
void* mReadCallbackUserData;
curl_ssl_ctx_callback mSSLCtxCallback;
void* mSSLCtxCallbackUserData;
curl_write_callback mHeaderCallback;
void* mHeaderCallbackUserData;
curl_write_callback mWriteCallback;
void* mWriteCallbackUserData;
curl_read_callback mReadCallback;
void* mReadCallbackUserData;
curl_ssl_ctx_callback mSSLCtxCallback;
void* mSSLCtxCallbackUserData;
public:
void setHeaderCallback(curl_write_callback callback, void* userdata);
void setWriteCallback(curl_write_callback callback, void* userdata);
void setReadCallback(curl_read_callback callback, void* userdata);
void setSSLCtxCallback(curl_ssl_ctx_callback callback, void* userdata);
public:
void setHeaderCallback(curl_write_callback callback, void* userdata);
void setWriteCallback(curl_write_callback callback, void* userdata);
void setReadCallback(curl_read_callback callback, void* userdata);
void setSSLCtxCallback(curl_ssl_ctx_callback callback, void* userdata);
// Call this if the set callbacks are about to be invalidated.
void revokeCallbacks(void);
// Call this if the set callbacks are about to be invalidated.
void revokeCallbacks(void);
// Reset everything to the state it was in when this object was just created.
void resetState(void);
// Reset everything to the state it was in when this object was just created.
void resetState(void);
private:
// Called from applyDefaultOptions.
void applyProxySettings(void);
private:
// Called from applyDefaultOptions.
void applyProxySettings(void);
// Used in applyDefaultOptions.
static CURLcode curlCtxCallback(CURL* curl, void* sslctx, void* parm);
// Used in applyDefaultOptions.
static CURLcode curlCtxCallback(CURL* curl, void* sslctx, void* parm);
public:
// Set default options that we want applied to all curl easy handles.
void applyDefaultOptions(void);
public:
// Set default options that we want applied to all curl easy handles.
void applyDefaultOptions(void);
// Prepare the request for adding it to a multi session, or calling perform.
// This actually adds the headers that were collected with addHeader.
void finalizeRequest(std::string const& url, AIHTTPTimeoutPolicy const& policy, AICurlEasyRequestStateMachine* state_machine);
// Prepare the request for adding it to a multi session, or calling perform.
// This actually adds the headers that were collected with addHeader.
void finalizeRequest(std::string const& url, AIHTTPTimeoutPolicy const& policy, AICurlEasyRequestStateMachine* state_machine);
//-------------------------------------------
// Timeout administration events:
// Last second initialization. Called from MultiHandle::add_easy_request.
void set_timeout_opts(void);
// Called by MultiHandle::add_easy_request when the easy handle is actually being added to the multi handle.
void timeout_add_easy_request(void);
public:
// Called by MultiHandle::check_run_count() to store result code that is returned by getResult.
void storeResult(CURLcode result) { mResult = result; }
// Called after sending all headers, when body data is written the first time.
void timeout_connected(void);
// Called by MultiHandle::check_run_count() when the curl easy handle is done.
void done(AICurlEasyRequest_wat& curl_easy_request_w) { finished(curl_easy_request_w); }
// Called when everything we had to send to the server has been sent.
void timeout_upload_finished(void);
// Called by MultiHandle::check_run_count() to fill info with the transfer info.
void getTransferInfo(AICurlInterface::TransferInfo* info);
// Called when data is sent. Returns true if transfer timed out.
bool timeout_data_sent(size_t n);
// If result != CURLE_FAILED_INIT then also info was filled.
void getResult(CURLcode* result, AICurlInterface::TransferInfo* info = NULL);
// Called when data is received. Returns true if transfer timed out.
bool timeout_data_received(size_t n);
// For debugging purposes.
void print_curl_timings(void) const;
// Called immediately before done() after curl finished, with code.
void timeout_done(CURLcode code);
private:
curl_slist* mHeaders;
AICurlEasyHandleEvents* mEventsTarget;
CURLcode mResult;
// Accessor.
bool timeout_has_stalled(void) const { return mTimeoutStalled < sTimeoutClockCount; }
// Called from CurlResponderBuffer::processOutput if a timeout occurred.
void timeout_print_diagnostics(AIHTTPTimeoutPolicy const& policy);
private:
// (Re)start low speed transer rate detection.
void timeout_reset_lowspeed(void);
// Common low speed detection, Called from timeout_data_sent or timeout_data_received.
bool timeout_lowspeed(size_t bytes);
// End of timeout stuff
//-------------------------------------------
public:
// Called by MultiHandle::check_run_count() to store result code that is returned by getResult.
void storeResult(CURLcode result) { mResult = result; }
// Called by MultiHandle::check_run_count() when the curl easy handle is done.
void done(AICurlEasyRequest_wat& curl_easy_request_w) { finished(curl_easy_request_w); }
// Called by MultiHandle::check_run_count() to fill info with the transfer info.
void getTransferInfo(AICurlInterface::TransferInfo* info);
// If result != CURLE_FAILED_INIT then also info was filled.
void getResult(CURLcode* result, AICurlInterface::TransferInfo* info = NULL);
private:
curl_slist* mHeaders;
bool mRequestFinalized;
AICurlEasyHandleEvents* mEventsTarget;
CURLcode mResult;
// AIFIXME: put all timeout stuff in it's own class.
AIHTTPTimeoutPolicy const* mTimeoutPolicy;
std::string mTimeoutLowercaseHostname; // Lowercase hostname (canonicalized) extracted from the url.
std::vector<U32> mTimeoutBuckets; // An array with the number of bytes transfered in each second.
U16 mTimeoutBucket; // The bucket corresponding to mTimeoutLastSecond.
bool mTimeoutNothingReceivedYet; // Set when connected, reset when the HTML reply header from the server is received.
bool mTimeoutLowSpeedOn; // Set while uploading or downloading data.
bool mTimeoutUploadFinished; // Only used for debugging.
S32 mTimeoutLastSecond; // The time at which timeout_lowspeed was last called, in seconds since mTimeoutLowSpeedClock.
U32 mTimeoutTotalBytes; // The sum of all bytes in mTimeoutBuckets.
U64 mTimeoutLowSpeedClock; // Clock count at which low speed transfer rate started.
U64 mTimeoutStalled; // The clock count at which this transaction is considered to be stalling.
public:
static F64 const sTimeoutClockWidth; // Time between two clock ticks in seconds.
static U64 sTimeoutClockCount; // Clock count used during one loop of the main loop.
AIHTTPTimeoutPolicy const* mTimeoutPolicy;
std::string mLowercaseHostname; // Lowercase hostname (canonicalized) extracted from the url.
LLPointer<curlthread::HTTPTimeout> mTimeout;
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
bool mDebugIsGetMethod;
public:
bool mDebugIsGetMethod;
#endif
private:
// This class may only be created by constructing a ThreadSafeCurlEasyRequest.
friend class ThreadSafeCurlEasyRequest;
// Throws AICurlNoEasyHandle.
CurlEasyRequest(void) :
mHeaders(NULL), mRequestFinalized(false), mEventsTarget(NULL), mResult(CURLE_FAILED_INIT),
public:
// These two are only valid after finalizeRequest.
AIHTTPTimeoutPolicy const* getTimeoutPolicy(void) const { return mTimeoutPolicy; }
std::string const& getLowercaseHostname(void) const { return mLowercaseHostname; }
// Called by CurlSocketInfo to allow access to the last (after a redirect) HTTPTimeout object related to this request.
void set_timeout_object(LLPointer<curlthread::HTTPTimeout>& timeout) { mTimeout = timeout; }
// And the accessor for it.
LLPointer<curlthread::HTTPTimeout>& httptimeout(void) { return mTimeout; }
// Return true if no data has been received on the latest socket (if any) for too long.
bool has_stalled(void) const { return mTimeout && mTimeout->has_stalled(); }
private:
// This class may only be created by constructing a ThreadSafeCurlEasyRequest.
friend class ThreadSafeCurlEasyRequest;
// Throws AICurlNoEasyHandle.
CurlEasyRequest(void) : mHeaders(NULL), mEventsTarget(NULL), mResult(CURLE_FAILED_INIT), mTimeoutPolicy(NULL)
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
mDebugIsGetMethod(false),
, mDebugIsGetMethod(false)
#endif
mTimeoutPolicy(NULL)
{ applyDefaultOptions(); }
{ applyDefaultOptions(); }
public:
~CurlEasyRequest();
@@ -358,8 +393,7 @@ namespace AICurlPrivate {
void send_events_to(AICurlEasyHandleEvents* target) { mEventsTarget = target; }
// For debugging purposes
bool is_finalized(void) const { return mRequestFinalized; }
void timeout_timings(void);
bool is_finalized(void) const { return mTimeoutPolicy; }
// Return pointer to the ThreadSafe (wrapped) version of this object.
ThreadSafeCurlEasyRequest* get_lockobj(void);

615
indra/aistatemachine/aicurlthread.cpp
View File

@@ -30,7 +30,10 @@
#include "linden_common.h"
#include "aicurlthread.h"
#include "aihttptimeoutpolicy.h"
#include "lltimer.h" // ms_sleep, get_clock_count
#include "llhttpstatuscodes.h"
#include "llbuffer.h"
#include <sys/types.h>
#if !LL_WINDOWS
#include <sys/select.h>
@@ -753,6 +756,7 @@ class CurlSocketInfo
curl_socket_t mSocketFd;
int mAction;
AICurlEasyRequest mEasyRequest;
LLPointer<HTTPTimeout> mTimeout;
};
CurlSocketInfo::CurlSocketInfo(MultiHandle& multi_handle, CURL* easy, curl_socket_t s, int action, ThreadSafeCurlEasyRequest* lockobj) :
@@ -763,6 +767,14 @@ CurlSocketInfo::CurlSocketInfo(MultiHandle& multi_handle, CURL* easy, curl_socke
llassert(!mMultiHandle.mReadPollSet->contains(s));
llassert(!mMultiHandle.mWritePollSet->contains(s));
set_action(action);
// Create a new HTTPTimeout object and keep a pointer to it in the corresponding CurlEasyRequest object.
// The reason for this seemingly redundant storage (we could just store it directly in the CurlEasyRequest
// and not in CurlSocketInfo) is because in the case of a redirection there exist temporarily two
// CurlSocketInfo objects for a request and we need upload_finished() to be called on the HTTPTimeout
// object related to THIS CurlSocketInfo.
AICurlEasyRequest_wat easy_request_w(*lockobj);
mTimeout = new HTTPTimeout(easy_request_w->getTimeoutPolicy(), lockobj);
easy_request_w->set_timeout_object(mTimeout);
}
CurlSocketInfo::~CurlSocketInfo()
@@ -798,7 +810,7 @@ void CurlSocketInfo::set_action(int action)
if (pretransfer_time > 0)
{
// If CURL_POLL_OUT is removed and CURLINFO_PRETRANSFER_TIME is already set, then we have nothing more to send apparently.
curl_easy_request_w->timeout_upload_finished(); // Update timeout administration.
mTimeout->upload_finished(); // Update timeout administration.
}
}
}
@@ -1369,8 +1381,8 @@ void AICurlThread::run(void)
continue;
}
// Clock count used for timeouts.
CurlEasyRequest::sTimeoutClockCount = get_clock_count();
Dout(dc::curl, "CurlEasyRequest::sTimeoutClockCount = " << CurlEasyRequest::sTimeoutClockCount);
HTTPTimeout::sClockCount = get_clock_count();
Dout(dc::curl, "HTTPTimeout::sClockCount = " << HTTPTimeout::sClockCount);
if (ready == 0)
{
multi_handle_w->socket_action(CURL_SOCKET_TIMEOUT, 0);
@@ -1437,7 +1449,7 @@ void MultiHandle::handle_stalls(void)
{
for(addedEasyRequests_type::iterator iter = mAddedEasyRequests.begin(); iter != mAddedEasyRequests.end();)
{
if (AICurlEasyRequest_wat(**iter)->timeout_has_stalled())
if (AICurlEasyRequest_wat(**iter)->has_stalled())
{
Dout(dc::curl, "MultiHandle::handle_stalls(): Easy request stalled! [" << (void*)iter->get_ptr().get() << "]");
finish_easy_request(*iter, CURLE_OPERATION_TIMEDOUT);
@@ -1526,7 +1538,7 @@ void MultiHandle::add_easy_request(AICurlEasyRequest const& easy_request)
CURLMcode ret;
{
AICurlEasyRequest_wat curl_easy_request_w(*easy_request);
curl_easy_request_w->timeout_add_easy_request();
curl_easy_request_w->set_timeout_opts();
ret = curl_easy_request_w->add_handle_to_multi(curl_easy_request_w, mMultiHandle);
}
if (ret == CURLM_OK)
@@ -1705,11 +1717,272 @@ void MultiHandle::finish_easy_request(AICurlEasyRequest const& easy_request, CUR
". [CURLINFO_PRIVATE = " << (void*)easy_request.get_ptr().get() << "]");
#endif
// Update timeout administration.
curl_easy_request_w->timeout_done(result);
curl_easy_request_w->httptimeout()->done(curl_easy_request_w, result);
// Signal that this easy handle finished.
curl_easy_request_w->done(curl_easy_request_w);
}
//-----------------------------------------------------------------------------
// HTTPTimeout
//static
F64 const HTTPTimeout::sClockWidth = 1.0 / calc_clock_frequency(); // Time between two clock ticks, in seconds.
U64 HTTPTimeout::sClockCount; // Clock count, set once per select() exit.
// CURL-THREAD
// This is called when body data was sent to the server socket.
// <-----mLowSpeedOn------>
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^ ^ ^ ^ ^ ^
// | | | | | |
bool HTTPTimeout::data_sent(size_t n)
{
// Generate events.
if (!mLowSpeedOn)
{
// If we can send data (for the first time) then that's our only way to know we connected.
reset_lowspeed();
}
// Detect low speed.
return lowspeed(n);
}
// CURL-THREAD
// This is called when the 'low speed' timer should be started.
// <-----mLowSpeedOn------> <-------mLowSpeedOn-------->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^ ^
// | |
void HTTPTimeout::reset_lowspeed(void)
{
mLowSpeedClock = sClockCount;
mLowSpeedOn = true;
mLastSecond = -1; // This causes lowspeed to initialize the rest.
mStalled = (U64)-1; // Stop reply delay timer.
DoutCurl("reset_lowspeed: mLowSpeedClock = " << mLowSpeedClock << "; mStalled = -1");
}
// CURL-THREAD
// This is called when everything we had to send to the server has been sent.
// <-----mLowSpeedOn------>
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^
// |
void HTTPTimeout::upload_finished(void)
{
llassert(!mUploadFinished); // If we get here twice, then the 'upload finished' detection failed.
mUploadFinished = true;
// We finished uploading (if there was a body to upload at all), so not more transfer rate timeouts.
mLowSpeedOn = false;
// Timeout if the server doesn't reply quick enough.
mStalled = sClockCount + mPolicy->getReplyDelay() / sClockWidth;
DoutCurl("upload_finished: mStalled set to sClockCount (" << sClockCount << ") + " << (mStalled - sClockCount) << " (" << mPolicy->getReplyDelay() << " seconds)");
}
// CURL-THREAD
// This is called when data was received from the server.
//
// <--------------------------------mNothingReceivedYet------------------------------><-------mLowSpeedOn-------->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^ ^ ^ ^ ^ ^ ^ ^
// | | | | | | | |
bool HTTPTimeout::data_received(size_t n)
{
// The HTTP header of the reply is the first thing we receive.
if (mNothingReceivedYet && n > 0)
{
if (!mUploadFinished)
{
// mUploadFinished not being set this point should only happen for GET requests (in fact, then it is normal),
// because in that case it is impossible to detect the difference between connecting and waiting for a reply without
// using CURLOPT_DEBUGFUNCTION. Note that mDebugIsGetMethod is only valid when the debug channel 'curlio' is on,
// because it is set in the debug callback function.
Debug(llassert(AICurlEasyRequest_wat(*mLockObj)->mDebugIsGetMethod || !dc::curlio.is_on()));
// 'Upload finished' detection failed, generate it now.
upload_finished();
}
// Turn this flag off again now that we received data, so that if 'upload_finished()' is called again
// for a future upload on the same descriptor, then that won't trigger an assert.
// Note that because we also set mNothingReceivedYet here, we won't enter this code block anymore,
// so it's safe to do this.
mUploadFinished = false;
// Mark that something was received.
mNothingReceivedYet = false;
// We received something; switch to getLowSpeedLimit()/getLowSpeedTime().
reset_lowspeed();
}
return mLowSpeedOn ? lowspeed(n) : false;
}
// CURL_THREAD
// bytes is the number of bytes we just sent or received (including headers).
// Returns true if the transfer should be aborted.
//
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^
// | | | | | | | | | | | | | |
bool HTTPTimeout::lowspeed(size_t bytes)
{
DoutCurlEntering("HTTPTimeout::lowspeed(" << bytes << ")");
// The algorithm to determine if we timed out if different from how libcurls CURLOPT_LOW_SPEED_TIME works.
//
// libcurl determines the transfer rate since the last call to an equivalent 'lowspeed' function, and then
// triggers a timeout if CURLOPT_LOW_SPEED_TIME long such a transfer value is less than CURLOPT_LOW_SPEED_LIMIT.
// That doesn't work right because once there IS data it can happen that this function is called a few
// times (with less than a milisecond in between) causing seemingly VERY high "transfer rate" spikes.
// The only correct way to determine the transfer rate is to actually average over CURLOPT_LOW_SPEED_TIME
// seconds.
//
// We do this as follows: we create low_speed_time (in seconds) buckets and fill them with the number
// of bytes received during that second. We also keep track of the sum of all bytes received between 'now'
// and 'now - llmax(starttime, low_speed_time)'. Then if that period reaches at least low_speed_time
// seconds, and the transfer rate (sum / low_speed_time) is less than low_speed_limit, we abort.
// When are we?
S32 second = (sClockCount - mLowSpeedClock) * sClockWidth;
llassert(sClockWidth > 0.0);
// This REALLY should never happen, but due to another bug it did happened
// and caused something so evil and hard to find that... NEVER AGAIN!
llassert(second >= 0);
// If this is the same second as last time, just add the number of bytes to the current bucket.
if (second == mLastSecond)
{
mTotalBytes += bytes;
mBuckets[mBucket] += bytes;
return false;
}
// We arrived at a new second.
// The below is at most executed once per second, even though for
// every currently connected transfer, CPU is not a big issue.
// Determine the number of buckets needed and increase the number of buckets if needed.
U16 const low_speed_time = mPolicy->getLowSpeedTime();
if (low_speed_time > mBuckets.size())
{
mBuckets.resize(low_speed_time, 0);
}
S32 s = mLastSecond;
mLastSecond = second;
// If this is the first time this function is called, we need to do some initialization.
if (s == -1)
{
mBucket = 0; // It doesn't really matter where we start.
mTotalBytes = bytes;
mBuckets[mBucket] = bytes;
return false;
}
// Update all administration.
U16 bucket = mBucket;
while(1) // Run over all the seconds that were skipped.
{
if (++bucket == low_speed_time)
bucket = 0;
if (++s == second)
break;
mTotalBytes -= mBuckets[bucket];
mBuckets[bucket] = 0;
}
mBucket = bucket;
mTotalBytes -= mBuckets[mBucket];
mTotalBytes += bytes;
mBuckets[mBucket] = bytes;
// Check if we timed out.
U32 const low_speed_limit = mPolicy->getLowSpeedLimit();
U32 mintotalbytes = low_speed_limit * low_speed_time;
DoutCurl("Transfered " << mTotalBytes << " bytes in " << llmin(second, (S32)low_speed_time) << " seconds after " << second << " second" << ((second == 1) ? "" : "s") << ".");
if (second >= low_speed_time)
{
DoutCurl("Average transfer rate is " << (mTotalBytes / low_speed_time) << " bytes/s (low speed limit is " << low_speed_limit << " bytes/s)");
if (mTotalBytes < mintotalbytes)
{
// The average transfer rate over the passed low_speed_time seconds is too low. Abort the transfer.
llwarns <<
#ifdef CWDEBUG
(void*)get_lockobj() << ": "
#endif
"aborting slow connection (average transfer rate below " << low_speed_limit <<
" for more than " << low_speed_time << " second" << ((low_speed_time == 1) ? "" : "s") << ")." << llendl;
return true;
}
}
// Calculate how long the data transfer may stall until we should timeout.
llassert_always(mintotalbytes > 0);
S32 max_stall_time = 0;
U32 dropped_bytes = 0;
while(1)
{
if (++bucket == low_speed_time) // The next second the next bucket will be emptied.
bucket = 0;
++max_stall_time;
dropped_bytes += mBuckets[bucket];
// Note how, when max_stall_time == low_speed_time, dropped_bytes has
// to be equal to mTotalBytes, the sum of all vector elements.
llassert_always(max_stall_time < low_speed_time || dropped_bytes == mTotalBytes);
// And thus the following will certainly abort.
if (second + max_stall_time >= low_speed_time && mTotalBytes - dropped_bytes < mintotalbytes)
break;
}
// If this function isn't called again within max_stall_time seconds, we stalled.
mStalled = sClockCount + max_stall_time / sClockWidth;
DoutCurl("mStalled set to sClockCount (" << sClockCount << ") + " << (mStalled - sClockCount) << " (" << max_stall_time << " seconds)");
return false;
}
// CURL-THREAD
// This is called immediately before done() after curl finished, with code.
// <-------mLowSpeedOn-------->
// queued--><--DNS lookup + connect + send headers-->[<--send body (if any)-->]<--replydelay--><--receive headers + body--><--done
// ^
// |
void HTTPTimeout::done(AICurlEasyRequest_wat const& curlEasyRequest_w, CURLcode code)
{
if (code == CURLE_OPERATION_TIMEDOUT || code == CURLE_COULDNT_RESOLVE_HOST)
{
bool dns_problem = false;
if (code == CURLE_COULDNT_RESOLVE_HOST)
{
// Note that CURLINFO_OS_ERRNO returns 0; we don't know any more than this.
llwarns << "Failed to resolve hostname " << curlEasyRequest_w->getLowercaseHostname() << llendl;
dns_problem = true;
}
else if (mNothingReceivedYet)
{
// Only consider this to possibly be related to a DNS lookup if we didn't
// resolved the host yet, which can be detected by asking for
// CURLINFO_NAMELOOKUP_TIME which is set when libcurl initiates the
// actual connect and thus knows the IP# (possibly from it's DNS cache).
double namelookup_time;
curlEasyRequest_w->getinfo(CURLINFO_NAMELOOKUP_TIME, &namelookup_time);
dns_problem = (namelookup_time == 0);
}
if (dns_problem)
{
// Inform policy object that there might be problems with resolving this host.
// This will increase the connect timeout the next time we try to connect to this host.
AIHTTPTimeoutPolicy::connect_timed_out(curlEasyRequest_w->getLowercaseHostname());
// AIFIXME: use return value to change priority
}
}
// Make sure no timeout will happen anymore.
mLowSpeedOn = false;
mStalled = (U64)-1;
DoutCurl("done: mStalled set to -1");
}
void HTTPTimeout::print_diagnostics(AICurlEasyRequest_wat const& curlEasyRequest_w)
{
llwarns << "Request to " << curlEasyRequest_w->getLowercaseHostname() << " timed out for " << curlEasyRequest_w->getTimeoutPolicy()->name() << llendl;
}
} // namespace curlthread
} // namespace AICurlPrivate
@@ -1770,6 +2043,336 @@ void stopCurlThread(void)
}
}
//-----------------------------------------------------------------------------
// CurlResponderBuffer
void CurlResponderBuffer::setStatusAndReason(U32 status, std::string const& reason)
{
mStatus = status;
mReason = reason;
}
void CurlResponderBuffer::added_to_multi_handle(AICurlEasyRequest_wat& curl_easy_request_w)
{
llerrs << "Unexpected call to added_to_multi_handle()." << llendl;
}
void CurlResponderBuffer::finished(AICurlEasyRequest_wat& curl_easy_request_w)
{
llerrs << "Unexpected call to finished()." << llendl;
}
void CurlResponderBuffer::removed_from_multi_handle(AICurlEasyRequest_wat& curl_easy_request_w)
{
DoutCurl("Calling CurlResponderBuffer::removed_from_multi_handle(@" << (void*)&*curl_easy_request_w << ") for this = " << (void*)this);
// Lock self.
ThreadSafeBufferedCurlEasyRequest* lockobj = get_lockobj();
llassert(dynamic_cast<ThreadSafeBufferedCurlEasyRequest*>(static_cast<ThreadSafeCurlEasyRequest*>(ThreadSafeCurlEasyRequest::wrapper_cast(&*curl_easy_request_w))) == lockobj);
AICurlResponderBuffer_wat buffer_w(*lockobj);
llassert(&*buffer_w == this);
processOutput(curl_easy_request_w);
}
void CurlResponderBuffer::processOutput(AICurlEasyRequest_wat& curl_easy_request_w)
{
U32 responseCode = 0;
std::string responseReason;
CURLcode code;
curl_easy_request_w->getResult(&code);
if (code == CURLE_OK)
{
curl_easy_request_w->getinfo(CURLINFO_RESPONSE_CODE, &responseCode);
// If getResult code is CURLE_OK then we should have decoded the first header line ourselves.
llassert(responseCode == mStatus);
if (responseCode == mStatus)
responseReason = mReason;
else
responseReason = "Unknown reason.";
}
else
{
responseCode = 499;
responseReason = AICurlInterface::strerror(code);
curl_easy_request_w->setopt(CURLOPT_FRESH_CONNECT, TRUE);
}
if (mResponder)
{
if (code == CURLE_OPERATION_TIMEDOUT)
{
curl_easy_request_w->httptimeout()->print_diagnostics(curl_easy_request_w);
}
if (mEventsTarget)
{
// Only the responder registers for these events.
llassert(mEventsTarget == mResponder.get());
// Allow clients to parse headers before we attempt to parse
// the body and provide completed/result/error calls.
mEventsTarget->completed_headers(responseCode, responseReason);
}
mResponder->completedRaw(responseCode, responseReason, sChannels, mOutput);
mResponder = NULL;
}
resetState(curl_easy_request_w);
}
void CurlResponderBuffer::received_HTTP_header(void)
{
if (mEventsTarget)
mEventsTarget->received_HTTP_header();
}
void CurlResponderBuffer::received_header(std::string const& key, std::string const& value)
{
if (mEventsTarget)
mEventsTarget->received_header(key, value);
}
void CurlResponderBuffer::completed_headers(U32 status, std::string const& reason)
{
if (mEventsTarget)
mEventsTarget->completed_headers(status, reason);
}
//static
size_t CurlResponderBuffer::curlWriteCallback(char* data, size_t size, size_t nmemb, void* user_data)
{
ThreadSafeBufferedCurlEasyRequest* lockobj = static_cast<ThreadSafeBufferedCurlEasyRequest*>(user_data);
// We need to lock the curl easy request object too, because that lock is used
// to make sure that callbacks and destruction aren't done simultaneously.
AICurlEasyRequest_wat buffered_easy_request_w(*lockobj);
S32 bytes = size * nmemb; // The amount to write.
AICurlResponderBuffer_wat buffer_w(*lockobj);
// CurlResponderBuffer::setBodyLimit is never called, so buffer_w->mBodyLimit is infinite.
//S32 bytes = llmin(size * nmemb, buffer_w->mBodyLimit); buffer_w->mBodyLimit -= bytes;
buffer_w->getOutput()->append(sChannels.in(), (U8 const*)data, bytes);
buffer_w->mResponseTransferedBytes += bytes; // Accumulate data received from the server.
if (buffered_easy_request_w->httptimeout()->data_received(bytes)) // Update timeout administration.
{
// Transfer timed out. Return 0 which will abort with error CURLE_WRITE_ERROR.
return 0;
}
return bytes;
}
//static
size_t CurlResponderBuffer::curlReadCallback(char* data, size_t size, size_t nmemb, void* user_data)
{
ThreadSafeBufferedCurlEasyRequest* lockobj = static_cast<ThreadSafeBufferedCurlEasyRequest*>(user_data);
// We need to lock the curl easy request object too, because that lock is used
// to make sure that callbacks and destruction aren't done simultaneously.
AICurlEasyRequest_wat buffered_easy_request_w(*lockobj);
S32 bytes = size * nmemb; // The maximum amount to read.
AICurlResponderBuffer_wat buffer_w(*lockobj);
buffer_w->mLastRead = buffer_w->getInput()->readAfter(sChannels.out(), buffer_w->mLastRead, (U8*)data, bytes);
buffer_w->mRequestTransferedBytes += bytes; // Accumulate data sent to the server.
if (buffered_easy_request_w->httptimeout()->data_sent(bytes)) // Timeout administration.
{
// Transfer timed out. Return CURL_READFUNC_ABORT which will abort with error CURLE_ABORTED_BY_CALLBACK.
return CURL_READFUNC_ABORT;
}
return bytes; // Return the amount actually read (might be lowered by readAfter()).
}
//static
size_t CurlResponderBuffer::curlHeaderCallback(char* data, size_t size, size_t nmemb, void* user_data)
{
ThreadSafeBufferedCurlEasyRequest* lockobj = static_cast<ThreadSafeBufferedCurlEasyRequest*>(user_data);
// We need to lock the curl easy request object, because that lock is used
// to make sure that callbacks and destruction aren't done simultaneously.
AICurlEasyRequest_wat buffered_easy_request_w(*lockobj);
// This used to be headerCallback() in llurlrequest.cpp.
char const* const header_line = static_cast<char const*>(data);
size_t const header_len = size * nmemb;
if (buffered_easy_request_w->httptimeout()->data_received(header_len)) // Update timeout administration.
{
// Transfer timed out. Return 0 which will abort with error CURLE_WRITE_ERROR.
return 0;
}
if (!header_len)
{
return header_len;
}
std::string header(header_line, header_len);
if (!LLStringUtil::_isASCII(header))
{
return header_len;
}
// Per HTTP spec the first header line must be the status line.
if (header.substr(0, 5) == "HTTP/")
{
std::string::iterator const begin = header.begin();
std::string::iterator const end = header.end();
std::string::iterator pos1 = std::find(begin, end, ' ');
if (pos1 != end) ++pos1;
std::string::iterator pos2 = std::find(pos1, end, ' ');
if (pos2 != end) ++pos2;
std::string::iterator pos3 = std::find(pos2, end, '\r');
U32 status;
std::string reason;
if (pos3 != end && std::isdigit(*pos1))
{
status = atoi(&header_line[pos1 - begin]);
reason.assign(pos2, pos3);
}
else
{
status = HTTP_INTERNAL_ERROR;
reason = "Header parse error.";
llwarns << "Received broken header line from server: \"" << header << "\"" << llendl;
}
{
AICurlResponderBuffer_wat curl_responder_buffer_w(*lockobj);
curl_responder_buffer_w->received_HTTP_header();
curl_responder_buffer_w->setStatusAndReason(status, reason);
}
return header_len;
}
std::string::iterator sep = std::find(header.begin(), header.end(), ':');
if (sep != header.end())
{
std::string key(header.begin(), sep);
std::string value(sep + 1, header.end());
key = utf8str_tolower(utf8str_trim(key));
value = utf8str_trim(value);
AICurlResponderBuffer_wat(*lockobj)->received_header(key, value);
}
else
{
LLStringUtil::trim(header);
if (!header.empty())
{
llwarns << "Unable to parse header: " << header << llendl;
}
}
return header_len;
}
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
int debug_callback(CURL*, curl_infotype infotype, char* buf, size_t size, void* user_ptr)
{
#ifdef CWDEBUG
using namespace ::libcwd;
CurlEasyRequest* request = (CurlEasyRequest*)user_ptr;
std::ostringstream marker;
marker << (void*)request->get_lockobj();
libcw_do.push_marker();
libcw_do.marker().assign(marker.str().data(), marker.str().size());
if (!debug::channels::dc::curlio.is_on())
debug::channels::dc::curlio.on();
LibcwDoutScopeBegin(LIBCWD_DEBUGCHANNELS, libcw_do, dc::curlio|cond_nonewline_cf(infotype == CURLINFO_TEXT))
#else
if (infotype == CURLINFO_TEXT)
{
while (size > 0 && (buf[size - 1] == '\r' || buf[size - 1] == '\n'))
--size;
}
LibcwDoutScopeBegin(LIBCWD_DEBUGCHANNELS, libcw_do, dc::curlio)
#endif
switch (infotype)
{
case CURLINFO_TEXT:
LibcwDoutStream << "* ";
break;
case CURLINFO_HEADER_IN:
LibcwDoutStream << "H> ";
break;
case CURLINFO_HEADER_OUT:
LibcwDoutStream << "H< ";
if (size >= 4 && strncmp(buf, "GET ", 4) == 0)
request->mDebugIsGetMethod = true;
break;
case CURLINFO_DATA_IN:
LibcwDoutStream << "D> ";
break;
case CURLINFO_DATA_OUT:
LibcwDoutStream << "D< ";
break;
case CURLINFO_SSL_DATA_IN:
LibcwDoutStream << "S> ";
break;
case CURLINFO_SSL_DATA_OUT:
LibcwDoutStream << "S< ";
break;
default:
LibcwDoutStream << "?? ";
}
if (infotype == CURLINFO_TEXT)
LibcwDoutStream.write(buf, size);
else if (infotype == CURLINFO_HEADER_IN || infotype == CURLINFO_HEADER_OUT)
LibcwDoutStream << libcwd::buf2str(buf, size);
else if (infotype == CURLINFO_DATA_IN)
{
LibcwDoutStream << size << " bytes";
bool finished = false;
size_t i = 0;
while (i < size)
{
char c = buf[i];
if (!('0' <= c && c <= '9') && !('a' <= c && c <= 'f'))
{
if (0 < i && i + 1 < size && buf[i] == '\r' && buf[i + 1] == '\n')
{
// Binary output: "[0-9a-f]*\r\n ...binary data..."
LibcwDoutStream << ": \"" << libcwd::buf2str(buf, i + 2) << "\"...";
finished = true;
}
break;
}
++i;
}
if (!finished && size > 9 && buf[0] == '<')
{
// Human readable output: html, xml or llsd.
if (!strncmp(buf, "<!DOCTYPE", 9) || !strncmp(buf, "<?xml", 5) || !strncmp(buf, "<llsd>", 6))
{
LibcwDoutStream << ": \"" << libcwd::buf2str(buf, size) << '"';
finished = true;
}
}
if (!finished)
{
// Unknown format. Only print the first and last 20 characters.
if (size > 40UL)
{
LibcwDoutStream << ": \"" << libcwd::buf2str(buf, 20) << "\"...\"" << libcwd::buf2str(&buf[size - 20], 20) << '"';
}
else
{
LibcwDoutStream << ": \"" << libcwd::buf2str(buf, size) << '"';
}
}
}
else if (infotype == CURLINFO_DATA_OUT)
LibcwDoutStream << size << " bytes: \"" << libcwd::buf2str(buf, size) << '"';
else
LibcwDoutStream << size << " bytes";
LibcwDoutScopeEnd;
#ifdef CWDEBUG
libcw_do.pop_marker();
#endif
return 0;
}
#endif // defined(CWDEBUG) || defined(DEBUG_CURLIO)
} // namespace AICurlPrivate
//-----------------------------------------------------------------------------