Commit 537f30ab authored by Mathieu Stefani's avatar Mathieu Stefani Committed by GitHub

Merge pull request #4 from oktal/io-reactor

Io reactor
parents 89146baf 877b03e1
......@@ -12,6 +12,7 @@
#include "flags.h"
#include "async.h"
#include "io.h"
#include "reactor.h"
#include <vector>
#include <memory>
#include <thread>
......@@ -66,8 +67,6 @@ public:
void pinWorker(size_t worker, const CpuSet& set);
private:
struct TransportFactory;
Address addr_;
int listen_fd;
int backlog_;
......@@ -78,10 +77,12 @@ private:
std::unique_ptr<std::thread> acceptThread;
size_t workers_;
Io::ServiceGroup io_;
std::shared_ptr<Transport> transport_;
std::shared_ptr<Handler> handler_;
std::shared_ptr<Aio::Reactor> reactor_;
Aio::Reactor::Key transportKey;
void handleNewConnection();
void dispatchPeer(const std::shared_ptr<Peer>& peer);
......
/*
Mathieu Stefani, 15 juin 2016
A lightweight implementation of the Reactor design-pattern.
The main goal of this component is to provide an solid abstraction
that can be used internally and by client code to dispatch I/O events
to callbacks and handlers, in an efficient way.
*/
#pragma once
#include "flags.h"
#include "os.h"
#include "net.h"
#include "prototype.h"
#include <thread>
#include <mutex>
#include <memory>
#include <unordered_map>
#include <sys/time.h>
#include <sys/resource.h>
#include <atomic>
namespace Aio {
// A set of fds that are ready
class FdSet {
public:
FdSet(std::vector<Polling::Event>&& events)
{
events_.reserve(events.size());
for (auto &&event: events) {
events_.push_back(std::move(event));
}
}
struct Entry : private Polling::Event {
Entry(Polling::Event&& event)
: Polling::Event(std::move(event))
{ }
bool isReadable() const {
return flags.hasFlag(Polling::NotifyOn::Read);
}
bool isWritable() const {
return flags.hasFlag(Polling::NotifyOn::Write);
}
bool isHangup() const {
return flags.hasFlag(Polling::NotifyOn::Hangup);
}
Fd getFd() const { return this->fd; }
Polling::Tag getTag() const { return this->tag; }
};
typedef std::vector<Entry>::iterator iterator;
typedef std::vector<Entry>::const_iterator const_iterator;
size_t size() const {
return events_.size();
}
const Entry& at(size_t index) const {
return events_.at(index);
}
const Entry& operator[](size_t index) const {
return events_[index];
}
iterator begin() {
return events_.begin();
}
iterator end() {
return events_.end();
}
const_iterator begin() const {
return events_.begin();
}
const_iterator end() const {
return events_.end();
}
private:
std::vector<Entry> events_;
};
class Handler;
class ExecutionContext;
class Reactor : public std::enable_shared_from_this<Reactor> {
public:
class Impl;
Reactor();
~Reactor();
struct Key {
Key();
friend class Reactor;
friend class Impl;
friend class SyncImpl;
friend class AsyncImpl;
uint64_t data() const {
return data_;
}
private:
Key(uint64_t data);
uint64_t data_;
};
static std::shared_ptr<Reactor> create();
void init();
void init(const ExecutionContext& context);
Key addHandler(const std::shared_ptr<Handler>& handler);
std::vector<std::shared_ptr<Handler>> handlers(const Key& key);
void registerFd(
const Key& key, Fd fd, Polling::NotifyOn interest, Polling::Tag tag,
Polling::Mode mode = Polling::Mode::Level);
void registerFdOneShot(
const Key& key, Fd fd, Polling::NotifyOn intereset, Polling::Tag tag,
Polling::Mode mode = Polling::Mode::Level);
void registerFd(
const Key& key, Fd fd, Polling::NotifyOn interest,
Polling::Mode mode = Polling::Mode::Level);
void registerFdOneShot(
const Key& key, Fd fd, Polling::NotifyOn interest,
Polling::Mode mode = Polling::Mode::Level);
void modifyFd(
const Key& key, Fd fd, Polling::NotifyOn interest,
Polling::Mode mode = Polling::Mode::Level);
void modifyFd(
const Key& key, Fd fd, Polling::NotifyOn interest, Polling::Tag tag,
Polling::Mode mode = Polling::Mode::Level);
void runOnce();
void run();
void shutdown();
private:
Impl* impl() const;
std::unique_ptr<Impl> impl_;
};
class ExecutionContext {
public:
virtual Reactor::Impl* makeImpl(Reactor* reactor) const = 0;
};
class SyncContext : public ExecutionContext {
public:
Reactor::Impl* makeImpl(Reactor* reactor) const;
};
class AsyncContext : public ExecutionContext {
public:
AsyncContext(size_t threads)
: threads_(threads)
{ }
Reactor::Impl* makeImpl(Reactor* reactor) const;
static AsyncContext singleThreaded();
private:
size_t threads_;
};
class Handler : public Prototype<Handler> {
public:
friend class Reactor;
friend class SyncImpl;
friend class AsyncImpl;
struct Context {
friend class SyncImpl;
std::thread::id thread() const { return tid; }
private:
std::thread::id tid;
};
virtual void onReady(const FdSet& fds) = 0;
virtual void registerPoller(Polling::Epoll& poller) { }
Reactor* reactor() const {
return reactor_;
}
Context context() const {
return context_;
}
Reactor::Key key() const {
return key_;
};
private:
Reactor* reactor_;
Context context_;
Reactor::Key key_;
};
} // namespace Aio
......@@ -7,6 +7,7 @@
#pragma once
#include "io.h"
#include "reactor.h"
#include "mailbox.h"
#include "optional.h"
#include "async.h"
......@@ -19,7 +20,7 @@ namespace Tcp {
class Peer;
class Handler;
class Transport : public Io::Handler {
class Transport : public Aio::Handler {
public:
Transport(const std::shared_ptr<Tcp::Handler>& handler);
......@@ -28,13 +29,14 @@ public:
void registerPoller(Polling::Epoll& poller);
void handleNewPeer(const std::shared_ptr<Peer>& peer);
void onReady(const Io::FdSet& fds);
void onReady(const Aio::FdSet& fds);
template<typename Buf>
Async::Promise<ssize_t> asyncWrite(Fd fd, const Buf& buffer, int flags = 0) {
// If the I/O operation has been initiated from an other thread, we queue it and we'll process
// it in our own thread so that we make sure that every I/O operation happens in the right thread
const bool isInRightThread = std::this_thread::get_id() == io()->thread();
auto ctx = context();
const bool isInRightThread = std::this_thread::get_id() == ctx.thread();
if (!isInRightThread) {
return Async::Promise<ssize_t>([=](Async::Deferred<ssize_t> deferred) mutable {
BufferHolder holder(buffer);
......@@ -76,7 +78,7 @@ public:
void disarmTimer(Fd fd);
std::shared_ptr<Io::Handler> clone() const;
std::shared_ptr<Aio::Handler> clone() const;
private:
enum WriteStatus {
......
This diff is collapsed.
......@@ -21,7 +21,7 @@ Transport::init(const std::shared_ptr<Tcp::Handler>& handler) {
handler_->associateTransport(this);
}
std::shared_ptr<Io::Handler>
std::shared_ptr<Aio::Handler>
Transport::clone() const {
return std::make_shared<Transport>(handler_->clone());
}
......@@ -36,7 +36,8 @@ Transport::registerPoller(Polling::Epoll& poller) {
void
Transport::handleNewPeer(const std::shared_ptr<Tcp::Peer>& peer) {
const bool isInRightThread = std::this_thread::get_id() == io()->thread();
auto ctx = context();
const bool isInRightThread = std::this_thread::get_id() == ctx.thread();
if (!isInRightThread) {
PeerEntry entry(peer);
auto *e = peersQueue.allocEntry(entry);
......@@ -47,7 +48,7 @@ Transport::handleNewPeer(const std::shared_ptr<Tcp::Peer>& peer) {
}
void
Transport::onReady(const Io::FdSet& fds) {
Transport::onReady(const Aio::FdSet& fds) {
for (const auto& entry: fds) {
if (entry.getTag() == writesQueue.tag()) {
handleWriteQueue();
......@@ -64,6 +65,7 @@ Transport::onReady(const Io::FdSet& fds) {
else if (entry.isReadable()) {
auto tag = entry.getTag();
auto val = tag.value();
if (isPeerFd(tag)) {
auto& peer = getPeer(tag);
handleIncoming(peer);
......@@ -87,7 +89,7 @@ Transport::onReady(const Io::FdSet& fds) {
throw std::runtime_error("Assertion Error: could not find write data");
}
io()->modifyFd(fd, NotifyOn::Read, Polling::Mode::Edge);
reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge);
auto& write = it->second;
asyncWriteImpl(fd, write, Retry);
......@@ -202,7 +204,8 @@ Transport::asyncWriteImpl(
std::make_pair(fd,
WriteEntry(std::move(deferred), buffer.detach(totalWritten), flags)));
}
io()->modifyFd(fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
}
else {
cleanUp();
......@@ -225,7 +228,9 @@ void
Transport::armTimerMs(
Fd fd, std::chrono::milliseconds value,
Async::Deferred<uint64_t> deferred) {
const bool isInRightThread = std::this_thread::get_id() == io()->thread();
auto ctx = context();
const bool isInRightThread = std::this_thread::get_id() == ctx.thread();
TimerEntry entry(fd, value, std::move(deferred));
if (!isInRightThread) {
......@@ -265,7 +270,7 @@ Transport::armTimerMsImpl(TimerEntry entry) {
return;
}
io()->registerFdOneShot(entry.fd, NotifyOn::Read, Polling::Mode::Edge);
reactor()->registerFdOneShot(key(), entry.fd, NotifyOn::Read, Polling::Mode::Edge);
timers.insert(std::make_pair(entry.fd, std::move(entry)));
}
......@@ -311,7 +316,7 @@ Transport::handlePeer(const std::shared_ptr<Peer>& peer) {
peer->associateTransport(this);
handler_->onConnection(peer);
io()->registerFd(fd, NotifyOn::Read | NotifyOn::Shutdown, Polling::Mode::Edge);
reactor()->registerFd(key(), fd, NotifyOn::Read | NotifyOn::Shutdown, Polling::Mode::Edge);
}
void
......
......@@ -75,12 +75,14 @@ void setSocketOptions(Fd fd, Flags<Options> options) {
Listener::Listener()
: listen_fd(-1)
, backlog_(Const::MaxBacklog)
, reactor_(Aio::Reactor::create())
{ }
Listener::Listener(const Address& address)
: addr_(address)
, listen_fd(-1)
, backlog_(Const::MaxBacklog)
, reactor_(Aio::Reactor::create())
{
}
......@@ -188,8 +190,8 @@ Listener::bind(const Address& address) {
transport_.reset(new Transport(handler_));
io_.init(workers_, transport_);
io_.start();
reactor_->init(Aio::AsyncContext(workers_));
transportKey = reactor_->addHandler(transport_);
return true;
}
......@@ -201,6 +203,8 @@ Listener::isBound() const {
void
Listener::run() {
reactor_->run();
for (;;) {
std::vector<Polling::Event> events;
......@@ -234,12 +238,18 @@ Listener::runThreaded() {
void
Listener::shutdown() {
if (shutdownFd.isBound()) shutdownFd.notify();
io_.shutdown();
reactor_->shutdown();
}
Async::Promise<Listener::Load>
Listener::requestLoad(const Listener::Load& old) {
auto loads = io_.load();
auto handlers = reactor_->handlers(transportKey);
std::vector<Async::Promise<rusage>> loads;
for (const auto& handler: handlers) {
auto transport = std::static_pointer_cast<Transport>(handler);
loads.push_back(transport->load());
}
return Async::whenAll(std::begin(loads), std::end(loads)).then([=](const std::vector<rusage>& usages) {
......@@ -248,7 +258,7 @@ Listener::requestLoad(const Listener::Load& old) {
if (old.raw.empty()) {
res.global = 0.0;
for (size_t i = 0; i < io_.size(); ++i) res.workers.push_back(0.0);
for (size_t i = 0; i < handlers.size(); ++i) res.workers.push_back(0.0);
} else {
auto totalElapsed = [](rusage usage) {
......@@ -311,8 +321,9 @@ Listener::handleNewConnection() {
void
Listener::dispatchPeer(const std::shared_ptr<Peer>& peer) {
auto service = io_.service(peer->fd());
auto transport = std::static_pointer_cast<Transport>(service->handler());
auto handlers = reactor_->handlers(transportKey);
auto idx = peer->fd() % handlers.size();
auto transport = std::static_pointer_cast<Transport>(handlers[idx]);
transport->handleNewPeer(peer);
......
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