Commit e673be32 authored by octal's avatar octal

Improved the client architecture and fixed a data-race in async

parent 4dd9f900
......@@ -15,11 +15,11 @@ int main() {
Http::Experimental::Client client("http://supnetwork.org:9080");
auto opts = Http::Experimental::Client::options()
.threads(1)
.maxConnections(20);
.maxConnections(64);
using namespace Net::Http;
constexpr size_t Requests = 1000;
constexpr size_t Requests = 10000;
std::atomic<int> responsesReceived(0);
client.init(opts);
......@@ -30,13 +30,17 @@ int main() {
.then([&](const Http::Response& response) {
responsesReceived.fetch_add(1);
//std::cout << "code = " << response.code() << std::endl;
// std::cout << "body = " << response.body() << std::endl;
//std::cout << "body = " << response.body() << std::endl;
}, Async::NoExcept);
const auto count = i + 1;
if (count % 10 == 0)
std::cout << "Sent " << count << " requests" << std::endl;
}
std::cout << "Sent " << Requests << " requests" << std::endl;
for (;;) {
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Received " << responsesReceived.load() << " responses" << std::endl;
auto count = responsesReceived.load();
std::cout << "Received " << count << " responses" << std::endl;
if (count == Requests) break;
}
client.shutdown();
}
......@@ -12,6 +12,7 @@
#include <memory>
#include <atomic>
#include <vector>
#include <mutex>
#include "optional.h"
#include "typeid.h"
......@@ -170,6 +171,19 @@ namespace Async {
State state;
std::exception_ptr exc;
/*
* We need this lock because a Promise might be resolved or rejected from a thread A
* while a continuation to the same Promise (Core) might be attached at the same from
* a thread B. If that's the case, then we need to serialize operations so that we
* avoid a race-condition.
*
* Since we have a lock, we have a blocking progress guarantee but I don't expect this
* to be a major bottleneck as I don't expect major contention on the lock
* If it ends up being a bottlenick, try @improving it by experimenting with a lock-free
* scheme
*/
std::mutex mtx;
std::vector<std::shared_ptr<Request>> requests;
TypeId id;
......@@ -536,10 +550,13 @@ namespace Async {
* than runtime. However, since types are erased, this looks like
* a difficult task
*/
if (core_->isVoid())
if (core_->isVoid()) {
throw Error("Attempt to resolve a void promise with arguments");
}
std::unique_lock<std::mutex> guard(core_->mtx);
core_->construct<Type>(std::forward<Arg>(arg));
for (const auto& req: core_->requests) {
req->resolve(core_);
}
......@@ -554,6 +571,7 @@ namespace Async {
if (!core_->isVoid())
throw Error("Attempt ro resolve a non-void promise with no argument");
std::unique_lock<std::mutex> guard(core_->mtx);
core_->state = State::Fulfilled;
for (const auto& req: core_->requests) {
req->resolve(core_);
......@@ -578,6 +596,7 @@ namespace Async {
if (core_->state != State::Pending)
throw Error("Attempt to reject a fulfilled promise");
std::unique_lock<std::mutex> guard(core_->mtx);
core_->exc = std::make_exception_ptr(exc);
core_->state = State::Rejected;
for (const auto& req: core_->requests) {
......@@ -716,6 +735,7 @@ namespace Async {
std::forward<ResolveFunc>(resolveFunc),
std::forward<RejectFunc>(rejectFunc)));
std::unique_lock<std::mutex> guard(core_->mtx);
if (isFulfilled()) {
req->resolve(core_);
}
......@@ -723,9 +743,9 @@ namespace Async {
req->reject(core_);
}
core_->requests.push_back(req);
core_->requests.push_back(req);
return promise;
return promise;
}
private:
......
......@@ -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>
......@@ -23,30 +24,42 @@ namespace Experimental {
class ConnectionPool;
class Transport;
struct Connection {
struct Connection : public std::enable_shared_from_this<Connection> {
friend class ConnectionPool;
typedef std::function<void()> OnDone;
Connection()
: fd(-1)
, connectionState_(NotConnected)
, inflightCount(0)
, responsesReceived(0)
{
state_.store(static_cast<uint32_t>(State::Idle));
}
struct RequestData {
RequestData(
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;
Http::Request request;
std::string host;
std::chrono::milliseconds timeout;
OnDone onDone;
};
......@@ -61,13 +74,6 @@ struct Connection {
Connected
};
Connection()
: fd(-1)
, connectionState_(NotConnected)
{
state_.store(static_cast<uint32_t>(State::Idle));
}
void connect(Net::Address addr);
void close();
bool isConnected() const;
......@@ -77,24 +83,58 @@ 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);
Fd fd;
void handleResponsePacket(const char* buffer, size_t totalBytes);
void handleTimeout();
std::string dump() const;
private:
std::atomic<int> inflightCount;
std::atomic<int> responsesReceived;
struct sockaddr_in saddr;
void processRequestQueue();
struct RequestEntry {
RequestEntry(
Async::Resolver resolve, Async::Rejection reject,
std::shared_ptr<TimerPool::Entry> timer,
OnDone onDone)
: resolve(std::move(resolve))
, reject(std::move(reject))
, timer(std::move(timer))
, onDone(std::move(onDone))
{ }
Async::Resolver resolve;
Async::Rejection reject;
std::shared_ptr<TimerPool::Entry> timer;
OnDone onDone;
};
std::atomic<uint32_t> state_;
ConnectionState connectionState_;
std::shared_ptr<Transport> transport_;
Queue<RequestData> requestsQueue;
std::deque<RequestEntry> inflightRequests;
Net::TimerPool timerPool_;
Private::Parser<Http::Response> parser_;
};
struct ConnectionPool {
......@@ -121,14 +161,12 @@ public:
void registerPoller(Polling::Epoll& poller);
Async::Promise<void>
asyncConnect(Fd fd, const struct sockaddr* address, socklen_t addr_len);
asyncConnect(const std::shared_ptr<Connection>& connection, const struct sockaddr* address, socklen_t addr_len);
void asyncSendRequest(
Fd fd,
const Buffer& buffer,
Async::Resolver resolve,
Async::Rejection reject,
OnResponseParsed onParsed);
Async::Promise<ssize_t> asyncSendRequest(
const std::shared_ptr<Connection>& connection,
std::shared_ptr<TimerPool::Entry> timer,
const Buffer& buffer);
private:
......@@ -137,74 +175,61 @@ private:
Retry
};
struct PendingConnection {
PendingConnection(
struct ConnectionEntry {
ConnectionEntry(
Async::Resolver resolve, Async::Rejection reject,
Fd fd, const struct sockaddr* addr, socklen_t addr_len)
std::shared_ptr<Connection> connection, const struct sockaddr* addr, socklen_t addr_len)
: resolve(std::move(resolve))
, reject(std::move(reject))
, fd(fd)
, connection(std::move(connection))
, addr(addr)
, addr_len(addr_len)
{ }
Async::Resolver resolve;
Async::Rejection reject;
Fd fd;
std::shared_ptr<Connection> connection;
const struct sockaddr* addr;
socklen_t addr_len;
};
struct InflightRequest {
InflightRequest(
struct RequestEntry {
RequestEntry(
Async::Resolver resolve, Async::Rejection reject,
Fd fd,
const Buffer& buffer,
OnResponseParsed onParsed = nullptr)
: resolve_(std::move(resolve))
std::shared_ptr<Connection> connection,
std::shared_ptr<TimerPool::Entry> timer,
const Buffer& buffer)
: resolve(std::move(resolve))
, reject(std::move(reject))
, fd(fd)
, connection(std::move(connection))
, timer(std::move(timer))
, buffer(buffer)
, onParsed(onParsed)
{
}
void feed(const char* buffer, size_t totalBytes) {
if (!parser)
parser.reset(new Private::Parser<Http::Response>());
parser->feed(buffer, totalBytes);
}
void resolve(Http::Response response) {
if (onParsed)
onParsed();
resolve_(std::move(response));
}
Async::Resolver resolve_;
Async::Resolver resolve;
Async::Rejection reject;
Fd fd;
std::shared_ptr<Connection> connection;
std::shared_ptr<TimerPool::Entry> timer;
Buffer buffer;
OnResponseParsed onParsed;
std::shared_ptr<Private::Parser<Http::Response>> parser;
};
PollableQueue<InflightRequest> requestsQueue;
PollableQueue<PendingConnection> connectionsQueue;
PollableQueue<RequestEntry> requestsQueue;
PollableQueue<ConnectionEntry> connectionsQueue;
std::unordered_map<Fd, PendingConnection> pendingConnections;
std::unordered_map<Fd, std::deque<InflightRequest>> inflightRequests;
std::unordered_map<Fd, ConnectionEntry> connections;
std::unordered_map<Fd, RequestEntry> requests;
std::unordered_map<Fd, std::shared_ptr<Connection>> timeouts;
void asyncSendRequestImpl(InflightRequest& req, WriteStatus status = FirstTry);
void asyncSendRequestImpl(const RequestEntry& req, WriteStatus status = FirstTry);
void handleRequestsQueue();
void handleConnectionQueue();
void handleIncoming(Fd fd);
void handleResponsePacket(Fd fd, const char* buffer, size_t totalBytes);
void handleIncoming(const std::shared_ptr<Connection>& connection);
void handleResponsePacket(const std::shared_ptr<Connection>& connection, const char* buffer, size_t totalBytes);
void handleTimeout(const std::shared_ptr<Connection>& connection);
};
......@@ -260,6 +285,7 @@ public:
void shutdown();
private:
Io::ServiceGroup io_;
std::string url_;
std::string host_;
......
/* 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"
#include "io.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)
, registered(false)
{
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();
void registerIo(Io::Service* io) {
if (!registered) {
io->registerFd(fd, Polling::NotifyOn::Read);
}
registered = true;
}
private:
void armMs(std::chrono::milliseconds value);
enum class State : uint32_t { Idle, Used };
std::atomic<uint32_t> state;
bool registered;
};
std::shared_ptr<Entry> pickTimer();
void releaseTimer(const std::shared_ptr<Entry>& timer);
private:
std::vector<std::shared_ptr<Entry>> timers;
};
} // namespace Net
This diff is collapsed.
/* 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
......@@ -2,6 +2,9 @@
#include "async.h"
#include <thread>
#include <algorithm>
#include <deque>
#include <mutex>
#include <condition_variable>
Async::Promise<int> doAsync(int N)
{
......@@ -291,3 +294,139 @@ TEST(async_test, rethrow_test) {
ASSERT_TRUE(p2.isRejected());
}
template<typename T>
struct MessageQueue {
public:
template<typename U>
void push(U&& arg) {
std::unique_lock<std::mutex> guard(mtx);
q.push_back(std::forward<U>(arg));
cv.notify_one();
}
T pop() {
std::unique_lock<std::mutex> lock(mtx);
cv.wait(lock, [=]() { return !q.empty(); });
T out = std::move(q.front());
q.pop_front();
return out;
}
bool tryPop(T& out, std::chrono::milliseconds timeout) {
std::unique_lock<std::mutex> lock(mtx);
if (!cv.wait_for(lock, timeout, [=]() { return !q.empty(); }))
return false;
out = std::move(q.front());
q.pop_front();
return true;
}
private:
std::deque<T> q;
std::mutex mtx;
std::condition_variable cv;
};
struct Worker {
public:
~Worker() {
thread->join();
}
void start() {
shutdown.store(false);
thread.reset(new std::thread([=]() { run(); }));
}
void stop() {
shutdown.store(true);
}
Async::Promise<int> doWork(int seq) {
return Async::Promise<int>([=](Async::Resolver& resolve, Async::Rejection& reject) {
queue.push(new WorkRequest(std::move(resolve), std::move(reject), seq));
});
}
private:
void run() {
while (!shutdown) {
WorkRequest *request;
if (queue.tryPop(request, std::chrono::milliseconds(200))) {
request->resolve(request->seq);
delete request;
}
}
}
struct WorkRequest {
WorkRequest(Async::Resolver resolve, Async::Rejection reject, int seq)
: resolve(std::move(resolve))
, reject(std::move(reject))
, seq(seq)
{
}
int seq;
Async::Resolver resolve;
Async::Rejection reject;
};
std::atomic<bool> shutdown;
MessageQueue<WorkRequest*> queue;
std::random_device rd;
std::unique_ptr<std::thread> thread;
};
TEST(async_test, stress_multithreaded_test) {
static constexpr size_t OpsPerThread = 100000;
static constexpr size_t Workers = 6;
static constexpr size_t Ops = OpsPerThread * Workers;
std::cout << "Starting stress testing promises, hang on, this test might take some time to complete" << std::endl;
std::cout << "=================================================" << std::endl;
std::cout << "Parameters for the test: " << std::endl;
std::cout << "Workers -> " << Workers << std::endl;
std::cout << "OpsPerThread -> " << OpsPerThread << std::endl;
std::cout << "Total Ops -> " << Ops << std::endl;
std::cout << "=================================================" << std::endl;
std::cout << std::endl << std::endl;
std::vector<std::unique_ptr<Worker>> workers;
for (size_t i = 0; i < Workers; ++i) {
std::unique_ptr<Worker> wrk(new Worker);
wrk->start();
workers.push_back(std::move(wrk));
}
std::vector<Async::Promise<int>> promises;
std::atomic<int> resolved(0);
size_t wrkIndex = 0;
for (size_t i = 0; i < Ops; ++i) {
auto &wrk = workers[wrkIndex];
wrk->doWork(i).then([&](int seq) {
++resolved;
}, Async::NoExcept);
wrkIndex = (wrkIndex + 1) % Workers;
}
for (;;) {
auto r = resolved.load();
std::cout << r << " promises resolved" << std::endl;
if (r == Ops) break;
std::this_thread::sleep_for(std::chrono::milliseconds(500));
}
std::cout << "Stopping worker" << std::endl;
for (auto& wrk: workers) {
wrk->stop();
}
}
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