added net module

source/core/net/acceptor.cpp

@@ -0,0 +1,119 @@
+#include <utility>
+#include "matador/net/acceptor.hpp"
+#include "matador/net/reactor.hpp"
+
+#include "matador/logger/log_manager.hpp"
+
+namespace matador {
+
+acceptor::acceptor()
+  : log_(matador::create_logger("Acceptor"))
+{}
+
+acceptor::acceptor(tcp::peer endpoint)
+: endpoint_(std::move(endpoint))
+, log_(matador::create_logger("Acceptor"))
+{}
+
+acceptor::acceptor(tcp::peer  endpoint, t_accept_handler make_handler)
+  : endpoint_(std::move(endpoint))
+  , accept_handler_(std::move(make_handler))
+  , log_(matador::create_logger("Acceptor"))
+{}
+
+acceptor::~acceptor()
+{
+  acceptor_.close();
+}
+
+void acceptor::accecpt(acceptor::t_accept_handler on_new_connection)
+{
+  accept_handler_ = std::move(on_new_connection);
+}
+
+void acceptor::accecpt(const tcp::peer &endpoint, acceptor::t_accept_handler on_new_connection)
+{
+  endpoint_ = endpoint;
+  accecpt(std::move(on_new_connection));
+}
+
+void acceptor::open()
+{
+  acceptor_.bind(endpoint_);
+  acceptor_.listen(10);
+  name_ += " (fd: " + std::to_string(acceptor_.id()) + ")";
+  log_.debug("fd %d: accepting connections", handle());
+}
+
+socket_type acceptor::handle() const
+{
+  return acceptor_.id();
+}
+
+void acceptor::on_input()
+{
+  tcp::socket sock;
+
+  tcp::peer endpoint = create_client_endpoint();
+  log_.debug("fd %d: accepting connection ...", handle());
+  int ret = acceptor_.accept(sock, endpoint);
+
+  if (ret < 0) {
+    char error_buffer[1024];
+    os::strerror(errno, error_buffer, 1024);
+    log_.error("accept failed: %s", error_buffer);
+  } else {
+    // create new client handler
+    log_.debug("accepted connection from %s", endpoint.to_string().c_str());
+
+    auto h = accept_handler_(sock, endpoint, this);
+
+    get_reactor()->register_handler(h, event_type::READ_WRITE_MASK);
+
+    log_.debug("fd %d: accepted socket id %d", handle(), sock.id());
+  }
+}
+
+void acceptor::close()
+{
+  log_.debug("closing acceptor %d", acceptor_.id());
+  acceptor_.close();
+  // Todo: unregister from reactor (maybe observer pattern?)
+  // notify()
+}
+
+bool acceptor::is_ready_write() const
+{
+  return false;
+}
+
+bool acceptor::is_ready_read() const
+{
+  return handle() > 0;
+}
+
+const tcp::peer &acceptor::endpoint() const
+{
+  return endpoint_;
+}
+
+tcp::peer acceptor::create_client_endpoint() const
+{
+  if (endpoint_.addr().is_v4()) {
+    return matador::tcp::peer(address::v4::empty());
+  } else {
+    return matador::tcp::peer(address::v6::empty());
+  }
+}
+
+void acceptor::notify_close(handler *)
+{
+
+}
+
+std::string acceptor::name() const
+{
+  return name_;
+}
+
+}

source/core/net/address.cpp

