Commit 41fada1c authored by octal's avatar octal

Completly god rid of the old I/O subsystem, RIP

parent f8b2d8fc
/* io.h
Mathieu Stefani, 05 novembre 2015
I/O handling
*/
#pragma once
#include "flags.h"
#include "os.h"
#include "async.h"
#include "net.h"
#include "prototype.h"
#include <thread>
#include <mutex>
#include <memory>
#include <unordered_map>
#include <sys/time.h>
#include <sys/resource.h>
namespace Io {
class Service;
class Message;
class FdSet {
public:
friend class Service;
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:
FdSet(std::vector<Polling::Event>&& events)
{
events_.reserve(events.size());
for (auto &&event: events) {
events_.push_back(std::move(event));
}
}
std::vector<Entry> events_;
};
class Handler;
class Service {
public:
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::Mode mode = Polling::Mode::Level);
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 modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode = Polling::Mode::Level);
void init(const std::shared_ptr<Handler>& handler);
void run();
void shutdown();
std::thread::id thread() const { return thisId; }
std::shared_ptr<Handler> handler() const { return handler_; }
private:
std::thread::id thisId;
std::shared_ptr<Handler> handler_;
std::atomic<bool> shutdown_;
NotifyFd shutdownFd;
Polling::Epoll poller;
};
class ServiceGroup {
public:
void init(
size_t threads,
const std::shared_ptr<Handler>& handler);
void start();
void shutdown();
std::shared_ptr<Service> service(Fd fd) const;
std::shared_ptr<Service> service(size_t index) const;
std::vector<Async::Promise<rusage>> load() const;
size_t size() const {
return workers_.size();
}
bool empty() const {
return workers_.empty();
}
private:
class Worker {
public:
Worker();
~Worker();
void init(const std::shared_ptr<Handler>& handler);
Async::Promise<rusage> load();
void run();
void shutdown();
std::shared_ptr<Service> service() const { return service_; }
private:
std::unique_ptr<std::thread> thread_;
std::shared_ptr<Service> service_;
};
std::vector<std::unique_ptr<Worker>> workers_;
};
class Handler : public Prototype<Handler> {
public:
friend class Service;
virtual void onReady(const FdSet& fds) = 0;
virtual void registerPoller(Polling::Epoll& poller) { }
virtual Async::Promise<rusage> load() {
return Async::Promise<rusage>::rejected(std::runtime_error("Unimplemented"));
}
Service* io() const {
return io_;
}
private:
Service* io_;
};
} // namespace Io
......@@ -11,7 +11,6 @@
#include "os.h"
#include "flags.h"
#include "async.h"
#include "io.h"
#include "reactor.h"
#include <vector>
#include <memory>
......
......@@ -16,7 +16,6 @@
#include <atomic>
#include <unistd.h>
#include "os.h"
#include "io.h"
#include "reactor.h"
namespace Net {
......
......@@ -6,7 +6,6 @@
#pragma once
#include "io.h"
#include "reactor.h"
#include "mailbox.h"
#include "optional.h"
......
/* io.cc
Mathieu Stefani, 05 novembre 2015
I/O handling
*/
#include <thread>
#include "io.h"
#include "peer.h"
#include "os.h"
#include <sys/timerfd.h>
namespace Io {
void
Service::registerFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode) {
poller.addFd(fd, interest, Polling::Tag(fd), mode);
}
void
Service::registerFdOneShot(Fd fd, Polling::NotifyOn interest, Polling::Mode mode) {
poller.addFdOneShot(fd, interest, Polling::Tag(fd), mode);
}
void
Service::modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode) {
poller.rearmFd(fd, interest, Polling::Tag(fd), mode);
}
void
Service::registerFd(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode) {
poller.addFd(fd, interest, tag, mode);
}
void
Service::registerFdOneShot(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode) {
poller.addFdOneShot(fd, interest, tag, mode);
}
void
Service::modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode) {
poller.rearmFd(fd, interest, tag, mode);
}
void
Service::init(const std::shared_ptr<Handler>& handler) {
handler_ = handler;
handler_->io_ = this;
handler_->registerPoller(poller);
shutdown_.store(false);
}
void
Service::shutdown() {
shutdown_.store(true);
shutdownFd.notify();
}
void
Service::run() {
if (!handler_)
throw std::runtime_error("You need to set a handler before running an io service");
shutdownFd.bind(poller);
thisId = std::this_thread::get_id();
std::chrono::milliseconds timeout(-1);
for (;;) {
std::vector<Polling::Event> events;
int ready_fds;
switch(ready_fds = poller.poll(events, 1024, timeout)) {
case -1:
break;
case 0:
timeout = std::chrono::milliseconds(-1);
break;
default:
if (shutdown_) return;
FdSet set(std::move(events));
handler_->onReady(set);
}
}
}
void
ServiceGroup::init(
size_t threads,
const std::shared_ptr<Handler>& handler) {
for (size_t i = 0; i < threads; ++i) {
std::unique_ptr<Worker> wrk(new Worker);
wrk->init(handler->clone());
workers_.push_back(std::move(wrk));
}
}
void
ServiceGroup::start() {
for (auto& worker: workers_) {
worker->run();
}
}
std::vector<Async::Promise<rusage>>
ServiceGroup::load() const {
std::vector<Async::Promise<rusage>> loads;
loads.reserve(workers_.size());
for (auto& worker: workers_) {
loads.push_back(worker->load());
}
return loads;
}
void
ServiceGroup::shutdown() {
for (auto& worker: workers_)
worker->shutdown();
}
std::shared_ptr<Service>
ServiceGroup::service(Fd fd) const {
size_t worker = fd % workers_.size();
auto& wrk = workers_[worker];
return wrk->service();
}
std::shared_ptr<Service>
ServiceGroup::service(size_t index) const {
if (index >= workers_.size())
throw std::out_of_range("index out of range");
return workers_[index]->service();
}
ServiceGroup::Worker::Worker() {
service_.reset(new Service);
}
ServiceGroup::Worker::~Worker() {
if (thread_) thread_->join();
}
void
ServiceGroup::Worker::init(const std::shared_ptr<Handler>& handler) {
service_->init(handler);
}
Async::Promise<rusage>
ServiceGroup::Worker::load() {
return service_->handler()->load();
}
void
ServiceGroup::Worker::run() {
thread_.reset(new std::thread([=]() {
service_->run();
}));
}
void
ServiceGroup::Worker::shutdown() {
service_->shutdown();
}
}
......@@ -5,7 +5,6 @@
#include "os.h"
#include "common.h"
#include "io.h"
#include <unistd.h>
#include <fcntl.h>
#include <fstream>
......
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