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

Moved PerHostRequestQueue[Ptr] outside of AICurlPrivate.

Browse Source 
Renamed PerHostRequestQueue and PerHostRequestQueuePtr to
AIPerHostRequestQueue and AIPerHostRequestQueuePtr respectively and
moved them to global namespace. This in preparation for using them in
the texture fetcher (as function of the hostname of the url that is
contructed there).
This commit is contained in:
Aleric Inglewood
2013-04-06 21:54:34 +02:00
parent db7c378160
commit 7866c68ab2
6 changed files with 82 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
indra/llmessage/aicurl.cpp
View File

@@ -961,7 +961,7 @@ CurlEasyRequest::~CurlEasyRequest()
revokeCallbacks();
if (mPerHostPtr)
{
PerHostRequestQueue::release(mPerHostPtr);
AIPerHostRequestQueue::release(mPerHostPtr);
}
// This wasn't freed yet if the request never finished.
curl_slist_free_all(mHeaders);
@@ -1286,14 +1286,14 @@ void CurlEasyRequest::queued_for_removal(AICurlEasyRequest_wat& curl_easy_reques
}
#endif
PerHostRequestQueuePtr CurlEasyRequest::getPerHostPtr(void)
AIPerHostRequestQueuePtr CurlEasyRequest::getPerHostPtr(void)
{
if (!mPerHostPtr)
{
// mPerHostPtr is really just a speed-up cache.
// The reason we can cache it is because mLowercaseHostname is only set
// in finalizeRequest which may only be called once: it never changes.
mPerHostPtr = PerHostRequestQueue::instance(mLowercaseHostname);
mPerHostPtr = AIPerHostRequestQueue::instance(mLowercaseHostname);
}
return mPerHostPtr;
}

72
indra/llmessage/aicurlperhost.cpp
View File

