Commit 17325ef6 authored by Dominik Charousset's avatar Dominik Charousset

more consistent interface for MM event handler

parent a3dac573
......@@ -77,30 +77,27 @@ struct fd_meta_info {
: fd(a0), ptr(a1), mask(a2) { }
};
struct fd_meta_info_less {
inline bool operator()(const fd_meta_info& lhs, native_socket_type rhs) const {
return lhs.fd < rhs;
}
};
enum class fd_meta_event { add, erase, mod };
class middleman_event_handler {
public:
typedef std::pair<event_bitmask, continuable_io*> io_event;
virtual ~middleman_event_handler();
void alteration(const continuable_io_ptr& ptr, event_bitmask e, fd_meta_event etype);
void add(const continuable_io_ptr& ptr, event_bitmask e);
/**
* @brief Enqueues an add operation.
*/
void add_later(const continuable_io_ptr& ptr, event_bitmask e);
void erase(const continuable_io_ptr& ptr, event_bitmask e);
event_bitmask next_bitmask(event_bitmask old, event_bitmask arg, fd_meta_event op) const;
/**
* @brief Enqueues an erase operation.
*/
void erase_later(const continuable_io_ptr& ptr, event_bitmask e);
/**
* @brief Poll all events.
*/
template<typename F>
void poll(const F& fun) {
poll_impl();
......@@ -113,40 +110,41 @@ class middleman_event_handler {
virtual void init() = 0;
virtual size_t num_sockets() const = 0;
// fills the event vector
virtual void poll_impl() = 0;
virtual void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_io* ptr) = 0;
inline size_t num_sockets() const { return m_meta.size(); }
// implemented in platform-dependent .cpp file
static std::unique_ptr<middleman_event_handler> create();
/**
* @brief Performs all actions enqueued by {@link add_later}
* or {@link erase_later}.
*/
void update();
protected:
fd_meta_info_less m_less;
std::vector<fd_meta_info> m_meta; // this vector is *always* sorted
std::vector<std::pair<fd_meta_info, fd_meta_event> > m_alterations;
std::vector<std::pair<fd_meta_info, fd_meta_event>> m_alterations;
std::vector<io_event> m_events;
std::vector<std::pair<event_bitmask, continuable_io*>> m_events;
middleman_event_handler();
// fills the event vector
virtual void poll_impl() = 0;
virtual void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_io* ptr) = 0;
private:
std::vector<fd_meta_info>::iterator find_meta(native_socket_type fd) {
auto last = end(m_meta);
auto iter = lower_bound(begin(m_meta), last, fd, m_less);
return (iter != last && iter->fd == fd) ? iter : last;
}
void alteration(const continuable_io_ptr& ptr, event_bitmask e, fd_meta_event etype);
event_bitmask next_bitmask(event_bitmask old, event_bitmask arg, fd_meta_event op) const;
};
......
......@@ -221,23 +221,23 @@ middleman_event_handler& abstract_middleman::handler() {
void abstract_middleman::continue_writer(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
handler().add(ptr, event::write);
handler().add_later(ptr, event::write);
}
void abstract_middleman::stop_writer(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
handler().erase(ptr, event::write);
handler().erase_later(ptr, event::write);
}
void abstract_middleman::continue_reader(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_readers.push_back(ptr);
handler().add(ptr, event::read);
handler().add_later(ptr, event::read);
}
void abstract_middleman::stop_reader(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
handler().erase(ptr, event::read);
handler().erase_later(ptr, event::read);
auto last = end(m_readers);
auto i = find_if(begin(m_readers), last, [ptr](const continuable_io_ptr& value) {
......@@ -305,7 +305,7 @@ void middleman_loop(default_middleman_impl* impl) {
}
CPPA_LOGF_DEBUG("event loop done, erase all readers");
// make sure to write everything before shutting down
for (auto ptr : impl->m_readers) { handler->erase(ptr, event::read); }
for (auto ptr : impl->m_readers) { handler->erase_later(ptr, event::read); }
handler->update();
CPPA_LOGF_DEBUG("flush outgoing messages");
CPPA_LOGF_DEBUG_IF(handler->num_sockets() == 0,
......@@ -318,19 +318,19 @@ void middleman_loop(default_middleman_impl* impl) {
case write_closed:
case write_failure:
case write_done:
handler->erase(io, event::write);
handler->erase_later(io, event::write);
break;
default: break;
}
break;
case event::error:
io->io_failed();
handler->erase(io, event::both);
handler->erase_later(io, event::both);
break;
default:
CPPA_LOGF_ERROR("expected event::write only "
"during shutdown phase");
handler->erase(io, event::read);
handler->erase_later(io, event::read);
break;
}
});
......
......@@ -67,12 +67,12 @@ void middleman_event_handler::alteration(const continuable_io_ptr& ptr,
m_alterations.emplace_back(fd_meta_info(fd, ptr, e), etype);
}
void middleman_event_handler::add(const continuable_io_ptr& ptr, event_bitmask e) {
void middleman_event_handler::add_later(const continuable_io_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::add);
}
void middleman_event_handler::erase(const continuable_io_ptr& ptr, event_bitmask e) {
void middleman_event_handler::erase_later(const continuable_io_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::erase);
}
......@@ -84,11 +84,14 @@ event_bitmask middleman_event_handler::next_bitmask(event_bitmask old, event_bit
void middleman_event_handler::update() {
CPPA_LOG_TRACE("");
auto mless = [](const fd_meta_info& lhs, native_socket_type rhs) {
return lhs.fd < rhs;
};
for (auto& elem_pair : m_alterations) {
auto& elem = elem_pair.first;
auto old = event::none;
auto last = end(m_meta);
auto iter = lower_bound(begin(m_meta), last, elem.fd, m_less);
auto last = m_meta.end();
auto iter = std::lower_bound(m_meta.begin(), last, elem.fd, mless);
if (iter != last) old = iter->mask;
auto mask = next_bitmask(old, elem.mask, elem_pair.second);
auto ptr = elem.ptr.get();
......
......@@ -58,7 +58,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
m_epoll_events.resize(64);
}
size_t num_sockets() const { return m_meta.size(); }
protected:
void poll_impl() {
CPPA_REQUIRE(m_meta.empty() == false);
......@@ -100,10 +100,9 @@ class middleman_event_handler_impl : public middleman_event_handler {
void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask,
event_bitmask new_bitmask,
continuable_io* ptr) {
static_cast<void>(old_bitmask); // no need for it
int operation;
epoll_event ee;
ee.data.ptr = ptr;
......
......@@ -58,7 +58,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
void init() { }
size_t num_sockets() const { return m_pollset.size(); }
protected:
void poll_impl() {
CPPA_REQUIRE(m_pollset.empty() == false);
......@@ -108,23 +108,22 @@ class middleman_event_handler_impl : public middleman_event_handler {
void handle_event(fd_meta_event me,
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask,
event_bitmask new_bitmask,
continuable_io*) {
static_cast<void>(old_bitmask); // no need for it
auto last = m_pollset.end();
auto iter = std::lower_bound(m_pollset.begin(), last, fd, pollfd_less);
switch (me) {
case fd_meta_event::add: {
pollfd tmp;
tmp.fd = fd;
tmp.events = to_poll_bitmask(new_bitmask);
tmp.revents = 0;
m_pollset.insert(lb(fd), tmp);
m_pollset.insert(iter, tmp);
CPPA_LOGMF(CPPA_DEBUG, self, "inserted new element");
break;
}
case fd_meta_event::erase: {
auto last = end(m_pollset);
auto iter = lb(fd);
CPPA_LOG_ERROR_IF(iter == last || iter->fd != fd,
"m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)");
......@@ -135,8 +134,6 @@ class middleman_event_handler_impl : public middleman_event_handler {
break;
}
case fd_meta_event::mod: {
auto last = end(m_pollset);
auto iter = lb(fd);
CPPA_LOG_ERROR_IF(iter == last || iter->fd != fd,
"m_meta and m_pollset out of sync; "
"no element found for fd (cannot erase)");
......@@ -151,10 +148,6 @@ class middleman_event_handler_impl : public middleman_event_handler {
private:
std::vector<pollfd>::iterator lb(native_socket_type fd) {
return lower_bound(begin(m_pollset), end(m_pollset), fd, pollfd_less);
}
std::vector<pollfd> m_pollset; // always in sync with m_meta
};
......
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