@@ -0,0 +1,154 @@
+#include "matador/net/address.hpp"
+
+#ifdef _WIN32
+#else
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#endif
+
+#include <stdexcept>
+
+namespace matador {
+
+const char* address_router<V6>::IP6ADDR_MULTICAST_ALLNODES = "FF02::1";
+
+address::address(const sockaddr_in &addr)
+  : size_(sizeof(sockaddr_in))
+{
+  socket_address_.sa_in = addr;
+}
+
+address::address(const sockaddr_in6 &addr)
+  : size_(sizeof(sockaddr_in6))
+{
+  socket_address_.sa_in6 = addr;
+}
+
+address& address::operator=(const address &x)
+{
+  if (this == &x) {
+    return *this;
+  }
+  clear();
+  size_ = x.size_;
+  socket_address_ = x.socket_address_;
+  return *this;
+}
+
+address::address(address &&x) noexcept
+  : socket_address_(x.socket_address_)
+  , size_(x.size_)
+{
+  x.size_ = 0;
+}
+
+address& address::operator=(address &&x) noexcept
+{
+  if (this == &x) {
+    return *this;
+  }
+  clear();
+  socket_address_ = x.socket_address_;
+  size_ = x.size_;
+  x.size_ = 0;
+  return *this;
+}
+
+address::~address()
+{
+  clear();
+}
+
+unsigned int address::to_ulong() const
+{
+  if (socket_address_.sa_raw.sa_family == PF_INET) {
+    return socket_address_.sa_in.sin_addr.s_addr;
+  } else {
+    return 0;
+    //return reinterpret_cast<sockaddr_in6 *>(addr_)->sin6_addr;
+  }
+}
+
+std::string address::to_string() const
+{
+  char addstr[INET6_ADDRSTRLEN];
+  if (is_v4()) {
+    os::inet_ntop(socket_address_.sa_raw.sa_family, &socket_address_.sa_in.sin_addr, addstr, INET6_ADDRSTRLEN);
+  } else {
+    os::inet_ntop(socket_address_.sa_raw.sa_family, &socket_address_.sa_in6.sin6_addr, addstr, INET6_ADDRSTRLEN);
+  }
+  return std::string(addstr);
+}
+
+void address::port(unsigned short pn)
+{
+  if (is_v4()) {
+    socket_address_.sa_in.sin_port = htons(pn);
+  } else {
+    socket_address_.sa_in6.sin6_port = htons(pn);
+  }
+}
+
+unsigned short address::port() const
+{
+  if (is_v4()) {
+    return ntohs(socket_address_.sa_in.sin_port);
+  } else {
+    return ntohs(socket_address_.sa_in6.sin6_port);
+  }
+}
+
+bool address::is_v4() const
+{
+  return socket_address_.sa_raw.sa_family == PF_INET;
+}
+
+bool address::is_v6() const
+{
+  return socket_address_.sa_raw.sa_family == PF_INET6;
+}
+
+sockaddr *address::addr()
+{
+  return &socket_address_.sa_raw;
+}
+
+const sockaddr *address::addr() const
+{
+  return &socket_address_.sa_raw;
+}
+
+sockaddr_in *address::addr_v4()
+{
+  return &socket_address_.sa_in;
+}
+
+const sockaddr_in *address::addr_v4() const
+{
+  return &socket_address_.sa_in;
+}
+
+sockaddr_in6 *address::addr_v6()
+{
+  return &socket_address_.sa_in6;
+}
+
+const sockaddr_in6 *address::addr_v6() const
+{
+  return &socket_address_.sa_in6;
+}
+
+socklen_t address::size() const
+{
+  return size_;
+}
+
+void address::clear()
+{
+  if (is_v4()) {
+    memset(&socket_address_.sa_in, 0, sizeof(socket_address_.sa_in));
+  } else {
+    memset(&socket_address_.sa_in6, 0, sizeof(socket_address_.sa_in6));
+  }
+}
+}

source/core/net/address_resolver.cpp

@@ -0,0 +1,19 @@
+#include "matador/net/address_resolver.hpp"
+
+namespace matador {
+namespace detail {
+
+template<>
+int determine_socktype<tcp>()
+{
+  return SOCK_STREAM;
+}
+
+template<>
+int determine_socktype<udp>()
+{
+  return SOCK_DGRAM;
+}
+
+}
+}

source/core/net/connector.cpp

@@ -0,0 +1,84 @@
+#include "matador/logger/log_manager.hpp"
+
+#include "matador/net/connector.hpp"
+#include "matador/net/reactor.hpp"
+
+#include "matador/utils/os.hpp"
+
+#include <utility>
+
+namespace matador {
+
+connector::connector()
+  : log_(matador::create_logger("Connector"))
+{}
+
+connector::connector(t_connect_handler on_new_connection)
+  : connect_handler_(std::move(on_new_connection))
+  , log_(matador::create_logger("Connector"))
+{}
+
+void connector::connect(reactor &r, const std::vector<tcp::peer> &endpoints)
+{
+  endpoints_ = endpoints;
+  r.schedule_timer(shared_from_this(), 0, 3);
+}
+
+void connector::connect(reactor &r, const std::vector<tcp::peer> &endpoints, t_connect_handler on_new_connection)
+{
+  connect_handler_ = std::move(on_new_connection);
+  connect(r, endpoints);
+}
+
+socket_type connector::handle() const
+{
+  return 0;
+}
+
+void connector::on_timeout()
+{
+  tcp::socket stream;
+  for (const auto &ep : endpoints_) {
+    if (!ep.addr().is_v4()) {
+      continue;
+    }
+    auto ret = matador::connect(stream, ep);
+    if (ret != 0) {
+      char error_buffer[1024];
+      log_.error("couldn't establish connection to: %s", ep.to_string().c_str(), os::strerror(errno, error_buffer, 1024));
+      continue;
+    } else {
+      log_.info("connection established to %s (fd: %d)", ep.to_string().c_str(), stream.id());
+    }
+    stream.non_blocking(true);
+    auto h = connect_handler_(stream, ep, this);
+
+    get_reactor()->register_handler(h, event_type::READ_WRITE_MASK);
+    get_reactor()->cancel_timer(shared_from_this());
+
+    break;
+  }
+  endpoints_.clear();
+}
+
+bool connector::is_ready_write() const
+{
+  return false;
+}
+
+bool connector::is_ready_read() const
+{
+  return false;
+}
+
+void connector::notify_close(handler *)
+{
+
+}
+
+std::string connector::name() const
+{
+  return "connector";
+}
+
+}

source/core/net/error.cpp

@@ -0,0 +1,38 @@
+#include "matador/net/error.hpp"
+#include "matador/utils/os.hpp"
+
+#include <stdexcept>
+
+#ifdef _WIN32
+#include <WS2tcpip.h>
+#else
+#include <netdb.h>
+#endif
+
+namespace matador {
+namespace detail {
+
+void throw_logic_error(const char* msg)
+{
+  throw std::logic_error(msg);
+}
+
+void throw_logic_error_with_errno(const char* msg, int err)
+{
+  char error_buffer[1024];
+  os::strerror(err, error_buffer, 1024);
+
+  char message_buffer[1024];
+  os::sprintf(message_buffer, 1024, msg, error_buffer);
+  throw std::logic_error(message_buffer);
+}
+
+void throw_logic_error_with_gai_errno(const char* msg, int err)
+{
+  char message_buffer[1024];
+  os::sprintf(message_buffer, 1024, msg, gai_strerror(err));
+  throw std::logic_error(message_buffer);
+}
+
+}
+}

