Commit 5569411a authored by octal's avatar octal

HttpClient: now keeping a map of request queues/domain at the expanse of full lock-freeness

parent b7ef9cf3
...@@ -23,7 +23,7 @@ int main(int argc, char *argv[]) { ...@@ -23,7 +23,7 @@ int main(int argc, char *argv[]) {
auto opts = Http::Experimental::Client::options() auto opts = Http::Experimental::Client::options()
.threads(1) .threads(1)
.maxConnections(64); .maxConnectionsPerHost(8);
client.init(opts); client.init(opts);
auto resp = client.get(page).cookie(Cookie("FOO", "bar")).send(); auto resp = client.get(page).cookie(Cookie("FOO", "bar")).send();
......
...@@ -137,6 +137,9 @@ private: ...@@ -137,6 +137,9 @@ private:
}; };
struct ConnectionPool { struct ConnectionPool {
void init(size_t maxConnsPerHost);
std::shared_ptr<Connection> pickConnection(const std::string& domain); std::shared_ptr<Connection> pickConnection(const std::string& domain);
void releaseConnection(const std::shared_ptr<Connection>& connection); void releaseConnection(const std::shared_ptr<Connection>& connection);
...@@ -154,6 +157,7 @@ private: ...@@ -154,6 +157,7 @@ private:
mutable Lock connsLock; mutable Lock connsLock;
std::unordered_map<std::string, Connections> conns; std::unordered_map<std::string, Connections> conns;
size_t maxConnectionsPerHost;
}; };
class Transport : public Io::Handler { class Transport : public Io::Handler {
...@@ -242,7 +246,7 @@ private: ...@@ -242,7 +246,7 @@ private:
namespace Default { namespace Default {
constexpr int Threads = 1; constexpr int Threads = 1;
constexpr int MaxConnections = 8; constexpr int MaxConnectionsPerHost = 8;
constexpr bool KeepAlive = true; constexpr bool KeepAlive = true;
} }
...@@ -296,19 +300,17 @@ public: ...@@ -296,19 +300,17 @@ public:
Options() Options()
: threads_(Default::Threads) : threads_(Default::Threads)
, maxConnections_(Default::MaxConnections) , maxConnectionsPerHost_(Default::MaxConnectionsPerHost)
, keepAlive_(Default::KeepAlive) , keepAlive_(Default::KeepAlive)
{ } { }
Options& threads(int val); Options& threads(int val);
Options& maxConnections(int val);
Options& keepAlive(bool val); Options& keepAlive(bool val);
Options& maxConnectionsPerHost(int val); Options& maxConnectionsPerHost(int val);
private: private:
int threads_; int threads_;
int maxConnections_;
int maxConnectionsPerHost_; int maxConnectionsPerHost_;
bool keepAlive_; bool keepAlive_;
}; };
...@@ -333,7 +335,12 @@ private: ...@@ -333,7 +335,12 @@ private:
std::shared_ptr<Transport> transport_; std::shared_ptr<Transport> transport_;
std::atomic<uint64_t> ioIndex; std::atomic<uint64_t> ioIndex;
Queue<Connection::RequestData> requestsQueue;
typedef std::mutex Lock;
typedef std::lock_guard<Lock> Guard;
Lock queuesLock;
std::unordered_map<std::string, Queue<Connection::RequestData>> requestsQueues;
RequestBuilder prepareRequest(std::string resource, Http::Method method); RequestBuilder prepareRequest(std::string resource, Http::Method method);
......
...@@ -205,6 +205,12 @@ public: ...@@ -205,6 +205,12 @@ public:
return nullptr; return nullptr;
} }
bool empty() {
auto *res = tail;
auto* next = res->next.load(std::memory_order_acquire);
return next != nullptr;
}
std::unique_ptr<Entry> popSafe() { std::unique_ptr<Entry> popSafe() {
return std::unique_ptr<Entry>(pop()); return std::unique_ptr<Entry>(pop());
} }
......
...@@ -582,6 +582,11 @@ Connection::processRequestQueue() { ...@@ -582,6 +582,11 @@ Connection::processRequestQueue() {
} }
void
ConnectionPool::init(size_t maxConnsPerHost) {
maxConnectionsPerHost = maxConnsPerHost;
}
std::shared_ptr<Connection> std::shared_ptr<Connection>
ConnectionPool::pickConnection(const std::string& domain) { ConnectionPool::pickConnection(const std::string& domain) {
Connections pool; Connections pool;
...@@ -591,7 +596,7 @@ ConnectionPool::pickConnection(const std::string& domain) { ...@@ -591,7 +596,7 @@ ConnectionPool::pickConnection(const std::string& domain) {
auto poolIt = conns.find(domain); auto poolIt = conns.find(domain);
if (poolIt == std::end(conns)) { if (poolIt == std::end(conns)) {
Connections connections; Connections connections;
for (size_t i = 0; i < 128; ++i) { for (size_t i = 0; i < maxConnectionsPerHost; ++i) {
connections.push_back(std::make_shared<Connection>()); connections.push_back(std::make_shared<Connection>());
} }
...@@ -710,12 +715,6 @@ Client::Options::threads(int val) { ...@@ -710,12 +715,6 @@ Client::Options::threads(int val) {
return *this; return *this;
} }
Client::Options&
Client::Options::maxConnections(int val) {
maxConnections_ = val;
return *this;
}
Client::Options& Client::Options&
Client::Options::keepAlive(bool val) { Client::Options::keepAlive(bool val) {
keepAlive_ = val; keepAlive_ = val;
...@@ -740,6 +739,7 @@ Client::options() { ...@@ -740,6 +739,7 @@ Client::options() {
void void
Client::init(const Client::Options& options) { Client::init(const Client::Options& options) {
pool.init(options.maxConnectionsPerHost_);
transport_.reset(new Transport); transport_.reset(new Transport);
io_.init(options.threads_, transport_); io_.init(options.threads_, transport_);
io_.start(); io_.start();
...@@ -798,9 +798,12 @@ Client::doRequest( ...@@ -798,9 +798,12 @@ Client::doRequest(
if (conn == nullptr) { if (conn == nullptr) {
return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) { return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) {
auto entry = requestsQueue.allocEntry( Guard guard(queuesLock);
auto& queue = requestsQueues[s.first];
auto entry = queue.allocEntry(
Connection::RequestData(std::move(resolve), std::move(reject), request, timeout, nullptr)); Connection::RequestData(std::move(resolve), std::move(reject), request, timeout, nullptr));
requestsQueue.push(entry); queue.push(entry);
}); });
} }
else { else {
...@@ -827,37 +830,36 @@ Client::doRequest( ...@@ -827,37 +830,36 @@ Client::doRequest(
void void
Client::processRequestQueue() { Client::processRequestQueue() {
std::vector<Queue<Connection::RequestData>::Entry *> skippedRequests; Guard guard(queuesLock);
for (;;) { for (auto& queues: requestsQueues) {
const auto& domain = queues.first;
auto& queue = queues.second;
auto entry = requestsQueue.pop(); if (queue.empty()) continue;
if (!entry) {
break;
}
auto& req = entry->data(); for (;;) {
auto resource = req.request.resource(); auto conn = pool.pickConnection(domain);
auto s = splitUrl(resource); if (!conn)
auto conn = pool.pickConnection(s.first); break;
if (!conn) {
skippedRequests.push_back(entry);
continue;
}
conn->performImpl( auto& queue = queues.second;
req.request, auto entry = queue.popSafe();
req.timeout, if (!entry) {
std::move(req.resolve), std::move(req.reject), pool.releaseConnection(conn);
[=]() { break;
pool.releaseConnection(conn); }
processRequestQueue();
});
delete entry;
}
for (auto& skipped: skippedRequests) { auto& req = entry->data();
requestsQueue.push(skipped); conn->performImpl(
req.request,
req.timeout,
std::move(req.resolve), std::move(req.reject),
[=]() {
pool.releaseConnection(conn);
processRequestQueue();
});
}
} }
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment