Commit 877b03e1 authored by octal's avatar octal

Completely got rid of the old Io::ServiceGroup in the Listener in favor of the...

Completely got rid of the old Io::ServiceGroup in the Listener in favor of the new Reactor. Also fixed a bunch of remaining bugs
parent 761f43b4
...@@ -78,7 +78,6 @@ private: ...@@ -78,7 +78,6 @@ private:
size_t workers_; size_t workers_;
std::shared_ptr<Transport> transport_; std::shared_ptr<Transport> transport_;
Io::ServiceGroup io_;
std::shared_ptr<Handler> handler_; std::shared_ptr<Handler> handler_;
std::shared_ptr<Aio::Reactor> reactor_; std::shared_ptr<Aio::Reactor> reactor_;
......
...@@ -136,7 +136,7 @@ struct SyncImpl : public Reactor::Impl { ...@@ -136,7 +136,7 @@ struct SyncImpl : public Reactor::Impl {
handleFds(std::move(events)); handleFds(std::move(events));
timeout = std::chrono::milliseconds(0); timeout = std::chrono::milliseconds(-1);
} }
} }
} }
...@@ -155,6 +155,10 @@ struct SyncImpl : public Reactor::Impl { ...@@ -155,6 +155,10 @@ struct SyncImpl : public Reactor::Impl {
shutdownFd.notify(); shutdownFd.notify();
} }
static constexpr size_t MaxHandlers() {
return HandlerList::MaxHandlers;
}
private: private:
Polling::Tag encodeTag(const Reactor::Key& key, Polling::Tag tag) const { Polling::Tag encodeTag(const Reactor::Key& key, Polling::Tag tag) const {
...@@ -192,6 +196,14 @@ private: ...@@ -192,6 +196,14 @@ private:
struct HandlerList { struct HandlerList {
// We are using the highest 8 bits of the fd to encode the index of the handler,
// which gives us a maximum of 2**8 - 1 handler, 255
static constexpr size_t HandlerBits = 8;
static constexpr size_t HandlerShift = sizeof(uint64_t) - HandlerBits;
static constexpr uint64_t DataMask = uint64_t(-1) >> HandlerBits;
static constexpr size_t MaxHandlers = (1 << HandlerBits) - 1;
HandlerList() { HandlerList() {
std::fill(std::begin(handlers), std::end(handlers), nullptr); std::fill(std::begin(handlers), std::end(handlers), nullptr);
} }
...@@ -267,15 +279,6 @@ private: ...@@ -267,15 +279,6 @@ private:
} }
private: private:
// We are using the highest 8 bits of the fd to encode the index of the handler,
// which gives us a maximum of 2**8 - 1 handler, 255
static constexpr size_t HandlerBits = 8;
static constexpr size_t HandlerShift = sizeof(uint64_t) - HandlerBits;
static constexpr uint64_t DataMask = uint64_t(-1) >> HandlerBits;
static constexpr size_t MaxHandlers = (1 << HandlerBits) - 1;
std::array<std::shared_ptr<Handler>, MaxHandlers> handlers; std::array<std::shared_ptr<Handler>, MaxHandlers> handlers;
size_t index_; size_t index_;
}; };
...@@ -315,6 +318,9 @@ private: ...@@ -315,6 +318,9 @@ private:
*/ */
struct AsyncImpl : public Reactor::Impl { struct AsyncImpl : public Reactor::Impl {
static constexpr uint32_t KeyMarker = 0xBADB0B;
AsyncImpl(Reactor* reactor, size_t threads) AsyncImpl(Reactor* reactor, size_t threads)
: Reactor::Impl(reactor) { : Reactor::Impl(reactor) {
...@@ -327,22 +333,32 @@ struct AsyncImpl : public Reactor::Impl { ...@@ -327,22 +333,32 @@ struct AsyncImpl : public Reactor::Impl {
Reactor::Key addHandler( Reactor::Key addHandler(
const std::shared_ptr<Handler>& handler, bool) { const std::shared_ptr<Handler>& handler, bool) {
Reactor::Key key; Reactor::Key keys[SyncImpl::MaxHandlers()];
for (size_t i = 0; i < workers_.size(); ++i) { for (size_t i = 0; i < workers_.size(); ++i) {
auto &wrk = workers_[i]; auto &wrk = workers_[i];
auto cl = handler->clone(); auto cl = handler->clone();
key = wrk->sync->addHandler(cl, false /* setKey */); auto key = wrk->sync->addHandler(cl, false /* setKey */);
auto newKey = encodeKey(key, i); auto newKey = encodeKey(key, i);
cl->key_ = newKey; cl->key_ = newKey;
keys[i] = key;
} }
return key; auto data = keys[0].data() << 32 | KeyMarker;
return Reactor::Key(data);
} }
std::vector<std::shared_ptr<Handler>> handlers(const Reactor::Key& key) const { std::vector<std::shared_ptr<Handler>> handlers(const Reactor::Key& key) const {
auto idx = decodeKey(key).first;
uint32_t idx;
uint32_t marker;
std::tie(idx, marker) = decodeKey(key);
if (marker != KeyMarker)
throw std::runtime_error("Invalid key");
Reactor::Key originalKey(idx); Reactor::Key originalKey(idx);
std::vector<std::shared_ptr<Handler>> res; std::vector<std::shared_ptr<Handler>> res;
...@@ -416,7 +432,7 @@ private: ...@@ -416,7 +432,7 @@ private:
auto decoded = decodeKey(key); auto decoded = decodeKey(key);
auto& wrk = workers_[decoded.second]; auto& wrk = workers_[decoded.second];
Reactor::Key originalKey(decoded.second); Reactor::Key originalKey(decoded.first);
CALL_MEMBER_FN(wrk->sync.get(), func)(originalKey, std::forward<Args>(args)...); CALL_MEMBER_FN(wrk->sync.get(), func)(originalKey, std::forward<Args>(args)...);
} }
......
...@@ -243,7 +243,13 @@ Listener::shutdown() { ...@@ -243,7 +243,13 @@ Listener::shutdown() {
Async::Promise<Listener::Load> Async::Promise<Listener::Load>
Listener::requestLoad(const Listener::Load& old) { 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) { return Async::whenAll(std::begin(loads), std::end(loads)).then([=](const std::vector<rusage>& usages) {
...@@ -252,7 +258,7 @@ Listener::requestLoad(const Listener::Load& old) { ...@@ -252,7 +258,7 @@ Listener::requestLoad(const Listener::Load& old) {
if (old.raw.empty()) { if (old.raw.empty()) {
res.global = 0.0; 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 { } else {
auto totalElapsed = [](rusage usage) { auto totalElapsed = [](rusage usage) {
......
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