source/core/net/fdset.cpp

@@ -0,0 +1,66 @@
+#include "matador/net/fdset.hpp"
+
+namespace matador {
+
+fdset::fdset()
+{
+  reset();
+}
+
+fdset::fdset(fdset&& x) noexcept
+: max_fd_set_(std::move(x.max_fd_set_))
+, fd_set_(x.fd_set_) {}
+
+fdset& fdset::operator=(fdset&& x) noexcept
+{
+  max_fd_set_ = std::move(x.max_fd_set_);
+  fd_set_ = x.fd_set_;
+  return *this;
+}
+
+// set all bits to zero
+void fdset::reset()
+{
+  FD_ZERO(&fd_set_);
+  max_fd_set_.clear();
+  max_fd_set_.insert(0);
+}
+
+bool fdset::is_set(socket_type fd) const
+{
+  return FD_ISSET(fd, &fd_set_) > 0;
+}
+
+void fdset::clear(socket_type fd)
+{
+  FD_CLR(fd, &fd_set_);
+  max_fd_set_.erase(fd);
+}
+
+void fdset::set(socket_type fd)
+{
+  FD_SET(fd, &fd_set_);
+  max_fd_set_.insert(fd);
+}
+
+socket_type fdset::maxp1() const
+{
+  return *max_fd_set_.begin();
+}
+
+size_t fdset::count() const
+{
+  return max_fd_set_.size() - 1;
+}
+
+bool fdset::empty() const
+{
+  return count() == 0;
+}
+
+fd_set* fdset::get()
+{
+  return &fd_set_;
+}
+
+}

source/core/net/handler.cpp

@@ -0,0 +1,48 @@
+#include "matador/net/handler.hpp"
+
+#include <ctime>
+
+namespace matador {
+
+time_t handler::next_timeout() const
+{
+  return next_timeout_;
+}
+
+time_t handler::interval() const
+{
+  return interval_;
+}
+
+reactor *handler::get_reactor() const
+{
+  return reactor_;
+}
+
+void handler::register_reactor(reactor *r)
+{
+  reactor_ = r;
+}
+
+void handler::schedule(time_t offset, time_t interval)
+{
+  next_timeout_ = ::time(nullptr) + offset;
+  interval_ = interval;
+}
+
+void handler::cancel_timer()
+{
+  next_timeout_ = 0;
+  interval_ = 0;
+}
+
+void handler::calculate_next_timeout(time_t now)
+{
+  if (interval_ > 0) {
+    next_timeout_ = now + interval_;
+  } else {
+    next_timeout_ = 0;
+  }
+}
+
+}

source/core/net/io_service.cpp

@@ -0,0 +1,31 @@
+#include "matador/net/io_service.hpp"
+
+#include "matador/logger/log_manager.hpp"
+
+namespace matador {
+
+io_service::io_service()
+  : log_(matador::create_logger("IOService"))
+{}
+
+io_service::~io_service()
+{
+  reactor_.shutdown();
+}
+
+void io_service::run()
+{
+  reactor_.run();
+}
+
+bool io_service::is_running() const
+{
+  return reactor_.is_running();
+}
+
+void io_service::shutdown()
+{
+  reactor_.shutdown();
+}
+
+}

source/core/net/leader_follower_thread_pool.cpp

@@ -0,0 +1,106 @@
+#include "matador/net/leader_follower_thread_pool.hpp"
+
+#include "matador/utils/thread_helper.hpp"
+
+#include <algorithm>
+
+namespace matador {
+
+leader_follower_thread_pool::~leader_follower_thread_pool()
+{
+  shutdown();
+}
+
+void leader_follower_thread_pool::start()
+{
+  is_running_ = true;
+  for (std::size_t i = 0; i < num_threads_; ++i) {
+    threads_.emplace_back([this] { execute(); });
+  }
+  log_.info("thread pool started with %d threads", num_threads_);
+}
+
+void leader_follower_thread_pool::stop()
+{
+  is_running_ = false;
+}
+
+void leader_follower_thread_pool::promote_new_leader()
+{
+  std::lock_guard<std::mutex> l(mutex_);
+
+  if (leader_ != std::this_thread::get_id()) {
+    return;
+  }
+
+  leader_ = null_id;
+
+  signal_ready_ = true;
+  condition_synchronizer_.notify_one();
+}
+
+std::size_t leader_follower_thread_pool::size() const
+{
+  return threads_.size();
+}
+
+void leader_follower_thread_pool::shutdown()
+{
+  {
+    const std::lock_guard<std::mutex> l(mutex_);
+//    if (!is_running_) {
+//      return;
+//    }
+    stop();
+    log_.info("shutting down; notifying all tasks");
+    signal_shutdown_ = true;
+    condition_synchronizer_.notify_all();
+  }
+
+  std::for_each(threads_.begin(), threads_.end(), [](thread_vector_t::reference item) {
+    if (item.joinable()) {
+      item.join();
+    }
+  });
+}
+
+std::thread::id leader_follower_thread_pool::leader() {
+  const std::lock_guard<std::mutex> l(mutex_);
+  return leader_;
+}
+
+std::size_t leader_follower_thread_pool::num_follower() const {
+  return follower_;
+}
+
+bool leader_follower_thread_pool::is_running() const
+{
+  return is_running_;
+}
+
+void leader_follower_thread_pool::execute() {
+  std::unique_lock<std::mutex> l(mutex_);
+  while (is_running_) {
+    while (leader_ != null_id) {
+//      log_.info("thread <%d> waiting for synchronizer (leader %d)",
+//                acquire_thread_index(std::this_thread::get_id()),
+//                acquire_thread_index(leader_));
+      condition_synchronizer_.wait(l, [this]() { return signal_ready_ || signal_shutdown_; });
+      signal_ready_ = false;
+      if (!is_running_) {
+        return;
+      }
+    }
+
+//    log_.info("new leader <%d> (thread <%d> is now follower)",acquire_thread_index(std::this_thread::get_id()) , acquire_thread_index(leader_));
+    leader_ = std::this_thread::get_id();
+
+    l.unlock();
+
+    join_();
+
+//    log_.info("thread <%d> finished work", acquire_thread_index(std::this_thread::get_id()));
+    l.lock();
+  }
+}
+}

