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

Added CurlUseMultipleThreads (requires restart to change)

Browse Source
This commit is contained in:
Shyotl
2011-08-06 02:29:04 -05:00
parent f1759e0a96
commit 9aa26648c9
4 changed files with 126 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
indra/llmessage/llcurl.cpp
View File

@@ -92,6 +92,9 @@ std::vector<LLMutex*> LLCurl::sSSLMutex;
std::string LLCurl::sCAPath;
std::string LLCurl::sCAFile;
bool LLCurl::sMultiThreaded = false;
static U32 sMainThreadID = 0;
void check_curl_code(CURLcode code)
{
if (code != CURLE_OK)
@@ -250,7 +253,7 @@ public:
U32 report(CURLcode);
void getTransferInfo(LLCurl::TransferInfo* info);
void prepRequest(const std::string& url, const std::vector<std::string>& headers, ResponderPtr, bool post = false);
void prepRequest(const std::string& url, const std::vector<std::string>& headers, ResponderPtr, S32 time_out = 0, bool post = false);
const char* getErrorBuffer();
@@ -549,7 +552,7 @@ size_t curlHeaderCallback(void* data, size_t size, size_t nmemb, void* user_data
void LLCurl::Easy::prepRequest(const std::string& url,
const std::vector<std::string>& headers,
ResponderPtr responder, bool post)
ResponderPtr responder, S32 time_out, bool post)
{
resetState();
@@ -599,7 +602,7 @@ void LLCurl::Easy::prepRequest(const std::string& url,
//don't verify host name so urls with scrubbed host names will work (improves DNS performance)
setopt(CURLOPT_SSL_VERIFYHOST, 0);
setopt(CURLOPT_TIMEOUT, CURL_REQUEST_TIMEOUT);
setopt(CURLOPT_TIMEOUT, llmax(time_out, CURL_REQUEST_TIMEOUT));
setoptString(CURLOPT_URL, url);
@@ -642,6 +645,7 @@ public:
S32 process();
void perform();
void doPerform();
virtual void run();
@@ -654,6 +658,7 @@ public:
LLCondition* mSignal;
bool mQuitting;
bool mThreaded;
private:
void easyFree(Easy*);
@@ -675,7 +680,16 @@ LLCurl::Multi::Multi()
mPerformState(PERFORM_STATE_READY)
{
mQuitting = false;
mSignal = new LLCondition;
mThreaded = LLCurl::sMultiThreaded && LLThread::currentID() == sMainThreadID;
if (mThreaded)
{
mSignal = new LLCondition;
}
else
{
mSignal = NULL;
}
mCurlMultiHandle = curl_multi_init();
if (!mCurlMultiHandle)
@@ -722,16 +736,25 @@ CURLMsg* LLCurl::Multi::info_read(S32* msgs_in_queue)
void LLCurl::Multi::perform()
{
mSignal->lock();
if (mPerformState == PERFORM_STATE_READY)
if (mThreaded)
{
mSignal->signal();
mSignal->lock();
if (mPerformState == PERFORM_STATE_READY)
{
mSignal->signal();
}
mSignal->unlock();
}
mSignal->unlock();
else
{
doPerform();
}
}
void LLCurl::Multi::run()
{
llassert(mThreaded);
mSignal->lock();
while (!mQuitting)
{
@@ -739,26 +762,31 @@ void LLCurl::Multi::run()
mPerformState = PERFORM_STATE_PERFORMING;
if (!mQuitting)
{
S32 q = 0;
for (S32 call_count = 0;
call_count < MULTI_PERFORM_CALL_REPEAT;
call_count += 1)
{
CURLMcode code = curl_multi_perform(mCurlMultiHandle, &q);
if (CURLM_CALL_MULTI_PERFORM != code || q == 0)
{
check_curl_multi_code(code);
break;
}
}
mQueued = q;
mPerformState = PERFORM_STATE_COMPLETED;
doPerform();
}
}
mSignal->unlock();
}
void LLCurl::Multi::doPerform()
{
S32 q = 0;
for (S32 call_count = 0;
call_count < MULTI_PERFORM_CALL_REPEAT;
call_count += 1)
{
CURLMcode code = curl_multi_perform(mCurlMultiHandle, &q);
if (CURLM_CALL_MULTI_PERFORM != code || q == 0)
{
check_curl_multi_code(code);
break;
}
}
mQueued = q;
mPerformState = PERFORM_STATE_COMPLETED;
}
S32 LLCurl::Multi::process()
{
perform();
@@ -883,16 +911,21 @@ LLCurlRequest::~LLCurlRequest()
for (curlmulti_set_t::iterator iter = mMultiSet.begin(); iter != mMultiSet.end(); ++iter)
{
LLCurl::Multi* multi = *iter;
multi->mSignal->lock();
multi->mQuitting = true;
while (!multi->isStopped())
{
multi->mSignal->signal();
multi->mSignal->unlock();
apr_sleep(1000);
if (multi->mThreaded)
multi->mSignal->lock();
multi->mQuitting = true;
if (multi->mThreaded)
{
while (!multi->isStopped())
{
multi->mSignal->signal();
multi->mSignal->unlock();
apr_sleep(1000);
multi->mSignal->lock();
}
}
multi->mSignal->unlock();
if (multi->mThreaded)
multi->mSignal->unlock();
}
for_each(mMultiSet.begin(), mMultiSet.end(), DeletePointer());
}
@@ -901,7 +934,10 @@ void LLCurlRequest::addMulti()
{
llassert_always(mThreadID == LLThread::currentID());
LLCurl::Multi* multi = new LLCurl::Multi();
multi->start();
if (multi->mThreaded)
{
multi->start();
}
mMultiSet.insert(multi);
mActiveMulti = multi;
mActiveRequestCount = 0;
@@ -963,14 +999,14 @@ bool LLCurlRequest::getByteRange(const std::string& url,
bool LLCurlRequest::post(const std::string& url,
const headers_t& headers,
const LLSD& data,
LLCurl::ResponderPtr responder)
LLCurl::ResponderPtr responder, S32 time_out)
{
LLCurl::Easy* easy = allocEasy();
if (!easy)
{
return false;
}
easy->prepRequest(url, headers, responder);
easy->prepRequest(url, headers, responder, time_out);
LLSDSerialize::toXML(data, easy->getInput());
S32 bytes = easy->getInput().str().length();
@@ -990,14 +1026,14 @@ bool LLCurlRequest::post(const std::string& url,
bool LLCurlRequest::post(const std::string& url,
const headers_t& headers,
const std::string& data,
LLCurl::ResponderPtr responder)
LLCurl::ResponderPtr responder, S32 time_out)
{
LLCurl::Easy* easy = allocEasy();
if (!easy)
{
return false;
}
easy->prepRequest(url, headers, responder);
easy->prepRequest(url, headers, responder, time_out);
easy->getInput().write(data.data(), data.size());
S32 bytes = easy->getInput().str().length();
@@ -1031,16 +1067,21 @@ S32 LLCurlRequest::process()
if (multi != mActiveMulti && tres == 0 && multi->mQueued == 0)
{
mMultiSet.erase(curiter);
multi->mSignal->lock();
if (multi->mThreaded)
multi->mSignal->lock();
multi->mQuitting = true;
while (!multi->isStopped())
if (multi->mThreaded)
{
multi->mSignal->signal();
multi->mSignal->unlock();
apr_sleep(1000);
multi->mSignal->unlock();
while (!multi->isStopped())
{
multi->mSignal->signal();
multi->mSignal->unlock();
apr_sleep(1000);
multi->mSignal->unlock();
}
}
multi->mSignal->unlock();
if (multi->mThreaded)
multi->mSignal->unlock();
delete multi;
}
@@ -1059,6 +1100,10 @@ S32 LLCurlRequest::getQueued()
curlmulti_set_t::iterator curiter = iter++;
LLCurl::Multi* multi = *curiter;
queued += multi->mQueued;
if (multi->mPerformState != LLCurl::Multi::PERFORM_STATE_READY)
{
++queued;
}
}
return queued;
}
@@ -1072,7 +1117,10 @@ LLCurlEasyRequest::LLCurlEasyRequest()
mResultReturned(false)
{
mMulti = new LLCurl::Multi();
mMulti->start();
if (mMulti->mThreaded)
{
mMulti->start();
}
mEasy = mMulti->allocEasy();
if (mEasy)
{
@@ -1083,16 +1131,21 @@ LLCurlEasyRequest::LLCurlEasyRequest()
LLCurlEasyRequest::~LLCurlEasyRequest()
{
mMulti->mSignal->lock();
mMulti->mQuitting = true;
while (!mMulti->isStopped())
{
mMulti->mSignal->signal();
mMulti->mSignal->unlock();
apr_sleep(1000);
if (mMulti->mThreaded)
mMulti->mSignal->lock();
mMulti->mQuitting = true;
if (mMulti->mThreaded)
{
while (!mMulti->isStopped())
{
mMulti->mSignal->signal();
mMulti->mSignal->unlock();
apr_sleep(1000);
mMulti->mSignal->lock();
}
}
mMulti->mSignal->unlock();
if (mMulti->mThreaded)
mMulti->mSignal->unlock();
delete mMulti;
}
@@ -1287,8 +1340,10 @@ unsigned long LLCurl::ssl_thread_id(void)
}
#endif
void LLCurl::initClass()
void LLCurl::initClass(bool multi_threaded)
{
sMainThreadID = LLThread::currentID();
sMultiThreaded = multi_threaded;
// Do not change this "unless you are familiar with and mean to control
// internal operations of libcurl"
// - http://curl.haxx.se/libcurl/c/curl_global_init.html