Commit e703fc45 authored by octal's avatar octal

Started implementing timeout for http client

parent 4dd9f900
......@@ -19,14 +19,14 @@ int main() {
using namespace Net::Http;
constexpr size_t Requests = 1000;
constexpr size_t Requests = 5000;
std::atomic<int> responsesReceived(0);
client.init(opts);
for (int i = 0; i < Requests; ++i) {
client.get(client
.request("/ping")
.cookie(Cookie("FOO", "bar")))
.cookie(Cookie("FOO", "bar")), std::chrono::milliseconds(1000))
.then([&](const Http::Response& response) {
responsesReceived.fetch_add(1);
//std::cout << "code = " << response.code() << std::endl;
......
......@@ -9,6 +9,7 @@
#include "os.h"
#include "http.h"
#include "io.h"
#include "timer_pool.h"
#include <atomic>
#include <sys/types.h>
#include <sys/socket.h>
......@@ -35,17 +36,20 @@ struct Connection {
Async::Resolver resolve, Async::Rejection reject,
const Http::Request& request,
std::string host,
std::chrono::milliseconds timeout,
OnDone onDone)
: resolve(std::move(resolve))
, reject(std::move(reject))
, request(request)
, host(std::move(host))
, timeout(timeout)
, onDone(std::move(onDone))
{ }
Async::Resolver resolve;
Async::Rejection reject;
std::string host;
std::chrono::milliseconds timeout;
Http::Request request;
OnDone onDone;
};
......@@ -77,11 +81,13 @@ struct Connection {
Async::Promise<Response> perform(
const Http::Request& request,
std::string host,
std::chrono::milliseconds timeout,
OnDone onDone);
void performImpl(
const Http::Request& request,
std::string host,
std::chrono::milliseconds timeout,
Async::Resolver resolve,
Async::Rejection reject,
OnDone onDone);
......@@ -91,10 +97,12 @@ struct Connection {
private:
void processRequestQueue();
std::atomic<uint32_t> state_;
ConnectionState connectionState_;
std::shared_ptr<Transport> transport_;
Queue<RequestData> requestsQueue;
Net::TimerPool timerPool_;
};
struct ConnectionPool {
......@@ -125,6 +133,7 @@ public:
void asyncSendRequest(
Fd fd,
std::shared_ptr<TimerPool::Entry> timer,
const Buffer& buffer,
Async::Resolver resolve,
Async::Rejection reject,
......@@ -159,11 +168,13 @@ private:
InflightRequest(
Async::Resolver resolve, Async::Rejection reject,
Fd fd,
std::shared_ptr<TimerPool::Entry> timer,
const Buffer& buffer,
OnResponseParsed onParsed = nullptr)
: resolve_(std::move(resolve))
, reject(std::move(reject))
, fd(fd)
, timer(std::move(timer))
, buffer(buffer)
, onParsed(onParsed)
{
......@@ -185,6 +196,7 @@ private:
Async::Resolver resolve_;
Async::Rejection reject;
Fd fd;
std::shared_ptr<TimerPool::Entry> timer;
Buffer buffer;
OnResponseParsed onParsed;
......@@ -198,6 +210,7 @@ private:
std::unordered_map<Fd, PendingConnection> pendingConnections;
std::unordered_map<Fd, std::deque<InflightRequest>> inflightRequests;
std::unordered_map<Fd, Fd> timeouts;
void asyncSendRequestImpl(InflightRequest& req, WriteStatus status = FirstTry);
......@@ -205,6 +218,7 @@ private:
void handleConnectionQueue();
void handleIncoming(Fd fd);
void handleResponsePacket(Fd fd, const char* buffer, size_t totalBytes);
void handleTimeout(Fd fd);
};
......
/* timer_pool.h
Mathieu Stefani, 09 février 2016
A pool of timer fd to avoid creating fds everytime we need a timer and
thus reduce the total number of system calls.
Most operations are lock-free except resize operations needed when the
pool is empty, in which case it's blocking but we expect it to be rare.
*/
#pragma once
#include <memory>
#include <vector>
#include <mutex>
#include <atomic>
#include <unistd.h>
#include "os.h"
namespace Net {
namespace Default {
static constexpr size_t InitialPoolSize = 128;
}
class TimerPool {
public:
TimerPool(size_t initialSize = Default::InitialPoolSize);
struct Entry {
friend class TimerPool;
Fd fd;
Entry()
: fd(-1)
{
state.store(static_cast<uint32_t>(State::Idle));
}
~Entry() {
if (fd != -1)
close(fd);
}
void initialize();
template<typename Duration>
void arm(Duration duration) {
if (fd == -1) return;
armMs(std::chrono::duration_cast<std::chrono::milliseconds>(duration));
}
void disarm();
private:
void armMs(std::chrono::milliseconds value);
enum class State : uint32_t { Idle, Used };
std::atomic<uint32_t> state;
};
std::shared_ptr<Entry> pickTimer();
void releaseTimer(const std::shared_ptr<Entry>& timer);
private:
std::vector<std::shared_ptr<Entry>> timers;
};
} // namespace Net
......@@ -98,7 +98,17 @@ Transport::onReady(const Io::FdSet& fds) {
else if (entry.isReadable()) {
auto tag = entry.getTag();
auto fd = tag.value();
handleIncoming(fd);
auto reqIt = inflightRequests.find(fd);
if (reqIt != std::end(inflightRequests))
handleIncoming(fd);
else {
auto timerIt = timeouts.find(fd);
if (timerIt != std::end(timeouts))
handleTimeout(fd);
else {
throw std::runtime_error("Unknown fd");
}
}
}
else if (entry.isWritable()) {
auto tag = entry.getTag();
......@@ -150,18 +160,19 @@ Transport::asyncConnect(Fd fd, const struct sockaddr* address, socklen_t addr_le
void
Transport::asyncSendRequest(
Fd fd,
std::shared_ptr<TimerPool::Entry> timer,
const Buffer& buffer,
Async::Resolver resolve,
Async::Rejection reject,
OnResponseParsed onParsed) {
if (std::this_thread::get_id() != io()->thread()) {
InflightRequest req(std::move(resolve), std::move(reject), fd, buffer.detach(), std::move(onParsed));
InflightRequest req(std::move(resolve), std::move(reject), fd, std::move(timer), buffer.detach(), std::move(onParsed));
auto detached = buffer.detach();
auto *e = requestsQueue.allocEntry(std::move(req));
requestsQueue.push(e);
} else {
InflightRequest req(std::move(resolve), std::move(reject), fd, buffer, std::move(onParsed));
InflightRequest req(std::move(resolve), std::move(reject), fd, std::move(timer), buffer, std::move(onParsed));
asyncSendRequestImpl(req);
}
......@@ -204,6 +215,12 @@ Transport::asyncSendRequestImpl(
if (totalWritten == len) {
cleanUp();
auto& queue = inflightRequests[fd];
auto timer = req.timer;
if (timer) {
auto timerFd = timer->fd;
timeouts[timerFd] = fd;
io()->registerFd(timerFd, NotifyOn::Read, Polling::Mode::Edge);
}
queue.push_back(std::move(req));
break;
}
......@@ -293,11 +310,28 @@ Transport::handleResponsePacket(Fd fd, const char* buffer, size_t totalBytes) {
auto &req = queue.front();
req.feed(buffer, totalBytes);
if (req.parser->parse() == Private::State::Done) {
req.timer->disarm();
req.resolve(std::move(req.parser->response));
queue.pop_front();
}
}
void
Transport::handleTimeout(Fd fd) {
auto timerIt = timeouts.find(fd);
auto reqFd = timerIt->second;
auto reqIt = inflightRequests.find(reqFd);
if (reqIt == std::end(inflightRequests))
throw std::runtime_error("Internal condition violation, received timeout for a non-inflight request");
auto& queue = reqIt->second;
auto& req = queue.front();
req.reject(std::runtime_error("Timeout"));
queue.pop_front();
timeouts.erase(fd);
}
void
Connection::connect(Net::Address addr)
{
......@@ -372,14 +406,15 @@ Async::Promise<Response>
Connection::perform(
const Http::Request& request,
std::string host,
std::chrono::milliseconds timeout,
OnDone onDone) {
return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) {
if (!isConnected()) {
auto* entry = requestsQueue.allocEntry(
RequestData(std::move(resolve), std::move(reject), request, std::move(host), std::move(onDone)));
RequestData(std::move(resolve), std::move(reject), request, std::move(host), timeout, std::move(onDone)));
requestsQueue.push(entry);
} else {
performImpl(request, std::move(host), std::move(resolve), std::move(reject), std::move(onDone));
performImpl(request, std::move(host), timeout, std::move(resolve), std::move(reject), std::move(onDone));
}
});
}
......@@ -388,6 +423,7 @@ void
Connection::performImpl(
const Http::Request& request,
std::string host,
std::chrono::milliseconds timeout,
Async::Resolver resolve,
Async::Rejection reject,
OnDone onDone) {
......@@ -398,7 +434,13 @@ Connection::performImpl(
reject(std::runtime_error("Could not write request"));
auto buffer = buf.buffer();
transport_->asyncSendRequest(fd, buffer, std::move(resolve), std::move(reject), std::move(onDone));
std::shared_ptr<TimerPool::Entry> timer(nullptr);
if (timeout.count() > 0) {
timer = timerPool_.pickTimer();
timer->arm(timeout);
}
transport_->asyncSendRequest(fd, timer, buffer, std::move(resolve), std::move(reject), std::move(onDone));
}
void
......@@ -409,7 +451,7 @@ Connection::processRequestQueue() {
auto &req = entry->data();
performImpl(req.request,
std::move(req.host), std::move(req.resolve), std::move(req.reject), std::move(req.onDone));
std::move(req.host), req.timeout, std::move(req.resolve), std::move(req.reject), std::move(req.onDone));
}
}
......@@ -565,7 +607,7 @@ Client::doRequest(
if (conn == nullptr) {
return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) {
auto entry = requestsQueue.allocEntry(
Connection::RequestData(std::move(resolve), std::move(reject), request, host_, nullptr));
Connection::RequestData(std::move(resolve), std::move(reject), request, host_, timeout, nullptr));
requestsQueue.push(entry);
});
}
......@@ -583,7 +625,7 @@ Client::doRequest(
if (!conn->isConnected())
conn->connect(addr_);
return conn->perform(request, host_, [=]() {
return conn->perform(request, host_, timeout, [=]() {
pool.releaseConnection(conn);
processRequestQueue();
});
......@@ -606,6 +648,7 @@ Client::processRequestQueue() {
auto& req = entry->data();
conn->performImpl(
req.request, std::move(req.host),
req.timeout,
std::move(req.resolve), std::move(req.reject),
[=]() {
pool.releaseConnection(conn);
......
/* timer_pool.cc
Mathieu Stefani, 09 février 2016
Implementation of the timer pool
*/
#include "timer_pool.h"
#include <sys/timerfd.h>
namespace Net {
void
TimerPool::Entry::initialize() {
if (fd == -1) {
fd = TRY_RET(timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK));
}
}
void
TimerPool::Entry::disarm() {
if (fd == -1) return;
itimerspec spec;
spec.it_interval.tv_sec = 0;
spec.it_interval.tv_nsec = 0;
spec.it_value.tv_sec = 0;
spec.it_value.tv_nsec = 0;
TRY(timerfd_settime(fd, 0, &spec, 0));
}
void
TimerPool::Entry::armMs(std::chrono::milliseconds value)
{
itimerspec spec;
spec.it_interval.tv_sec = 0;
spec.it_interval.tv_nsec = 0;
if (value.count() < 1000) {
spec.it_value.tv_sec = 0;
spec.it_value.tv_nsec
= std::chrono::duration_cast<std::chrono::nanoseconds>(value).count();
} else {
spec.it_value.tv_sec
= std::chrono::duration_cast<std::chrono::seconds>(value).count();
spec.it_value.tv_nsec = 0;
}
TRY(timerfd_settime(fd, 0, &spec, 0));
}
TimerPool::TimerPool(size_t initialSize)
{
for (size_t i = 0; i < initialSize; ++i) {
timers.push_back(std::make_shared<TimerPool::Entry>());
}
}
std::shared_ptr<TimerPool::Entry>
TimerPool::pickTimer() {
for (auto& entry: timers) {
auto curState = static_cast<uint32_t>(TimerPool::Entry::State::Idle);
auto newState = static_cast<uint32_t>(TimerPool::Entry::State::Used);
if (entry->state.compare_exchange_strong(curState, newState)) {
entry->initialize();
return entry;
}
}
return nullptr;
}
void
TimerPool::releaseTimer(const std::shared_ptr<Entry>& timer) {
timer->state.store(static_cast<uint32_t>(TimerPool::Entry::State::Idle));
}
} // namespace Net
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