diff options
| -rw-r--r-- | indra/llmessage/llcurl.cpp | 356 | ||||
| -rwxr-xr-x | indra/llmessage/llcurl.h | 105 | ||||
| -rw-r--r-- | indra/llmessage/llurlrequest.cpp | 6 | ||||
| -rwxr-xr-x | indra/newview/llappviewer.cpp | 5 | ||||
| -rw-r--r-- | indra/newview/llxmlrpctransaction.cpp | 16 |
5 files changed, 308 insertions, 180 deletions
diff --git a/indra/llmessage/llcurl.cpp b/indra/llmessage/llcurl.cpp index 330028c926..7f61e1ac04 100644 --- a/indra/llmessage/llcurl.cpp +++ b/indra/llmessage/llcurl.cpp @@ -86,9 +86,7 @@ S32 gCurlMultiCount = 0; std::vector<LLMutex*> LLCurl::sSSLMutex; std::string LLCurl::sCAPath; std::string LLCurl::sCAFile; - -bool LLCurl::sMultiThreaded = false; -static U32 sMainThreadID = 0; +LLCurlThread* LLCurl::sCurlThread = NULL ; void check_curl_code(CURLcode code) { @@ -221,14 +219,11 @@ namespace boost std::set<CURL*> LLCurl::Easy::sFreeHandles; std::set<CURL*> LLCurl::Easy::sActiveHandles; -LLMutex* LLCurl::Easy::sHandleMutex = NULL; -LLMutex* LLCurl::Easy::sMultiMutex = NULL; //static CURL* LLCurl::Easy::allocEasyHandle() { CURL* ret = NULL; - LLMutexLock lock(sHandleMutex); if (sFreeHandles.empty()) { ret = curl_easy_init(); @@ -256,8 +251,6 @@ void LLCurl::Easy::releaseEasyHandle(CURL* handle) llerrs << "handle cannot be NULL!" << llendl; } - LLMutexLock lock(sHandleMutex); - if (sActiveHandles.find(handle) != sActiveHandles.end()) { sActiveHandles.erase(handle); @@ -521,23 +514,13 @@ void LLCurl::Easy::prepRequest(const std::string& url, //////////////////////////////////////////////////////////////////////////// LLCurl::Multi::Multi() - : LLThread("Curl Multi"), - mQueued(0), + : mQueued(0), mErrorCount(0), - mPerformState(PERFORM_STATE_READY) + mState(STATE_READY), + mDead(FALSE), + mMutexp(NULL), + mDeletionMutexp(NULL) { - mQuitting = false; - - mThreaded = LLCurl::sMultiThreaded && LLThread::currentID() == sMainThreadID; - if (mThreaded) - { - mSignal = new LLCondition(NULL); - } - else - { - mSignal = NULL; - } - mCurlMultiHandle = curl_multi_init(); if (!mCurlMultiHandle) { @@ -545,22 +528,20 @@ LLCurl::Multi::Multi() mCurlMultiHandle = curl_multi_init(); } - llassert_always(mCurlMultiHandle); - ++gCurlMultiCount; -} - -LLCurl::Multi::~Multi() -{ - llassert(isStopped()); + llassert_always(mCurlMultiHandle); - if (LLCurl::sMultiThreaded) + if(LLCurl::getCurlThread()->getThreaded()) { - LLCurl::Easy::sMultiMutex->lock(); + mMutexp = new LLMutex(NULL) ; + mDeletionMutexp = new LLMutex(NULL) ; } + LLCurl::getCurlThread()->addMulti(this) ; - delete mSignal; - mSignal = NULL; + ++gCurlMultiCount; +} +LLCurl::Multi::~Multi() +{ // Clean up active for(easy_active_list_t::iterator iter = mEasyActiveList.begin(); iter != mEasyActiveList.end(); ++iter) @@ -577,75 +558,149 @@ LLCurl::Multi::~Multi() mEasyFreeList.clear(); check_curl_multi_code(curl_multi_cleanup(mCurlMultiHandle)); + + delete mMutexp ; + mMutexp = NULL ; + delete mDeletionMutexp ; + mDeletionMutexp = NULL ; + --gCurlMultiCount; +} - if (LLCurl::sMultiThreaded) +void LLCurl::Multi::lock() +{ + if(mMutexp) { - LLCurl::Easy::sMultiMutex->unlock(); + mMutexp->lock() ; } } -CURLMsg* LLCurl::Multi::info_read(S32* msgs_in_queue) +void LLCurl::Multi::unlock() { - CURLMsg* curlmsg = curl_multi_info_read(mCurlMultiHandle, msgs_in_queue); - return curlmsg; + if(mMutexp) + { + mMutexp->unlock() ; + } } -void LLCurl::Multi::perform() +void LLCurl::Multi::markDead() { - if (mThreaded) + if(mDeletionMutexp) { - if (mPerformState == PERFORM_STATE_READY) - { - mSignal->signal(); - } + mDeletionMutexp->lock() ; } - else + + mDead = TRUE ; + + if(mDeletionMutexp) + { + mDeletionMutexp->unlock() ; + } +} + +void LLCurl::Multi::setState(LLCurl::Multi::ePerformState state) +{ + lock() ; + mState = state ; + if(mState == STATE_READY) { - doPerform(); + LLCurl::getCurlThread()->setPriority(mHandle, LLQueuedThread::PRIORITY_NORMAL) ; } + unlock() ; } -void LLCurl::Multi::run() +LLCurl::Multi::ePerformState LLCurl::Multi::getState() { - llassert(mThreaded); + ePerformState state ; + + lock() ; + state = mState ; + unlock() ; + + return state ; +} + +bool LLCurl::Multi::isCompleted() +{ + return STATE_COMPLETED == getState() ; +} - while (!mQuitting) +bool LLCurl::Multi::waitToComplete() +{ + if(!mMutexp) //not threaded { - mSignal->wait(); - mPerformState = PERFORM_STATE_PERFORMING; - if (!mQuitting) - { - LLMutexLock lock(LLCurl::Easy::sMultiMutex); - doPerform(); - } + doPerform() ; + return true ; + } + + bool completed ; + + lock() ; + completed = (STATE_COMPLETED == mState) ; + if(!completed) + { + LLCurl::getCurlThread()->setPriority(mHandle, LLQueuedThread::PRIORITY_URGENT) ; } + unlock() ; + + return completed; } -void LLCurl::Multi::doPerform() +CURLMsg* LLCurl::Multi::info_read(S32* msgs_in_queue) { - S32 q = 0; - for (S32 call_count = 0; - call_count < MULTI_PERFORM_CALL_REPEAT; - call_count += 1) + CURLMsg* curlmsg = curl_multi_info_read(mCurlMultiHandle, msgs_in_queue); + return curlmsg; +} + +//return true if dead +bool LLCurl::Multi::doPerform() +{ + if(mDeletionMutexp) + { + mDeletionMutexp->lock() ; + } + bool dead = mDead ; + + if(mDead) { - CURLMcode code = curl_multi_perform(mCurlMultiHandle, &q); - if (CURLM_CALL_MULTI_PERFORM != code || q == 0) + setState(STATE_COMPLETED); + mQueued = 0 ; + } + else if(getState() != STATE_COMPLETED) + { + setState(STATE_PERFORMING); + + S32 q = 0; + for (S32 call_count = 0; + call_count < MULTI_PERFORM_CALL_REPEAT; + call_count++) { - check_curl_multi_code(code); - break; + CURLMcode code = curl_multi_perform(mCurlMultiHandle, &q); + if (CURLM_CALL_MULTI_PERFORM != code || q == 0) + { + check_curl_multi_code(code); + + break; + } } - + + mQueued = q; + setState(STATE_COMPLETED) ; } - mQueued = q; - mPerformState = PERFORM_STATE_COMPLETED; + + if(mDeletionMutexp) + { + mDeletionMutexp->unlock() ; + } + + return dead ; } S32 LLCurl::Multi::process() { - perform(); + waitToComplete() ; - if (mPerformState != PERFORM_STATE_COMPLETED) + if (getState() != STATE_COMPLETED) { return 0; } @@ -681,7 +736,8 @@ S32 LLCurl::Multi::process() } } - mPerformState = PERFORM_STATE_READY; + setState(STATE_READY); + return processed; } @@ -739,6 +795,75 @@ void LLCurl::Multi::removeEasy(Easy* easy) easyFree(easy); } +//------------------------------------------------------------ +//LLCurlThread +LLCurlThread::CurlRequest::CurlRequest(handle_t handle, LLCurl::Multi* multi) : + LLQueuedThread::QueuedRequest(handle, LLQueuedThread::PRIORITY_NORMAL, FLAG_AUTO_COMPLETE), + mMulti(multi) +{ +} + +LLCurlThread::CurlRequest::~CurlRequest() +{ + if(mMulti) + { + delete mMulti ; + mMulti = NULL ; + } +} + +bool LLCurlThread::CurlRequest::processRequest() +{ + bool completed = true ; + if(mMulti) + { + completed = mMulti->doPerform() ; + setPriority(LLQueuedThread::PRIORITY_LOW) ; + } + + return completed ; +} + +void LLCurlThread::CurlRequest::finishRequest(bool completed) +{ + delete mMulti ; + mMulti = NULL ; +} + +LLCurlThread::LLCurlThread(bool threaded) : + LLQueuedThread("curlthread", threaded) +{ +} + +//virtual +LLCurlThread::~LLCurlThread() +{ +} + +S32 LLCurlThread::update(U32 max_time_ms) +{ + return LLQueuedThread::update(max_time_ms); +} + +void LLCurlThread::addMulti(LLCurl::Multi* multi) +{ + multi->mHandle = generateHandle() ; + + CurlRequest* req = new CurlRequest(multi->mHandle, multi) ; + + if (!addRequest(req)) + { + llwarns << "curl request added when the thread is quitted" << llendl; + } +} + +void LLCurlThread::deleteMulti(LLCurl::Multi* multi) +{ + multi->markDead() ; +} + +//------------------------------------------------------------ + //static std::string LLCurl::strerror(CURLcode errorcode) { @@ -753,39 +878,23 @@ LLCurlRequest::LLCurlRequest() : mActiveMulti(NULL), mActiveRequestCount(0) { - mThreadID = LLThread::currentID(); mProcessing = FALSE; } LLCurlRequest::~LLCurlRequest() { - llassert_always(mThreadID == LLThread::currentID()); - //stop all Multi handle background threads for (curlmulti_set_t::iterator iter = mMultiSet.begin(); iter != mMultiSet.end(); ++iter) { - LLCurl::Multi* multi = *iter; - multi->mQuitting = true; - if (multi->mThreaded) - { - while (!multi->isStopped()) - { - multi->mSignal->signal(); - apr_sleep(1000); - } - } + LLCurl::getCurlThread()->deleteMulti(*iter) ; } - for_each(mMultiSet.begin(), mMultiSet.end(), DeletePointer()); + mMultiSet.clear() ; } void LLCurlRequest::addMulti() { - llassert_always(mThreadID == LLThread::currentID()); LLCurl::Multi* multi = new LLCurl::Multi(); - if (multi->mThreaded) - { - multi->start(); - } + mMultiSet.insert(multi); mActiveMulti = multi; mActiveRequestCount = 0; @@ -901,7 +1010,6 @@ bool LLCurlRequest::post(const std::string& url, // Note: call once per frame S32 LLCurlRequest::process() { - llassert_always(mThreadID == LLThread::currentID()); S32 res = 0; mProcessing = TRUE; @@ -915,17 +1023,7 @@ S32 LLCurlRequest::process() if (multi != mActiveMulti && tres == 0 && multi->mQueued == 0) { mMultiSet.erase(curiter); - multi->mQuitting = true; - if (multi->mThreaded) - { - while (!multi->isStopped()) - { - multi->mSignal->signal(); - apr_sleep(1000); - } - } - - delete multi; + LLCurl::getCurlThread()->deleteMulti(multi); } } mProcessing = FALSE; @@ -934,7 +1032,6 @@ S32 LLCurlRequest::process() S32 LLCurlRequest::getQueued() { - llassert_always(mThreadID == LLThread::currentID()); S32 queued = 0; for (curlmulti_set_t::iterator iter = mMultiSet.begin(); iter != mMultiSet.end(); ) @@ -942,7 +1039,7 @@ S32 LLCurlRequest::getQueued() curlmulti_set_t::iterator curiter = iter++; LLCurl::Multi* multi = *curiter; queued += multi->mQueued; - if (multi->mPerformState != LLCurl::Multi::PERFORM_STATE_READY) + if (multi->getState() != LLCurl::Multi::STATE_READY) { ++queued; } @@ -959,10 +1056,7 @@ LLCurlEasyRequest::LLCurlEasyRequest() mResultReturned(false) { mMulti = new LLCurl::Multi(); - if (mMulti->mThreaded) - { - mMulti->start(); - } + mEasy = mMulti->allocEasy(); if (mEasy) { @@ -975,16 +1069,7 @@ LLCurlEasyRequest::LLCurlEasyRequest() LLCurlEasyRequest::~LLCurlEasyRequest() { - mMulti->mQuitting = true; - if (mMulti->mThreaded) - { - while (!mMulti->isStopped()) - { - mMulti->mSignal->signal(); - apr_sleep(1000); - } - } - delete mMulti; + LLCurl::getCurlThread()->deleteMulti(mMulti) ; } void LLCurlEasyRequest::setopt(CURLoption option, S32 value) @@ -1080,19 +1165,14 @@ void LLCurlEasyRequest::requestComplete() } } -void LLCurlEasyRequest::perform() -{ - mMulti->perform(); -} - // Usage: Call getRestult until it returns false (no more messages) bool LLCurlEasyRequest::getResult(CURLcode* result, LLCurl::TransferInfo* info) { - if (mMulti->mPerformState != LLCurl::Multi::PERFORM_STATE_COMPLETED) + if (!mMulti->isCompleted()) { //we're busy, try again later return false; } - mMulti->mPerformState = LLCurl::Multi::PERFORM_STATE_READY; + mMulti->setState(LLCurl::Multi::STATE_READY) ; if (!mEasy) { @@ -1180,8 +1260,6 @@ unsigned long LLCurl::ssl_thread_id(void) 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 @@ -1189,9 +1267,6 @@ void LLCurl::initClass(bool multi_threaded) check_curl_code(code); - Easy::sHandleMutex = new LLMutex(NULL); - Easy::sMultiMutex = new LLMutex(NULL); - #if SAFE_SSL S32 mutex_count = CRYPTO_num_locks(); for (S32 i=0; i<mutex_count; i++) @@ -1201,20 +1276,29 @@ void LLCurl::initClass(bool multi_threaded) CRYPTO_set_id_callback(&LLCurl::ssl_thread_id); CRYPTO_set_locking_callback(&LLCurl::ssl_locking_callback); #endif + + sCurlThread = new LLCurlThread(multi_threaded) ; } void LLCurl::cleanupClass() { + //shut down curl thread + while(1) + { + if(!sCurlThread->update(1)) //finish all tasks + { + break ; + } + } + sCurlThread->shutdown() ; + delete sCurlThread ; + sCurlThread = NULL ; + #if SAFE_SSL CRYPTO_set_locking_callback(NULL); for_each(sSSLMutex.begin(), sSSLMutex.end(), DeletePointer()); #endif - delete Easy::sHandleMutex; - Easy::sHandleMutex = NULL; - delete Easy::sMultiMutex; - Easy::sMultiMutex = NULL; - for (std::set<CURL*>::iterator iter = Easy::sFreeHandles.begin(); iter != Easy::sFreeHandles.end(); ++iter) { CURL* curl = *iter; diff --git a/indra/llmessage/llcurl.h b/indra/llmessage/llcurl.h index 87de202717..23a6ca67e3 100755 --- a/indra/llmessage/llcurl.h +++ b/indra/llmessage/llcurl.h @@ -42,8 +42,10 @@ #include "lliopipe.h" #include "llsd.h" #include "llthread.h" +#include "llqueuedthread.h" class LLMutex; +class LLCurlThread; // For whatever reason, this is not typedef'd in curl.h typedef size_t (*curl_header_callback)(void *ptr, size_t size, size_t nmemb, void *stream); @@ -56,8 +58,6 @@ public: class Easy; class Multi; - static bool sMultiThreaded; - struct TransferInfo { TransferInfo() : mSizeDownload(0.0), mTotalTime(0.0), mSpeedDownload(0.0) {} @@ -181,10 +181,12 @@ public: static void ssl_locking_callback(int mode, int type, const char *file, int line); static unsigned long ssl_thread_id(void); + static LLCurlThread* getCurlThread() { return sCurlThread ;} private: static std::string sCAPath; static std::string sCAFile; static const unsigned int MAX_REDIRECTS; + static LLCurlThread* sCurlThread; }; class LLCurl::Easy @@ -216,7 +218,7 @@ public: U32 report(CURLcode); void getTransferInfo(LLCurl::TransferInfo* info); - void prepRequest(const std::string& url, const std::vector<std::string>& headers, ResponderPtr, S32 time_out = 0, bool post = false); + void prepRequest(const std::string& url, const std::vector<std::string>& headers, LLCurl::ResponderPtr, S32 time_out = 0, bool post = false); const char* getErrorBuffer(); @@ -247,64 +249,105 @@ private: // Note: char*'s not strings since we pass pointers to curl std::vector<char*> mStrings; - ResponderPtr mResponder; + LLCurl::ResponderPtr mResponder; static std::set<CURL*> sFreeHandles; static std::set<CURL*> sActiveHandles; - static LLMutex* sHandleMutex; - static LLMutex* sMultiMutex; }; -class LLCurl::Multi : public LLThread +class LLCurl::Multi { LOG_CLASS(Multi); + + friend class LLCurlThread ; + +private: + ~Multi(); + + void markDead() ; + bool doPerform(); + public: typedef enum { - PERFORM_STATE_READY=0, - PERFORM_STATE_PERFORMING=1, - PERFORM_STATE_COMPLETED=2 + STATE_READY=0, + STATE_PERFORMING=1, + STATE_COMPLETED=2 } ePerformState; - Multi(); - ~Multi(); + Multi(); - Easy* allocEasy(); - bool addEasy(Easy* easy); + LLCurl::Easy* allocEasy(); + bool addEasy(LLCurl::Easy* easy); + void removeEasy(LLCurl::Easy* easy); - void removeEasy(Easy* easy); + void lock() ; + void unlock() ; + + void setState(ePerformState state) ; + ePerformState getState() ; + bool isCompleted() ; + + bool waitToComplete() ; S32 process(); - void perform(); - void doPerform(); - virtual void run(); - CURLMsg* info_read(S32* msgs_in_queue); S32 mQueued; S32 mErrorCount; - S32 mPerformState; - - LLCondition* mSignal; - bool mQuitting; - bool mThreaded; - private: - void easyFree(Easy*); + void easyFree(LLCurl::Easy*); CURLM* mCurlMultiHandle; - typedef std::set<Easy*> easy_active_list_t; + typedef std::set<LLCurl::Easy*> easy_active_list_t; easy_active_list_t mEasyActiveList; - typedef std::map<CURL*, Easy*> easy_active_map_t; + typedef std::map<CURL*, LLCurl::Easy*> easy_active_map_t; easy_active_map_t mEasyActiveMap; - typedef std::set<Easy*> easy_free_list_t; + typedef std::set<LLCurl::Easy*> easy_free_list_t; easy_free_list_t mEasyFreeList; + + LLQueuedThread::handle_t mHandle ; + ePerformState mState; + + BOOL mDead ; + LLMutex* mMutexp ; + LLMutex* mDeletionMutexp ; }; +class LLCurlThread : public LLQueuedThread +{ +public: + + class CurlRequest : public LLQueuedThread::QueuedRequest + { + protected: + virtual ~CurlRequest(); // use deleteRequest() + + public: + CurlRequest(handle_t handle, LLCurl::Multi* multi); + + /*virtual*/ bool processRequest(); + /*virtual*/ void finishRequest(bool completed); + + private: + // input + LLCurl::Multi* mMulti; + }; + +public: + LLCurlThread(bool threaded = true) ; + virtual ~LLCurlThread() ; + + S32 update(U32 max_time_ms); + + void addMulti(LLCurl::Multi* multi) ; + void deleteMulti(LLCurl::Multi* multi) ; +} ; + namespace boost { void intrusive_ptr_add_ref(LLCurl::Responder* p); @@ -339,7 +382,6 @@ private: LLCurl::Multi* mActiveMulti; S32 mActiveRequestCount; BOOL mProcessing; - U32 mThreadID; // debug }; class LLCurlEasyRequest @@ -357,9 +399,10 @@ public: void slist_append(const char* str); void sendRequest(const std::string& url); void requestComplete(); - void perform(); bool getResult(CURLcode* result, LLCurl::TransferInfo* info = NULL); std::string getErrorString(); + bool isCompleted() {return mMulti->isCompleted() ;} + bool wait() { return mMulti->waitToComplete(); } LLCurl::Easy* getEasy() const { return mEasy; } diff --git a/indra/llmessage/llurlrequest.cpp b/indra/llmessage/llurlrequest.cpp index fa03bb7512..a3a2b2b1b8 100644 --- a/indra/llmessage/llurlrequest.cpp +++ b/indra/llmessage/llurlrequest.cpp @@ -170,6 +170,7 @@ LLURLRequest::~LLURLRequest() { LLMemType m1(LLMemType::MTYPE_IO_URL_REQUEST); delete mDetail; + mDetail = NULL ; } void LLURLRequest::setURL(const std::string& url) @@ -344,7 +345,10 @@ LLIOPipe::EStatus LLURLRequest::process_impl( static LLFastTimer::DeclareTimer FTM_URL_PERFORM("Perform"); { LLFastTimer t(FTM_URL_PERFORM); - mDetail->mCurlRequest->perform(); + if(!mDetail->mCurlRequest->wait()) + { + return status ; + } } while(1) diff --git a/indra/newview/llappviewer.cpp b/indra/newview/llappviewer.cpp index c937f604fc..8b7108e1e2 100755 --- a/indra/newview/llappviewer.cpp +++ b/indra/newview/llappviewer.cpp @@ -1417,6 +1417,11 @@ bool LLAppViewer::mainLoop() } } gMeshRepo.update() ; + + if(!LLCurl::getCurlThread()->update(1)) + { + LLCurl::getCurlThread()->pause() ; //nothing in the curl thread. + } if(!total_work_pending) //pause texture fetching threads if nothing to process. { diff --git a/indra/newview/llxmlrpctransaction.cpp b/indra/newview/llxmlrpctransaction.cpp index f483ba5af8..920a9a3752 100644 --- a/indra/newview/llxmlrpctransaction.cpp +++ b/indra/newview/llxmlrpctransaction.cpp @@ -382,19 +382,11 @@ bool LLXMLRPCTransaction::Impl::process() // continue onward } } - - //const F32 MAX_PROCESSING_TIME = 0.05f; - //LLTimer timer; - - mCurlRequest->perform(); - - /*while (mCurlRequest->perform() > 0) + + if(!mCurlRequest->wait()) { - if (timer.getElapsedTimeF32() >= MAX_PROCESSING_TIME) - { - return false; - } - }*/ + return false ; + } while(1) { |