diff --git a/docs/user/netservices/HttpSession.dox b/docs/user/netservices/HttpSession.dox index d570ed90c9..f2e8239e7b 100644 --- a/docs/user/netservices/HttpSession.dox +++ b/docs/user/netservices/HttpSession.dox @@ -219,6 +219,47 @@ namespace Network { */ +/*! + \fn void BHttpSession::SetMaxConnectionsPerHost(size_t maxConnections) + \brief Set the maximum number of connections per host. + + A host is identified by the domain name and the port. You can limit the number of concurrent + connections to a host by tweaking this value. + + The default value is 2 connections per host. + + If the value is decreased, any requests that already started will not be affected. The new + value will only be applied when any new requests are added. + + \param maxConnections The maximum number of connections per host. This value must between 1 + and \c INT32_MAX. + + \exception BRuntimeError In case the \a maxConnections is invalid. + + \since Haiku R1 +*/ + + +/*! + \fn void BPrivate::Network::BHttpSession::SetMaxHosts(size_t maxConnections) + \brief Set the maximum number of concurrent hosts that can be connected to. + + A host is identified by the domain name and the port. You can limit the number of concurrent + hosts by tweaking this value. + + The default value is 10 concurrent hosts. + + If the value is decreased, any requests that already started will not be affected. The new + value will only be applied when any new requests are added. + + \param maxConnections The maximum number of hosts. The value must be at least 1. + + \exception BRuntimeError In case the \a maxConnections is 0. + + \since Haiku R1 +*/ + + /*! \var UrlEvent::HttpStatus \brief The HTTP status code has been received, and can be accessed through the result object. diff --git a/headers/private/netservices2/HttpSession.h b/headers/private/netservices2/HttpSession.h index 5a011d87d0..5f06d9f323 100644 --- a/headers/private/netservices2/HttpSession.h +++ b/headers/private/netservices2/HttpSession.h @@ -40,6 +40,10 @@ public: void Cancel(int32 identifier); void Cancel(const BHttpResult& request); + // Concurrency limits + void SetMaxConnectionsPerHost(size_t maxConnections); + void SetMaxHosts(size_t maxConnections); + private: struct Redirect; class Request; diff --git a/src/kits/network/libnetservices2/HttpSession.cpp b/src/kits/network/libnetservices2/HttpSession.cpp index f84939a578..1c532d0b19 100644 --- a/src/kits/network/libnetservices2/HttpSession.cpp +++ b/src/kits/network/libnetservices2/HttpSession.cpp @@ -7,6 +7,7 @@ */ #include +#include #include #include #include @@ -46,6 +47,14 @@ using namespace BPrivate::Network; static constexpr ssize_t kMaxHeaderLineSize = 64 * 1024; +struct CounterDeleter { + void operator()(int32* counter) const noexcept + { + atomic_add(counter, -1); + } +}; + + class BHttpSession::Request { public: Request(BHttpRequest&& request, @@ -71,6 +80,10 @@ public: Result() { return fResult; } void SetError(std::exception_ptr e); + // Helpers for maintaining the connection count + std::pair GetHost() const; + void SetCounter(int32* counter) noexcept; + // Operational methods void ResolveHostName(); void OpenConnection(); @@ -116,6 +129,10 @@ private: // Redirection std::optional fRedirectStatus; int8 fRemainingRedirects; + + // Connection counter + std::unique_ptr + fConnectionCounter; }; @@ -128,6 +145,8 @@ public: std::unique_ptr target, BMessenger observer); void Cancel(int32 identifier); + void SetMaxConnectionsPerHost(size_t maxConnections); + void SetMaxHosts(size_t maxConnections); private: // Thread functions @@ -145,13 +164,21 @@ private: // locking mechanism BLocker fLock; - int32 fQuitting; + std::atomic fQuitting = false; // queues & shared data std::list fControlQueue; std::deque fDataQueue; std::vector fCancelList; + // data owned by the controlThread + using Host = std::pair; + std::map fConnectionCount; + + // data that can only be accessed atomically + std::atomic fMaxConnectionsPerHost = 2; + std::atomic fMaxHosts = 10; + // data owned by the dataThread std::map connectionMap; std::vector objectList; @@ -195,7 +222,7 @@ BHttpSession::Impl::Impl() BHttpSession::Impl::~Impl() noexcept { - atomic_set(&fQuitting, 1); + fQuitting.store(true); delete_sem(fControlQueueSem); delete_sem(fDataQueueSem); status_t threadResult; @@ -242,6 +269,26 @@ BHttpSession::Impl::Cancel(int32 identifier) } +void +BHttpSession::Impl::SetMaxConnectionsPerHost(size_t maxConnections) +{ + if (maxConnections <= 0 || maxConnections >= INT32_MAX) { + throw BRuntimeError(__PRETTY_FUNCTION__, + "MaxConnectionsPerHost must be between 1 and INT32_MAX"); + } + fMaxConnectionsPerHost.store(maxConnections, std::memory_order_relaxed); +} + + +void +BHttpSession::Impl::SetMaxHosts(size_t maxConnections) +{ + if (maxConnections <= 0) + throw BRuntimeError(__PRETTY_FUNCTION__, "MaxHosts must be 1 or more"); + fMaxHosts.store(maxConnections, std::memory_order_relaxed); +} + + /*static*/ status_t BHttpSession::Impl::ControlThreadFunc(void* arg) { @@ -257,7 +304,7 @@ BHttpSession::Impl::ControlThreadFunc(void* arg) } // Check if we have woken up because we are quitting - if (atomic_get(&impl->fQuitting) == 1) + if (impl->fQuitting.load()) break; // Get items to process (locking done by the helper) @@ -279,7 +326,10 @@ BHttpSession::Impl::ControlThreadFunc(void* arg) } if (hasError) { - // Do not add the request back to the queue + // Do not add the request back to the queue; release the sem to do another round + // in case there is another item waiting because the limits of concurrent requests + // were reached + release_sem(impl->fControlQueueSem); continue; } @@ -291,7 +341,7 @@ BHttpSession::Impl::ControlThreadFunc(void* arg) } // Clean up and make sure we are quitting - if (atomic_get(&impl->fQuitting) == 1) { + if (impl->fQuitting.load()) { // First wait for the data thread to complete status_t threadResult; wait_for_thread(impl->fDataThread, &threadResult); @@ -406,6 +456,8 @@ BHttpSession::Impl::DataThreadFunc(void* arg) if (error) { request.Disconnect(); data->connectionMap.erase(item.object); + release_sem(data->fControlQueueSem); + // wake up control thread; there may queued requests unblocked. resizeObjectList = true; } } else if ((item.events & B_EVENT_READ) == B_EVENT_READ) { @@ -436,6 +488,8 @@ BHttpSession::Impl::DataThreadFunc(void* arg) // Clean up finished requests; including redirected requests request.Disconnect(); data->connectionMap.erase(item.object); + release_sem(data->fControlQueueSem); + // wake up control thread; there may queued requests unblocked. resizeObjectList = true; } } else if ((item.events & B_EVENT_DISCONNECTED) == B_EVENT_DISCONNECTED) { @@ -456,6 +510,8 @@ BHttpSession::Impl::DataThreadFunc(void* arg) request.SetError(std::current_exception()); } data->connectionMap.erase(item.object); + release_sem(data->fControlQueueSem); + // wake up control thread; there may queued requests unblocked. resizeObjectList = true; } else if (item.events == 0) { // No events for this item, skip @@ -492,7 +548,7 @@ BHttpSession::Impl::DataThreadFunc(void* arg) } } // Clean up and make sure we are quitting - if (atomic_get(&data->fQuitting) == 1) { + if (data->fQuitting.load()) { // Cancel all requests for (auto it = data->connectionMap.begin(); it != data->connectionMap.end(); it++) { try { @@ -522,8 +578,46 @@ std::vector BHttpSession::Impl::GetRequestsForControlThread() { std::vector requests; + std::cout << __PRETTY_FUNCTION__ << ": number of items in fConnectionCount: " << fConnectionCount.size() << std::endl; + + // Clean up connection list if it is at the max number of hosts + if (fConnectionCount.size() >= fMaxHosts.load()) { + for (auto it = fConnectionCount.begin(); it != fConnectionCount.end(); ) { + if (atomic_get(std::addressof(it->second)) == 0) { + it = fConnectionCount.erase(it); + } else { + it++; + } + } + } + + // Process the list of pending requests and review if they can be started. auto lock = AutoLocker(fLock); fControlQueue.remove_if([this, &requests](auto& request){ + auto host = request.GetHost(); + auto it = fConnectionCount.find(host); + if (it != fConnectionCount.end()) { + std::cout << __PRETTY_FUNCTION__ << ": found connnections for host, count: " << it->second << std::endl; + if (static_cast(atomic_get(std::addressof(it->second))) + >= fMaxConnectionsPerHost.load(std::memory_order_relaxed)) { + std::cout << "\tskip loading this request as max connections per host is reached" << std::endl; + return false; + } else { + atomic_add(std::addressof(it->second), 1); + request.SetCounter(std::addressof(it->second)); + } + } else { + if (fConnectionCount.size() == fMaxHosts.load()) { + std::cout << "\tskip loading this request as max hosts is reached" << std::endl; + return false; + } + auto[newIt, success] = fConnectionCount.insert({host, 1}); + if (!success) { + throw BRuntimeError(__PRETTY_FUNCTION__, + "Cannot insert into fConnectionCount"); + } + request.SetCounter(std::addressof(newIt->second)); + } requests.emplace_back(std::move(request)); return true; }); @@ -571,6 +665,20 @@ BHttpSession::Cancel(const BHttpResult& request) } +void +BHttpSession::SetMaxConnectionsPerHost(size_t maxConnections) +{ + fImpl->SetMaxConnectionsPerHost(maxConnections); +} + + +void +BHttpSession::SetMaxHosts(size_t maxConnections) +{ + fImpl->SetMaxHosts(maxConnections); +} + + // #pragma mark -- BHttpSession::Request (helpers) BHttpSession::Request::Request(BHttpRequest&& request, std::unique_ptr target, @@ -618,6 +726,20 @@ BHttpSession::Request::SetError(std::exception_ptr e) } +std::pair +BHttpSession::Request::GetHost() const +{ + return {fRequest.Url().Host(), fRequest.Url().Port()}; +} + + +void +BHttpSession::Request::SetCounter(int32* counter) noexcept +{ + fConnectionCounter = std::unique_ptr(counter); +} + + /*! \brief Resolve the hostname for a request */ diff --git a/src/tests/kits/net/netservices2/HttpProtocolTest.cpp b/src/tests/kits/net/netservices2/HttpProtocolTest.cpp index 5fe86067b1..d5415ddbc5 100644 --- a/src/tests/kits/net/netservices2/HttpProtocolTest.cpp +++ b/src/tests/kits/net/netservices2/HttpProtocolTest.cpp @@ -455,7 +455,8 @@ public: HttpIntegrationTest::HttpIntegrationTest(TestServerMode mode) : fTestServer(mode) { - + // increase number of concurrent connections to 4 (from 2) + fSession.SetMaxConnectionsPerHost(4); } @@ -692,10 +693,14 @@ HttpIntegrationTest::BasicAuthTest() CPPUNIT_ASSERT(result.Status().code == 200); // Basic Authentication with incorrect credentials + try { request = BHttpRequest(BUrl(fTestServer.BaseUrl(), "/auth/basic/walter/secret")); request.SetAuthentication({"invaliduser", "invalidpassword"}); result = fSession.Execute(std::move(request)); CPPUNIT_ASSERT(result.Status().code == 401); + } catch (const BPrivate::Network::BError& e) { + CPPUNIT_FAIL(e.DebugMessage().String()); + } }