source/core/net/os.cpp

@@ -0,0 +1,84 @@
+#include "matador/net/os.hpp"
+
+#ifdef _WIN32
+#include <Ws2tcpip.h>
+#else
+#include <arpa/inet.h>
+#include <unistd.h>
+#endif
+
+namespace matador {
+namespace net {
+
+void init()
+{
+#ifdef _WIN32
+  WSADATA wsaData;   // if this doesn't work
+  //WSAData wsaData; // then try this instead
+
+  // MAKEWORD(1,1) for Winsock 1.1, MAKEWORD(2,0) for Winsock 2.0:
+
+  if (WSAStartup(MAKEWORD(1,1), &wsaData) != 0) {
+      fprintf(stderr, "WSAStartup failed.\n");
+      exit(1);
+  }
+#endif
+}
+
+void cleanup()
+{
+#ifdef _WIN32
+  WSACleanup();
+#endif
+}
+
+}
+
+bool is_valid_socket(socket_type fd) {
+#ifdef _WIN32
+  return fd != INVALID_SOCKET;
+#else
+  return fd >= 0;
+#endif
+}
+
+namespace os {
+
+int inet_pton(int af, const char *src, void *dst)
+{
+#ifdef _WIN32
+  return ::InetPton(af, src, dst);
+#else
+  return ::inet_pton(af, src, dst);
+#endif
+}
+
+const char* inet_ntop(int af, const void* src, char* dst, size_t size)
+{
+#ifdef _WIN32
+    return ::InetNtop(af, const_cast<void*>(src), dst, size);
+#else
+    return ::inet_ntop(af, src, dst, size);
+#endif
+}
+
+int close(socket_type fd)
+{
+#ifdef _WIN32
+  return ::closesocket(fd);
+#else
+  return ::close(fd);
+#endif
+}
+
+int shutdown(socket_type fd, shutdown_type type)
+{
+#ifdef _WIN32
+    return ::shutdown(fd, static_cast<int>(type));
+#else
+    return ::shutdown(fd, static_cast<int>(type));
+#endif
+}
+
+}
+}

source/core/net/reactor.cpp

