Commit 8228a32b authored by octal's avatar octal

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

parent 6be5c664
......@@ -23,7 +23,7 @@ int main(int argc, char *argv[]) {
auto opts = Http::Experimental::Client::options()
.threads(1)
.maxConnections(64);
.maxConnectionsPerHost(8);
client.init(opts);
auto resp = client.get(page).cookie(Cookie("FOO", "bar")).send();
......
......@@ -137,6 +137,9 @@ private:
};
struct ConnectionPool {
void init(size_t maxConnsPerHost);
std::shared_ptr<Connection> pickConnection(const std::string& domain);
void releaseConnection(const std::shared_ptr<Connection>& connection);
......@@ -154,6 +157,7 @@ private:
mutable Lock connsLock;
std::unordered_map<std::string, Connections> conns;
size_t maxConnectionsPerHost;
};
class Transport : public Io::Handler {
......@@ -242,7 +246,7 @@ private:
namespace Default {
constexpr int Threads = 1;
constexpr int MaxConnections = 8;
constexpr int MaxConnectionsPerHost = 8;
constexpr bool KeepAlive = true;
}
......@@ -296,19 +300,17 @@ public:
Options()
: threads_(Default::Threads)
, maxConnections_(Default::MaxConnections)
, maxConnectionsPerHost_(Default::MaxConnectionsPerHost)
, keepAlive_(Default::KeepAlive)
{ }
Options& threads(int val);
Options& maxConnections(int val);
Options& keepAlive(bool val);
Options& maxConnectionsPerHost(int val);
private:
int threads_;
int maxConnections_;
int maxConnectionsPerHost_;
bool keepAlive_;
};
......@@ -333,7 +335,12 @@ private:
std::shared_ptr<Transport> transport_;
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);
......
......@@ -205,6 +205,12 @@ public:
return nullptr;
}
bool empty() {
auto *res = tail;
auto* next = res->next.load(std::memory_order_acquire);
return next != nullptr;
}
std::unique_ptr<Entry> popSafe() {
return std::unique_ptr<Entry>(pop());
}
......
......@@ -582,6 +582,11 @@ Connection::processRequestQueue() {
}
void
ConnectionPool::init(size_t maxConnsPerHost) {
maxConnectionsPerHost = maxConnsPerHost;
}
std::shared_ptr<Connection>
ConnectionPool::pickConnection(const std::string& domain) {
Connections pool;
......@@ -591,7 +596,7 @@ ConnectionPool::pickConnection(const std::string& domain) {
auto poolIt = conns.find(domain);
if (poolIt == std::end(conns)) {
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>());
}
......@@ -710,12 +715,6 @@ Client::Options::threads(int val) {
return *this;
}
Client::Options&
Client::Options::maxConnections(int val) {
maxConnections_ = val;
return *this;
}
Client::Options&
Client::Options::keepAlive(bool val) {
keepAlive_ = val;
......@@ -740,6 +739,7 @@ Client::options() {
void
Client::init(const Client::Options& options) {
pool.init(options.maxConnectionsPerHost_);
transport_.reset(new Transport);
io_.init(options.threads_, transport_);
io_.start();
......@@ -798,9 +798,12 @@ Client::doRequest(
if (conn == nullptr) {
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));
requestsQueue.push(entry);
queue.push(entry);
});
}
else {
......@@ -827,37 +830,36 @@ Client::doRequest(
void
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 (!entry) {
break;
}
if (queue.empty()) continue;
auto& req = entry->data();
auto resource = req.request.resource();
auto s = splitUrl(resource);
auto conn = pool.pickConnection(s.first);
if (!conn) {
skippedRequests.push_back(entry);
continue;
}
for (;;) {
auto conn = pool.pickConnection(domain);
if (!conn)
break;
conn->performImpl(
req.request,
req.timeout,
std::move(req.resolve), std::move(req.reject),
[=]() {
pool.releaseConnection(conn);
processRequestQueue();
});
delete entry;
}
auto& queue = queues.second;
auto entry = queue.popSafe();
if (!entry) {
pool.releaseConnection(conn);
break;
}
for (auto& skipped: skippedRequests) {
requestsQueue.push(skipped);
auto& req = entry->data();
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