Commit 761f43b4 authored by octal's avatar octal

Cleaned-up the Reactor. Reactor::Key will now encode the index of the worker...

Cleaned-up the Reactor. Reactor::Key will now encode the index of the worker for the AsyncImpl to allow for right handler dispatching
parent 0a5390e5
......@@ -77,11 +77,11 @@ private:
std::unique_ptr<std::thread> acceptThread;
size_t workers_;
Io::ServiceGroup io_;
std::shared_ptr<Transport> transport_;
Io::ServiceGroup io_;
std::shared_ptr<Handler> handler_;
Aio::Reactor reactor_;
std::shared_ptr<Aio::Reactor> reactor_;
Aio::Reactor::Key transportKey;
void handleNewConnection();
......
......@@ -110,37 +110,13 @@ public:
friend class SyncImpl;
friend class AsyncImpl;
std::shared_ptr<Handler> handler() const {
return handler_;
}
void registerFd(
Fd fd, Polling::NotifyOn interest, Polling::Tag tag,
Polling::Mode mode = Polling::Mode::Level);
void registerFdOneShot(
Fd fd, Polling::NotifyOn interest, Polling::Tag tag,
Polling::Mode mode = Polling::Mode::Level);
void registerFd(
Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level);
void registerFdOneShot(
Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level);
void modifyFd(
Fd fd, Polling::NotifyOn interest, Polling::Tag tag,
Polling::Mode mode = Polling::Mode::Level);
void modifyFd(
Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level);
size_t index() const {
return index_;
uint64_t data() const {
return data_;
}
private:
Key(size_t index, const std::shared_ptr<Handler>& handler);
size_t index_;
std::shared_ptr<Handler> handler_;
Key(uint64_t data);
uint64_t data_;
};
......@@ -151,7 +127,7 @@ public:
Key addHandler(const std::shared_ptr<Handler>& handler);
std::shared_ptr<Handler> handler(const Key& key, Fd fd);
std::vector<std::shared_ptr<Handler>> handlers(const Key& key);
void registerFd(
const Key& key, Fd fd, Polling::NotifyOn interest, Polling::Tag tag,
......@@ -182,9 +158,7 @@ public:
private:
Impl* impl() const;
std::unique_ptr<Impl> impl_;
};
class ExecutionContext {
......@@ -244,7 +218,6 @@ public:
private:
Reactor* reactor_;
Context context_;
Reactor::Key key_;
};
......
This diff is collapsed.
......@@ -65,6 +65,7 @@ Transport::onReady(const Aio::FdSet& fds) {
else if (entry.isReadable()) {
auto tag = entry.getTag();
auto val = tag.value();
if (isPeerFd(tag)) {
auto& peer = getPeer(tag);
handleIncoming(peer);
......@@ -88,7 +89,7 @@ Transport::onReady(const Aio::FdSet& fds) {
throw std::runtime_error("Assertion Error: could not find write data");
}
key().modifyFd(fd, NotifyOn::Read, Polling::Mode::Edge);
reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge);
auto& write = it->second;
asyncWriteImpl(fd, write, Retry);
......@@ -204,7 +205,7 @@ Transport::asyncWriteImpl(
WriteEntry(std::move(deferred), buffer.detach(totalWritten), flags)));
}
key().modifyFd(fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
}
else {
cleanUp();
......@@ -269,7 +270,7 @@ Transport::armTimerMsImpl(TimerEntry entry) {
return;
}
key().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)));
}
......
......@@ -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,11 +190,8 @@ Listener::bind(const Address& address) {
transport_.reset(new Transport(handler_));
reactor_.init(Aio::AsyncContext(workers_));
transportKey = reactor_.addHandler(transport_);
//io_.init(workers_, transport_);
io_.start();
reactor_->init(Aio::AsyncContext(workers_));
transportKey = reactor_->addHandler(transport_);
return true;
}
......@@ -204,7 +203,7 @@ Listener::isBound() const {
void
Listener::run() {
reactor_.run();
reactor_->run();
for (;;) {
std::vector<Polling::Event> events;
......@@ -239,7 +238,7 @@ Listener::runThreaded() {
void
Listener::shutdown() {
if (shutdownFd.isBound()) shutdownFd.notify();
io_.shutdown();
reactor_->shutdown();
}
Async::Promise<Listener::Load>
......@@ -316,8 +315,9 @@ Listener::handleNewConnection() {
void
Listener::dispatchPeer(const std::shared_ptr<Peer>& peer) {
auto handler = reactor_.handler(transportKey, peer->fd());
auto transport = std::static_pointer_cast<Transport>(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