@@ -0,0 +1,411 @@
+#include "matador/net/reactor.hpp"
+#include "matador/net/handler.hpp"
+
+#include "matador/logger/log_manager.hpp"
+
+#include <algorithm>
+#include <limits>
+#include <cerrno>
+#include <ctime>
+#include <iostream>
+
+namespace matador {
+
+reactor::reactor()
+  : sentinel_(std::shared_ptr<handler>(nullptr))
+  , log_(create_logger("Reactor"))
+  , thread_pool_(4, [this]() { handle_events(); })
+{
+}
+reactor::~reactor()
+{
+    log_.debug("destroying reactor");
+    thread_pool_.shutdown();
+}
+
+void reactor::register_handler(const handler_ptr& h, event_type et)
+{
+  h->register_reactor(this);
+  h->open();
+
+  std::lock_guard<std::mutex> l(mutex_);
+  auto it = find_handler_type(h);
+
+  if (it == handlers_.end()) {
+    handlers_.emplace_back(h, et);
+  } else if (it->first != h) {
+    throw std::logic_error("given handler isn't expected handler");
+  } else {
+    it->second = it->second | et;
+  }
+  interrupt_without_lock();
+}
+
+void reactor::unregister_handler(const handler_ptr& h, event_type)
+{
+  std::lock_guard<std::mutex> l(mutex_);
+  auto it = find_handler_type(h);
+
+  if (it != handlers_.end()) {
+    (*it).first->close();
+    handlers_.erase(it);
+  }
+  interrupt_without_lock();
+}
+
+void reactor::schedule_timer(const std::shared_ptr<handler>& h, time_t offset, time_t interval)
+{
+  h->register_reactor(this);
+
+  std::lock_guard<std::mutex> l(mutex_);
+  auto it = find_handler_type(h);
+
+  if (it == handlers_.end()) {
+    handlers_.emplace_back(h, event_type::TIMEOUT_MASK);
+  }
+
+  h->schedule(offset, interval);
+}
+
+void reactor::cancel_timer(const std::shared_ptr<handler>& h)
+{
+  std::lock_guard<std::mutex> l(mutex_);
+  auto it = find_handler_type(h);
+
+  if (it != handlers_.end()) {
+    handlers_.erase(it);
+  }
+
+  h->cancel_timer();
+}
+
+void reactor::run()
+{
+//  log_.info("start dispatching all clients");
+  thread_pool_.start();
+
+  {
+//    log_.info("waiting for reactor shutdown");
+    std::unique_lock<std::mutex> l(mutex_);
+    shutdown_.wait(l, [this]() { return shutdown_requested_.load(); });
+    cleanup();
+  }
+//  log_.info("all clients dispatched; shutting down");
+  thread_pool_.stop();
+}
+
+void reactor::handle_events()
+{
+//  std::cout << this << " start handle events\n" << std::flush;
+
+//  log_.info("handle events");
+
+  running_ = true;
+  time_t timeout;
+  select_fdsets fd_sets;
+  prepare_select_bits(timeout, fd_sets);
+
+//  std::cout << this << " fd sets r: " << fd_sets.read_set().count() << ", w: " << fd_sets.write_set().count() << ", e: " << fd_sets.except_set().count() << ", max: " << fd_sets.maxp1() << "\n" << std::flush;
+
+//  log_.debug("fds [r: %d, w: %d, e: %d]",
+//             fdsets_.read_set().count(),
+//             fdsets_.write_set().count(),
+//             fdsets_.except_set().count());
+
+//  if (timeout != (std::numeric_limits<time_t>::max)()) {
+//    log_.debug("next timeout in %d sec", timeout);
+//  }
+  struct timeval tselect{};
+  struct timeval* p = nullptr;
+  if (timeout < (std::numeric_limits<time_t>::max)()) {
+    tselect.tv_sec = timeout;
+    tselect.tv_usec = 0;
+    p = &tselect;
+//    std::cout << this << " next timeout in " << p->tv_sec << " seconds\n" << std::flush;
+  }
+
+  if (!has_clients_to_handle(timeout, fd_sets)) {
+//    std::cout << this << " no clients to handle; returning\n" << std::flush;
+//    log_.info("no clients to handle, exiting");
+    return;
+  }
+
+  int ret;
+  while ((ret = select(p, fd_sets)) < 0) {
+//    std::cout << this << " select returned with error " << ret << "\n" << std::flush;
+    if(errno != EINTR) {
+      char error_buffer[1024];
+      log_.warn("select failed: %s", os::strerror(errno, error_buffer, 1024));
+      shutdown();
+    } else {
+      return;
+    }
+  }
+
+//  std::cout << this << " select returned with active requests " << ret << "\n" << std::flush;
+
+  bool interrupted = is_interrupted(fd_sets);
+
+  if (interrupted) {
+    if (!shutdown_requested_) {
+//      std::cout << this << " reactor was interrupted - promote new leader\n" << std::flush;
+//      log_.info("reactor was interrupted");
+      thread_pool_.promote_new_leader();
+      return;
+    } else {
+//      std::cout << this << " reactor was interrupted for shutdown\n" << std::flush;
+//      log_.info("shutting down");
+//      cleanup();
+      shutdown_.notify_one();
+      return;
+    }
+  }
+
+  time_t now = ::time(nullptr);
+  t_handler_type handler_type = resolve_next_handler(now, fd_sets);
+
+  if (handler_type.first) {
+    deactivate_handler(handler_type.first, handler_type.second);
+//    std::cout << this << " handling client " << handler_type.first->name() << " - promoting new leader\n" << std::flush;
+
+    thread_pool_.promote_new_leader();
+//    log_.info("start handling event");
+    // handle event
+    if (handler_type.second == event_type::WRITE_MASK) {
+      on_write_mask(handler_type.first);
+    } else if (handler_type.second == event_type::READ_MASK) {
+      on_read_mask(handler_type.first);
+    } else if (handler_type.second == event_type::TIMEOUT_MASK) {
+      on_timeout(handler_type.first, now);
+    } else {
+//      log_.info("unknown event type");
+    }
+    activate_handler(handler_type.first, handler_type.second);
+    remove_deleted();
+  } else {
+    // no handler found
+//    log_.info("no handler found");
+    thread_pool_.promote_new_leader();
+  }
+}
+
+void reactor::shutdown()
+{
+  if (!is_running()) {
+    return;
+  }
+  // shutdown the reactor properly
+  log_.info("shutting down reactor");
+  shutdown_requested_ = true;
+  interrupt();
+}
+
+bool reactor::is_running() const
+{
+  return running_;
+}
+
+void reactor::prepare_select_bits(time_t& timeout, select_fdsets& fd_sets) const
+{
+  std::lock_guard<std::mutex> l(mutex_);
+  fd_sets.reset();
+  time_t now = ::time(nullptr);
+  timeout = (std::numeric_limits<time_t>::max)();
+
+  // set interrupter fd
+  fd_sets.read_set().set(interrupter_.socket_id());
+
+  for (const auto &h : handlers_) {
+    if (h.first == nullptr) {
+      continue;
+    }
+    if (h.first->is_ready_read() && is_event_type_set(h.second, event_type::READ_MASK)) {
+      fd_sets.read_set().set(h.first->handle());
+    }
+    if (h.first->is_ready_write() && is_event_type_set(h.second, event_type::WRITE_MASK)) {
+      fd_sets.write_set().set(h.first->handle());
+    }
+    if (h.first->next_timeout() > 0 && is_event_type_set(h.second, event_type::TIMEOUT_MASK)) {
+      timeout = (std::min)(timeout, h.first->next_timeout() <= now ? 0 : (h.first->next_timeout() - now));
+    }
+  }
+}
+
+void reactor::remove_deleted()
+{
+  while (!handlers_to_delete_.empty()) {
+    auto h = handlers_to_delete_.front();
+    handlers_to_delete_.pop_front();
+    auto fi = std::find_if(handlers_.begin(), handlers_.end(), [&h](const t_handler_type &ht) {
+      return ht.first.get() == h.get();
+    });
+
+    if (fi != handlers_.end()) {
+      log_.debug("removing handler %d", fi->first->handle());
+      handlers_.erase(fi);
+    }
+  }
+}
+
+void reactor::cleanup()
+{
+  while (!handlers_.empty()) {
+    auto hndlr = handlers_.front();
+    handlers_.pop_front();
+    hndlr.first->close();
+  }
+}
+
+int reactor::select(struct timeval *timeout, select_fdsets& fd_sets)
+{
+  log_.debug("calling select; waiting for io events");
+  return ::select(
+    static_cast<int>(fd_sets.maxp1()) + 1,
+    fd_sets.read_set().get(),
+    fd_sets.write_set().get(),
+    fd_sets.except_set().get(),
+    timeout
+  );
+}
+
+
+//void reactor::process_handler(int /*ret*/)
+//{
+//  handlers_.emplace_back(sentinel_, event_type::NONE_MASK);
+//  time_t now = ::time(nullptr);
+//  while (handlers_.front().first != nullptr) {
+//    auto h = handlers_.front();
+//    handlers_.pop_front();
+//    handlers_.push_back(h);
+//    // check for read/accept
+//    if (h.first->handle() > 0 && fdsets_.write_set().is_set(h.first->handle())) {
+//      on_write_mask(h.first);
+//    }
+//    if (h.first->handle() > 0 && fdsets_.read_set().is_set(h.first->handle())) {
+//      on_read_mask(h.first);
+//    }
+//    if (h.first->next_timeout() > 0 && h.first->next_timeout() <= now) {
+//      on_timeout(h.first, now);
+//    }
+//  }
+//  handlers_.pop_front();
+//}
+
+reactor::t_handler_type reactor::resolve_next_handler(time_t now, select_fdsets& fd_sets)
+{
+  std::lock_guard<std::mutex> l(mutex_);
+  for (auto &h : handlers_) {
+    if (h.first->handle() > 0 && fd_sets.write_set().is_set(h.first->handle())) {
+      return std::make_pair(h.first, event_type::WRITE_MASK);
+    }
+    if (h.first->handle() > 0 && fd_sets.read_set().is_set(h.first->handle())) {
+      return std::make_pair(h.first, event_type::READ_MASK);
+    }
+    if (h.first->next_timeout() > 0 && h.first->next_timeout() <= now) {
+      return std::make_pair(h.first, event_type::TIMEOUT_MASK);
+    }
+  }
+  return std::make_pair(nullptr, event_type::NONE_MASK);
+}
+
+void reactor::on_read_mask(const handler_ptr& handler)
+{
+//  log_.debug("read bit for handler %d is set; handle input", h->handle());
+//  std::cout << this << " (handler " << h.get() << "): handle read\n" << std::flush;
+  handler->on_input();
+}
+
+void reactor::on_write_mask(const handler_ptr& handler)
+{
+//  log_.debug("write bit for handler %d is set; handle output", h->handle());
+//  std::cout << this << " (handler " << h.get() << "): handle write\n" << std::flush;
+  handler->on_output();
+}
+
+void reactor::on_except_mask(const handler_ptr& /*handler*/)
+{
+//  std::cout << this << " (handler " << h.get() << "): handle exception\n" << std::flush;
+
+}
+
+void reactor::on_timeout(const handler_ptr &h, time_t now)
+{
+//  log_.debug("timeout expired for handler %d; handle timeout", h->handle());
+//  std::cout << this << " (handler " << h.get() << "): handle timeout\n" << std::flush;
+  h->calculate_next_timeout(now);
+  h->on_timeout();
+}
+
+select_fdsets reactor::fdsets() const
+{
+  time_t timeout;
+  select_fdsets fd_sets;
+  prepare_select_bits(timeout, fd_sets);
+  return fd_sets;
+}
+
+void reactor::mark_handler_for_delete(const handler_ptr& h)
+{
+  std::lock_guard<std::mutex> l(mutex_);
+  handlers_to_delete_.push_back(h);
+}
+
+bool reactor::is_interrupted(select_fdsets& fd_sets)
+{
+  std::lock_guard<std::mutex> l(mutex_);
+  if (fd_sets.read_set().is_set(interrupter_.socket_id())) {
+    log_.debug("interrupt byte received; resetting interrupter");
+    if (shutdown_requested_) {
+      running_ = false;
+    }
+    return interrupter_.reset();
+  }
+  return false;
+}
+
+bool reactor::has_clients_to_handle(time_t timeout, select_fdsets& fd_sets) const {
+  std::lock_guard<std::mutex> lock(mutex_);
+  return fd_sets.maxp1() > 0 || timeout != (std::numeric_limits<time_t>::max)();
+}
+
+std::list<reactor::t_handler_type>::iterator reactor::find_handler_type(const reactor::handler_ptr &h)
+{
+  return std::find_if(handlers_.begin(), handlers_.end(), [&h](const t_handler_type &ht) {
+    return ht.first.get() == h.get();
+  });
+}
+
+void reactor::activate_handler(const reactor::handler_ptr &h, event_type ev)
+{
+  std::lock_guard<std::mutex> l(mutex_);
+  auto it = find_handler_type(h);
+  if (it == handlers_.end()) {
+    return;
+  }
+  it->second |= ev;
+}
+
+void reactor::deactivate_handler(const reactor::handler_ptr &h, event_type ev)
+{
+  std::lock_guard<std::mutex> l(mutex_);
+  auto it = find_handler_type(h);
+  if (it == handlers_.end()) {
+    return;
+  }
+  it->second &= ~ev;
+}
+
+void reactor::interrupt()
+{
+  log_.trace("interrupting reactor");
+  std::lock_guard<std::mutex> l(mutex_);
+  interrupter_.interrupt();
+}
+
+void reactor::interrupt_without_lock()
+{
+  log_.trace("interrupting reactor");
+  interrupter_.interrupt();
+}
+
+}