@@ -1,8 +1,8 @@
/**
* @file aiperhost.cpp
* @brief Implementation of PerHostRequestQueue
* @brief Implementation of AIPerHostRequestQueue
*
* Copyright (c) 2012, Aleric Inglewood.
* Copyright (c) 2012, 2013, Aleric Inglewood.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
@@ -26,36 +26,60 @@
*
* 04/11/2012
* Initial version, written by Aleric Inglewood @ SL
*
* 06/04/2013
* Renamed AICurlPrivate::PerHostRequestQueue[Ptr] to AIPerHostRequestQueue[Ptr]
* to allow public access.
*/
#include "sys.h"
#include "aicurlperhost.h"
#include "aicurlthread.h"
AIPerHostRequestQueue::threadsafe_instance_map_type AIPerHostRequestQueue::sInstanceMap;
LLAtomicS32 AIPerHostRequestQueue::sTotalQueued;
#undef AICurlPrivate
namespace AICurlPrivate {
PerHostRequestQueue::threadsafe_instance_map_type PerHostRequestQueue::sInstanceMap;
LLAtomicS32 PerHostRequestQueue::sTotalQueued;
U32 curl_concurrent_connections_per_host;
// Friend functions of RefCountedThreadSafePerHostRequestQueue
void intrusive_ptr_add_ref(RefCountedThreadSafePerHostRequestQueue* per_host)
{
per_host->mReferenceCount++;
}
void intrusive_ptr_release(RefCountedThreadSafePerHostRequestQueue* per_host)
{
if (--per_host->mReferenceCount == 0)
{
delete per_host;
}
}
} // namespace AICurlPrivate
using namespace AICurlPrivate;
//static
PerHostRequestQueuePtr PerHostRequestQueue::instance(std::string const& hostname)
AIPerHostRequestQueuePtr AIPerHostRequestQueue::instance(std::string const& hostname)
{
llassert(!hostname.empty());
instance_map_wat instance_map_w(sInstanceMap);
PerHostRequestQueue::iterator iter = instance_map_w->find(hostname);
AIPerHostRequestQueue::iterator iter = instance_map_w->find(hostname);
if (iter == instance_map_w->end())
{
iter = instance_map_w->insert(instance_map_type::value_type(hostname, new RefCountedThreadSafePerHostRequestQueue)).first;
}
// Note: the creation of PerHostRequestQueuePtr MUST be protected by the lock on sInstanceMap (see release()).
// Note: the creation of AIPerHostRequestQueuePtr MUST be protected by the lock on sInstanceMap (see release()).
return iter->second;
}
//static
void PerHostRequestQueue::release(PerHostRequestQueuePtr& instance)
void AIPerHostRequestQueue::release(AIPerHostRequestQueuePtr& instance)
{
if (instance->exactly_two_left()) // Being 'instance' and the one in sInstanceMap.
{
@@ -91,31 +115,31 @@ void PerHostRequestQueue::release(PerHostRequestQueuePtr& instance)
instance.reset();
}
bool PerHostRequestQueue::throttled() const
bool AIPerHostRequestQueue::throttled() const
{
llassert(mAdded <= int(curl_concurrent_connections_per_host));
return mAdded == int(curl_concurrent_connections_per_host);
}
void PerHostRequestQueue::added_to_multi_handle(void)
void AIPerHostRequestQueue::added_to_multi_handle(void)
{
llassert(mAdded < int(curl_concurrent_connections_per_host));
++mAdded;
}
void PerHostRequestQueue::removed_from_multi_handle(void)
void AIPerHostRequestQueue::removed_from_multi_handle(void)
{
--mAdded;
llassert(mAdded >= 0);
}
void PerHostRequestQueue::queue(AICurlEasyRequest const& easy_request)
void AIPerHostRequestQueue::queue(AICurlEasyRequest const& easy_request)
{
mQueuedRequests.push_back(easy_request.get_ptr());
sTotalQueued++;
}
bool PerHostRequestQueue::cancel(AICurlEasyRequest const& easy_request)
bool AIPerHostRequestQueue::cancel(AICurlEasyRequest const& easy_request)
{
queued_request_type::iterator const end = mQueuedRequests.end();
queued_request_type::iterator cur = std::find(mQueuedRequests.begin(), end, easy_request.get_ptr());
@@ -128,7 +152,7 @@ bool PerHostRequestQueue::cancel(AICurlEasyRequest const& easy_request)
// the back with swap (could just swap with the end immediately, but I don't
// want to break the order in which requests where added). Swap is also not
// thread-safe, but OK here because it only touches the objects in the deque,
// and the deque is protected by the lock on the PerHostRequestQueue object.
// and the deque is protected by the lock on the AIPerHostRequestQueue object.
queued_request_type::iterator prev = cur;
while (++cur != end)
{
@@ -141,7 +165,7 @@ bool PerHostRequestQueue::cancel(AICurlEasyRequest const& easy_request)
return true;
}
void PerHostRequestQueue::add_queued_to(curlthread::MultiHandle* multi_handle)
void AIPerHostRequestQueue::add_queued_to(curlthread::MultiHandle* multi_handle)
{
if (!mQueuedRequests.empty())
{
@@ -153,7 +177,7 @@ void PerHostRequestQueue::add_queued_to(curlthread::MultiHandle* multi_handle)
}
//static
void PerHostRequestQueue::purge(void)
void AIPerHostRequestQueue::purge(void)
{
instance_map_wat instance_map_w(sInstanceMap);
for (iterator host = instance_map_w->begin(); host != instance_map_w->end(); ++host)
@@ -167,19 +191,3 @@ void PerHostRequestQueue::purge(void)
}
}
// Friend functions of RefCountedThreadSafePerHostRequestQueue
void intrusive_ptr_add_ref(RefCountedThreadSafePerHostRequestQueue* per_host)
{
per_host->mReferenceCount++;
}
void intrusive_ptr_release(RefCountedThreadSafePerHostRequestQueue* per_host)
{
if (--per_host->mReferenceCount == 0)
{
delete per_host;
}
}
} // namespace AICurlPrivate

55
indra/llmessage/aicurlperhost.h
View File

@@ -1,8 +1,8 @@
/**
* @file aicurlperhost.h
* @brief Definition of class PerHostRequestQueue
* @brief Definition of class AIPerHostRequestQueue
*
* Copyright (c) 2012, Aleric Inglewood.
* Copyright (c) 2012, 2013, Aleric Inglewood.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
@@ -26,6 +26,10 @@
*
* 04/11/2012
* Initial version, written by Aleric Inglewood @ SL
*
* 06/04/2013
* Renamed AIPrivate::PerHostRequestQueue[Ptr] to AIPerHostRequestQueue[Ptr]
* to allow public access.
*/
#ifndef AICURLPERHOST_H
@@ -39,68 +43,70 @@
#include "aithreadsafe.h"
class AICurlEasyRequest;
class AIPerHostRequestQueue;
namespace AICurlPrivate {
namespace curlthread { class MultiHandle; }
class PerHostRequestQueue;
class RefCountedThreadSafePerHostRequestQueue;
class ThreadSafeBufferedCurlEasyRequest;
// Forward declaration of BufferedCurlEasyRequestPtr (see aicurlprivate.h).
typedef boost::intrusive_ptr<ThreadSafeBufferedCurlEasyRequest> BufferedCurlEasyRequestPtr;
// PerHostRequestQueue objects are created by the curl thread and destructed by the main thread.
// AIPerHostRequestQueue objects are created by the curl thread and destructed by the main thread.
// We need locking.
typedef AIThreadSafeSimpleDC<PerHostRequestQueue> threadsafe_PerHostRequestQueue;
typedef AIAccessConst<PerHostRequestQueue> PerHostRequestQueue_crat;
typedef AIAccess<PerHostRequestQueue> PerHostRequestQueue_rat;
typedef AIAccess<PerHostRequestQueue> PerHostRequestQueue_wat;
typedef AIThreadSafeSimpleDC<AIPerHostRequestQueue> threadsafe_PerHostRequestQueue;
typedef AIAccessConst<AIPerHostRequestQueue> PerHostRequestQueue_crat;
typedef AIAccess<AIPerHostRequestQueue> PerHostRequestQueue_rat;
typedef AIAccess<AIPerHostRequestQueue> PerHostRequestQueue_wat;
} // namespace AICurlPrivate
// We can't put threadsafe_PerHostRequestQueue in a std::map because you can't copy a mutex.
// Therefore, use an intrusive pointer for the threadsafe type.
typedef boost::intrusive_ptr<RefCountedThreadSafePerHostRequestQueue> PerHostRequestQueuePtr;
typedef boost::intrusive_ptr<AICurlPrivate::RefCountedThreadSafePerHostRequestQueue> AIPerHostRequestQueuePtr;
//-----------------------------------------------------------------------------
// PerHostRequestQueue
// AIPerHostRequestQueue
// This class provides a static interface to create and maintain instances
// of PerHostRequestQueue objects, so that at any moment there is at most
// of AIPerHostRequestQueue objects, so that at any moment there is at most
// one instance per hostname. Those instances then are used to queue curl
// requests when the maximum number of connections for that host already
// have been reached.
class PerHostRequestQueue {
class AIPerHostRequestQueue {
private:
typedef std::map<std::string, PerHostRequestQueuePtr> instance_map_type;
typedef std::map<std::string, AIPerHostRequestQueuePtr> instance_map_type;
typedef AIThreadSafeSimpleDC<instance_map_type> threadsafe_instance_map_type;
typedef AIAccess<instance_map_type> instance_map_rat;
typedef AIAccess<instance_map_type> instance_map_wat;
static threadsafe_instance_map_type sInstanceMap; // Map of PerHostRequestQueue instances with the hostname as key.
static threadsafe_instance_map_type sInstanceMap; // Map of AIPerHostRequestQueue instances with the hostname as key.
friend class AIThreadSafeSimpleDC<PerHostRequestQueue>; //threadsafe_PerHostRequestQueue
PerHostRequestQueue(void) : mAdded(0) { }
friend class AIThreadSafeSimpleDC<AIPerHostRequestQueue>; //threadsafe_PerHostRequestQueue
AIPerHostRequestQueue(void) : mAdded(0) { }
public:
typedef instance_map_type::iterator iterator;
typedef instance_map_type::const_iterator const_iterator;
// Return (possibly create) a unique instance for the given hostname.
static PerHostRequestQueuePtr instance(std::string const& hostname);
static AIPerHostRequestQueuePtr instance(std::string const& hostname);
// Release instance (object will be deleted if this was the last instance).
static void release(PerHostRequestQueuePtr& instance);
static void release(AIPerHostRequestQueuePtr& instance);
// Remove everything. Called upon viewer exit.
static void purge(void);
private:
typedef std::deque<BufferedCurlEasyRequestPtr> queued_request_type;
typedef std::deque<AICurlPrivate::BufferedCurlEasyRequestPtr> queued_request_type;
int mAdded; // Number of active easy handles with this host.
queued_request_type mQueuedRequests; // Waiting (throttled) requests.
static LLAtomicS32 sTotalQueued; // The sum of mQueuedRequests.size() of all PerHostRequestQueue objects together.
static LLAtomicS32 sTotalQueued; // The sum of mQueuedRequests.size() of all AIPerHostRequestQueue objects together.
public:
void added_to_multi_handle(void); // Called when an easy handle for this host has been added to the multi handle.
@@ -110,7 +116,8 @@ class PerHostRequestQueue {
void queue(AICurlEasyRequest const& easy_request); // Add easy_request to the queue.
bool cancel(AICurlEasyRequest const& easy_request); // Remove easy_request from the queue (if it's there).
void add_queued_to(curlthread::MultiHandle* mh); // Add queued easy handle (if any) to the multi handle. The request is removed from the queue,
void add_queued_to(AICurlPrivate::curlthread::MultiHandle* mh);
// Add queued easy handle (if any) to the multi handle. The request is removed from the queue,
// followed by either a call to added_to_multi_handle() or to queue() to add it back.
S32 host_queued_plus_added_size(void) const { return mQueuedRequests.size() + mAdded; }
@@ -118,16 +125,18 @@ class PerHostRequestQueue {
private:
// Disallow copying.
PerHostRequestQueue(PerHostRequestQueue const&) { }
AIPerHostRequestQueue(AIPerHostRequestQueue const&) { }
};
namespace AICurlPrivate {
class RefCountedThreadSafePerHostRequestQueue : public threadsafe_PerHostRequestQueue {
public:
RefCountedThreadSafePerHostRequestQueue(void) : mReferenceCount(0) { }
bool exactly_two_left(void) const { return mReferenceCount == 2; }
private:
// Used by PerHostRequestQueuePtr. Object is deleted when reference count reaches zero.
// Used by AIPerHostRequestQueuePtr. Object is deleted when reference count reaches zero.
LLAtomicU32 mReferenceCount;
friend void intrusive_ptr_add_ref(RefCountedThreadSafePerHostRequestQueue* p);

6
indra/llmessage/aicurlprivate.h
View File

@@ -305,7 +305,7 @@ class CurlEasyRequest : public CurlEasyHandle {
AIHTTPTimeoutPolicy const* mTimeoutPolicy;
std::string mLowercaseHostname; // Lowercase hostname (canonicalized) extracted from the url.
PerHostRequestQueuePtr mPerHostPtr; // Pointer to the corresponding PerHostRequestQueue.
AIPerHostRequestQueuePtr mPerHostPtr; // Pointer to the corresponding AIPerHostRequestQueue.
LLPointer<curlthread::HTTPTimeout> mTimeout;// Timeout administration object associated with last created CurlSocketInfo.
bool mTimeoutIsOrphan; // Set to true when mTimeout is not (yet) associated with a CurlSocketInfo.
#if defined(CWDEBUG) || defined(DEBUG_CURLIO)
@@ -348,8 +348,8 @@ class CurlEasyRequest : public CurlEasyHandle {
inline ThreadSafeBufferedCurlEasyRequest const* get_lockobj(void) const;
// PerHost API.
PerHostRequestQueuePtr getPerHostPtr(void); // (Optionally create and) return a pointer to the unique
// PerHostRequestQueue corresponding to mLowercaseHostname.
AIPerHostRequestQueuePtr getPerHostPtr(void); // (Optionally create and) return a pointer to the unique
// AIPerHostRequestQueue corresponding to mLowercaseHostname.
bool removeFromPerHostQueue(AICurlEasyRequest const&) const; // Remove this request from the per-host queue, if queued at all.
// Returns true if it was queued.
protected:

6
indra/llmessage/aicurlthread.cpp
View File

@@ -1573,7 +1573,7 @@ void AICurlThread::run(void)
multi_handle_w->check_msg_queue();
}
// Clear the queued requests.
PerHostRequestQueue::purge();
AIPerHostRequestQueue::purge();
}
AICurlMultiHandle::destroyInstance();
}
@@ -1700,7 +1700,7 @@ static U32 curl_max_total_concurrent_connections = 32; // Initialized on st
void MultiHandle::add_easy_request(AICurlEasyRequest const& easy_request)
{
bool throttled = true; // Default.
PerHostRequestQueuePtr per_host;
AIPerHostRequestQueuePtr per_host;
{
AICurlEasyRequest_wat curl_easy_request_w(*easy_request);
per_host = curl_easy_request_w->getPerHostPtr();
@@ -1759,7 +1759,7 @@ CURLMcode MultiHandle::remove_easy_request(AICurlEasyRequest const& easy_request
CURLMcode MultiHandle::remove_easy_request(addedEasyRequests_type::iterator const& iter, bool as_per_command)
{
CURLMcode res;
PerHostRequestQueuePtr per_host;
AIPerHostRequestQueuePtr per_host;
{
AICurlEasyRequest_wat curl_easy_request_w(**iter);
res = curl_easy_request_w->remove_handle_from_multi(curl_easy_request_w, mMultiHandle);

2
indra/llmessage/aicurlthread.h
View File

@@ -81,7 +81,7 @@ class MultiHandle : public CurlMultiHandle
// Store result and trigger events for easy request.
void finish_easy_request(AICurlEasyRequest const& easy_request, CURLcode result);
// Remove easy request at iter (must exist).
// Note that it's possible that a new request from a PerHostRequestQueue::mQueuedRequests is inserted before iter.
// Note that it's possible that a new request from a AIPerHostRequestQueue::mQueuedRequests is inserted before iter.
CURLMcode remove_easy_request(addedEasyRequests_type::iterator const& iter, bool as_per_command);
static int socket_callback(CURL* easy, curl_socket_t s, int action, void* userp, void* socketp);