From 9b33a931f18da6f270daff99fa84d61d2718285f Mon Sep 17 00:00:00 2001 From: Niels Sascha Reedijk Date: Wed, 27 Jul 2022 09:02:09 +0100 Subject: [PATCH] NetServices: Implement controls on the number of concurrent requests. This change allows the user to control how many concurrent request can be done per session. This is going to be helpful to running the tests as well; they were all fired up in parallel, which sometimes leads to our test server refusing a connection. Change-Id: I4f1f40b28b8e0199ea5589b36cd8d00ecd849a63 --- docs/user/netservices/HttpSession.dox | 41 ++++++ headers/private/netservices2/HttpSession.h | 4 + .../network/libnetservices2/HttpSession.cpp | 134 +++++++++++++++++- .../net/netservices2/HttpProtocolTest.cpp | 7 +- 4 files changed, 179 insertions(+), 7 deletions(-) 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()); + } }