source/core/net/select_fdsets.cpp

@@ -0,0 +1,74 @@
+#include "matador/net/select_fdsets.hpp"
+
+#include <algorithm>
+
+namespace matador {
+
+socket_type select_fdsets::maxp1() const
+{
+  return (std::max)(fdsets_[0].maxp1(), (std::max)(fdsets_[1].maxp1(), fdsets_[2].maxp1()));
+}
+
+fdset& select_fdsets::fd_set(fdset_type type)
+{
+  return fdsets_[type];
+}
+
+fdset& select_fdsets::read_set()
+{
+  return fdsets_[0];
+}
+
+const fdset& select_fdsets::read_set() const
+{
+  return fdsets_[0];
+}
+
+fdset& select_fdsets::write_set()
+{
+  return fdsets_[1];
+}
+
+const fdset& select_fdsets::write_set() const
+{
+  return fdsets_[1];
+}
+
+fdset& select_fdsets::except_set()
+{
+  return fdsets_[2];
+}
+
+const fdset& select_fdsets::except_set() const
+{
+  return fdsets_[2];
+}
+
+void select_fdsets::reset()
+{
+  for (auto &fdset : fdsets_) {
+    fdset.reset();
+  }
+}
+
+void select_fdsets::reset(fdset_type type)
+{
+  fdsets_[type].reset();
+}
+
+bool select_fdsets::is_set(fdset_type type, int fd) const
+{
+  return fdsets_[type].is_set(fd);
+}
+
+void select_fdsets::clear(fdset_type type, int fd)
+{
+  fdsets_[type].clear(fd);
+}
+
+void select_fdsets::set(fdset_type type, int fd)
+{
+  fdsets_[type].set(fd);
+}
+
+}

source/core/net/socket_interrupter.cpp

@@ -0,0 +1,74 @@
+#include "matador/net/socket_interrupter.hpp"
+
+#include "matador/utils/buffer_view.hpp"
+
+#include "matador/logger/log_manager.hpp"
+
+#ifndef _WIN32
+#include <netinet/tcp.h>
+#endif
+
+namespace matador {
+
+socket_interrupter::socket_interrupter()
+  : client_(tcp::v4())
+  , log_(create_logger("SocketInterrupter"))
+{
+  /*
+   * setup acceptor
+   * - create socket
+   * - set reuse address option
+   * - bind to localhost:0 (to get random port)
+   * - get address
+   * - listen
+   */
+  tcp::acceptor acceptor;
+  acceptor.reuse_address(true);
+  tcp::peer local(address::v4::loopback());
+  acceptor.bind(local);
+  acceptor.listen(SOMAXCONN);
+
+  log_.debug("listening for interruptions at %d", acceptor.id());
+  /*
+   * setup connection
+   * - connect to server
+   * - accept client
+   * - prepare server
+   */
+  client_.connect(tcp::peer(address::v4::loopback(), local.port()));
+  acceptor.accept(server_);
+  client_.non_blocking(true);
+  client_.options(TCP_NODELAY, true);
+}
+
+socket_interrupter::~socket_interrupter()
+{
+  client_.close();
+  server_.close();
+}
+
+socket_type socket_interrupter::socket_id() const
+{
+  return server_.id();
+}
+
+void socket_interrupter::interrupt()
+{
+  buffer_view buf(indicator_);
+  log_.debug("fd %d: sending interrupt to fd %d", client_.id(), server_.id());
+  client_.send(buf);
+}
+
+bool socket_interrupter::reset()
+{
+  buffer_view buf(consumer_);
+  log_.debug("reading interrupt byte");
+  auto nread = server_.receive(buf);
+  bool interrupted = nread > 0;
+  while (nread == static_cast<ssize_t>(buf.capacity())) {
+    nread = server_.receive(buf);
+  }
+  return interrupted;
+}
+
+}

source/core/net/stream_handler.cpp

@@ -0,0 +1,153 @@
+#include "matador/net/stream_handler.hpp"
+#include "matador/net/handler_creator.hpp"
+#include "matador/net/reactor.hpp"
+
+#include "matador/utils/buffer_view.hpp"
+
+#include "matador/logger/log_manager.hpp"
+
+#include <cerrno>
+#include <chrono>
+#include <iostream>
+
+namespace matador {
+
+stream_handler::stream_handler(tcp::socket sock, tcp::peer endpoint, handler_creator *creator, t_init_handler init_handler)
+  : log_(create_logger("StreamHandler"))
+  , stream_(std::move(sock))
+  , endpoint_(std::move(endpoint))
+  , name_(endpoint.to_string() + " (fd: " + std::to_string(stream_.id()) + ")")
+  , creator_(creator)
+  , init_handler_(std::move(init_handler))
+{
+
+  log_.debug("%s: created stream handler", name_.c_str());
+}
+
+void stream_handler::open()
+{
+  init_handler_(endpoint_, *this);
+}
+
+socket_type stream_handler::handle() const
+{
+  return stream_.id();
+}
+
+void stream_handler::on_input()
+{
+  auto len = stream_.receive(read_buffer_);
+  log_.trace("%s: read %d bytes", name().c_str(), len);
+  if (len == 0) {
+    on_close();
+  } else if (len < 0 && errno != EWOULDBLOCK) {
+    char error_buffer[1024];
+    log_.error("%s: error on read: %s", name().c_str(), os::strerror(errno, error_buffer, 1024));
+    is_ready_to_read_ = false;
+    on_read_(static_cast<long>(len), static_cast<long>(len));
+    on_close();
+  } else {
+    log_.debug("%s: received %d bytes (data: %s)", name().c_str(), len, read_buffer_.data());
+    read_buffer_.bump(len);
+    is_ready_to_read_ = false;
+    on_read_(0, static_cast<long>(len));
+  }
+}
+
+void stream_handler::on_output()
+{
+  ssize_t bytes_total = 0;
+  auto start = std::chrono::high_resolution_clock::now();
+  while (!write_buffers_.empty()) {
+    buffer_view &bv = write_buffers_.front();
+
+    auto len = stream_.send(bv);
+    log_.trace("%s: sent %d bytes", name().c_str(), len);
+
+    if (len == 0) {
+      on_close();
+    } else if (len < 0 && errno != EWOULDBLOCK) {
+      char error_buffer[1024];
+      log_.error("%s: error on write: %s", name().c_str(), os::strerror(errno, error_buffer, 1024));
+      on_close();
+      is_ready_to_write_ = false;
+      on_write_(static_cast<long>(len), static_cast<long>(len));
+    } else if (len < 0 && errno == EWOULDBLOCK) {
+      log_.debug("%s: sent %d bytes (blocked)", name().c_str(), bytes_total);
+    } else {
+      bytes_total += len;
+      bv.bump(len);
+      if (bv.full()) {
+        write_buffers_.pop_front();
+      }
+    }
+  }
+  auto end = std::chrono::high_resolution_clock::now();
+  auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
+  log_.debug("%s: sent %d bytes (%dms)", name().c_str(), bytes_total, elapsed);
+  is_ready_to_write_ = false;
+  on_write_(0, static_cast<long>(bytes_total));
+}
+
+void stream_handler::on_close()
+{
+  log_.debug("%s: closing connection", name().c_str(), handle());
+  stream_.close();
+  creator_->notify_close( this );
+  auto self = shared_from_this();
+  get_reactor()->mark_handler_for_delete(self);
+  get_reactor()->unregister_handler(self, event_type::READ_WRITE_MASK);
+}
+
+void stream_handler::close()
+{
+  if (!stream_.is_open()) {
+    return;
+  }
+  log_.debug("%s: closing connection", name().c_str(), handle());
+  stream_.close();
+  creator_->notify_close( this );
+}
+
+bool stream_handler::is_ready_write() const
+{
+  return is_ready_to_write_ && !write_buffers_.empty();
+}
+
+bool stream_handler::is_ready_read() const
+{
+  return is_ready_to_read_ && !read_buffer_.full();
+}
+
+void stream_handler::read(buffer_view buf, t_read_handler read_handler)
+{
+  on_read_ = std::move(read_handler);
+  read_buffer_ = std::move(buf);
+  is_ready_to_read_ = true;
+  get_reactor()->interrupt();
+}
+
+void stream_handler::write(std::list<buffer_view> buffers, io_stream::t_write_handler write_handler)
+{
+  on_write_ = std::move(write_handler);
+  write_buffers_ = std::move(buffers);
+  is_ready_to_write_ = true;
+  get_reactor()->interrupt();
+}
+
+void stream_handler::close_stream()
+{
+  on_close();
+}
+
+tcp::socket &stream_handler::stream()
+{
+  return stream_;
+}
+
+std::string stream_handler::name() const
+{
+  return name_;
+}
+
+}