bahman/scylla
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

checkpoint

Browse Source
This commit is contained in:
Avi Kivity
2014-08-11 17:59:46 +03:00
parent 30b73db070
commit 1b05c9ab9c
4 changed files with 118 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
Makefile
View File

@@ -1,7 +1,16 @@
mode = release
sanitize = -fsanitize=address -fsanitize=leak -fsanitize=undefined
CXXFLAGS = -std=gnu++1y -g -Wall -O0 -MD -MT $@ -MP -flto $(sanitize)
sanitize.debug = -fsanitize=address -fsanitize=leak -fsanitize=undefined
sanitize.release =
opt.debug = -O0
opt.release = -O2 -flto
sanitize = $(sanitize.$(mode))
opt = $(opt.$(mode))
CXXFLAGS = -std=gnu++1y -g -Wall $(opt) -MD -MT $@ -MP -flto $(sanitize) -fvisibility=hidden
tests = test-reactor

30
reactor.cc
View File

@@ -21,7 +21,7 @@ reactor::~reactor() {
void reactor::epoll_add_in(pollable_fd& pfd, std::unique_ptr<task> t) {
auto ctl = pfd.events ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;
pfd.events |= EPOLLIN | EPOLLONESHOT;
pfd.events |= EPOLLIN;
assert(!pfd.pollin);
pfd.pollin = std::move(t);
::epoll_event eevt;
@@ -33,7 +33,7 @@ void reactor::epoll_add_in(pollable_fd& pfd, std::unique_ptr<task> t) {
void reactor::epoll_add_out(pollable_fd& pfd, std::unique_ptr<task> t) {
auto ctl = pfd.events ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;
pfd.events |= EPOLLOUT | EPOLLONESHOT;
pfd.events |= EPOLLOUT;
assert(!pfd.pollout);
pfd.pollout = std::move(t);
::epoll_event eevt;
@@ -43,9 +43,14 @@ void reactor::epoll_add_out(pollable_fd& pfd, std::unique_ptr<task> t) {
assert(r == 0);
}
void reactor::forget(pollable_fd& fd) {
if (fd.events) {
::epoll_ctl(_epollfd, EPOLL_CTL_DEL, fd.fd, nullptr);
}
}
std::unique_ptr<pollable_fd>
reactor::listen(socket_address sa, listen_options opts)
{
reactor::listen(socket_address sa, listen_options opts) {
int fd = ::socket(sa.u.sa.sa_family, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
assert(fd != -1);
if (opts.reuse_address) {
@@ -55,7 +60,7 @@ reactor::listen(socket_address sa, listen_options opts)
int r = ::bind(fd, &sa.u.sa, sizeof(sa.u.sas));
assert(r != -1);
::listen(fd, 100);
return std::unique_ptr<pollable_fd>(new pollable_fd(fd));
return std::unique_ptr<pollable_fd>(new pollable_fd(*this, fd));
}
void reactor::run() {
@@ -67,13 +72,20 @@ void reactor::run() {
auto& evt = eevt[i];
auto pfd = reinterpret_cast<pollable_fd*>(evt.data.ptr);
auto events = evt.events;
pfd->events = 0;
::epoll_ctl(_epollfd, EPOLL_CTL_DEL, pfd->fd, &evt);
std::unique_ptr<task> t_in, t_out;
if (events & EPOLLIN) {
auto t = std::move(pfd->pollin);
t->run();
t_in = std::move(pfd->pollin);
}
if (events & EPOLLOUT) {
auto t = std::move(pfd->pollout);
t->run();
t_out = std::move(pfd->pollout);
}
if (t_in) {
t_in->run();
}
if (t_out) {
t_out->run();
}
}
}

75
reactor.hh
View File

@@ -19,6 +19,7 @@
#include <cassert>
#include <stdexcept>
#include <iostream>
#include <unistd.h>
class socket_address;
class reactor;
@@ -131,7 +132,6 @@ struct future_state {
assert(_state == state::future);
_state = state::result;
new (&_u.value) T(std::forward(a)...);
std::cout << "checking task at " << &_task << "\n";
if (_task) {
_task->run();
}
@@ -155,7 +155,6 @@ struct future_state {
}
template <typename Func>
void schedule(Func&& func) {
std::cout << "scheduling task at " << &_task << "\n";
_task = make_task(std::forward<Func>(func));
}
};
@@ -203,14 +202,26 @@ public:
T get() {
return _state->get();
}
template <typename Func, typename Enable>
void then(Func, Enable);
template <typename Func>
void then(Func&& func) {
void then(Func&& func, std::enable_if_t<std::is_same<std::result_of_t<Func(future&&)>, void>::value, void*> = nullptr) {
auto state = _state;
state->schedule([fut = std::move(*this), func = std::forward<Func>(func)] () mutable {
std::cout << "running task\n";
func(std::move(fut));
});
}
#if 0
template <typename Func>
auto then(Func&& func) -> std::enable_if<is_future<decltype(func(*this)), void>::value, decltype(func(*this))>::type {
auto state = _state;
state->schedule([fut = std::move(*this), func = std::forward<Func>(func)] () mutable {
func(std::move(fut));
});
}
#endif
friend class promise<T>;
};
@@ -236,6 +247,7 @@ public:
private:
void epoll_add_in(pollable_fd& fd, std::unique_ptr<task> t);
void epoll_add_out(pollable_fd& fd, std::unique_ptr<task> t);
void forget(pollable_fd& fd);
void abort_on_error(int ret);
public:
reactor();
@@ -249,16 +261,27 @@ public:
future<size_t> read_some(pollable_fd& fd, void* buffer, size_t size);
future<size_t> write_some(pollable_fd& fd, void* buffer, size_t size);
future<size_t> write_all(pollable_fd& fd, void* buffer, size_t size);
void run();
private:
void write_all_part(pollable_fd& fd, void* buffer, size_t size,
promise<size_t> result, size_t completed);
friend class pollable_fd;
};
class pollable_fd {
public:
~pollable_fd() { r.forget(*this); ::close(fd); }
protected:
explicit pollable_fd(int fd) : fd(fd) {}
explicit pollable_fd(reactor& r, int fd) : r(r), fd(fd) {}
pollable_fd(const pollable_fd&) = delete;
void operator=(const pollable_fd&) = delete;
reactor& r;
int fd;
int events = 0;
std::unique_ptr<task> pollin;
@@ -271,12 +294,12 @@ future<accept_result>
reactor::accept(pollable_fd& listenfd) {
promise<accept_result> pr;
future<accept_result> fut = pr.get_future();
epoll_add_in(listenfd, make_task([pr = std::move(pr), lfd = listenfd.fd] () mutable {
epoll_add_in(listenfd, make_task([this, pr = std::move(pr), lfd = listenfd.fd] () mutable {
socket_address sa;
socklen_t sl = sizeof(&sa.u.sas);
int fd = ::accept4(lfd, &sa.u.sa, &sl, SOCK_NONBLOCK | SOCK_CLOEXEC);
assert(fd != -1);
pr.set_value(accept_result{std::unique_ptr<pollable_fd>(new pollable_fd(fd)), sa});
pr.set_value(accept_result{std::unique_ptr<pollable_fd>(new pollable_fd(*this, fd)), sa});
}));
return fut;
}
@@ -294,5 +317,43 @@ reactor::read_some(pollable_fd& fd, void* buffer, size_t len) {
return fut;
}
inline
future<size_t>
reactor::write_some(pollable_fd& fd, void* buffer, size_t len) {
promise<size_t> pr;
auto fut = pr.get_future();
epoll_add_out(fd, make_task([pr = std::move(pr), sfd = fd.fd, buffer, len] () mutable {
ssize_t r = ::send(sfd, buffer, len, 0);
assert(r != -1);
pr.set_value(r);
}));
return fut;
}
inline
void
reactor::write_all_part(pollable_fd& fd, void* buffer, size_t len,
promise<size_t> result, size_t completed) {
if (completed == len) {
result.set_value(completed);
} else {
write_some(fd, static_cast<char*>(buffer) + completed, len - completed).then(
[&fd, buffer, len, result = std::move(result), completed, this] (future<size_t> part) mutable {
write_all_part(fd, buffer, len, std::move(result), completed + part.get());
});
}
}
inline
future<size_t>
reactor::write_all(pollable_fd& fd, void* buffer, size_t len) {
assert(len);
promise<size_t> pr;
auto fut = pr.get_future();
write_all_part(fd, buffer, len, std::move(pr), 0);
return fut;
}
#endif /* REACTOR_HH_ */

28
test-reactor.cc
View File

@@ -12,23 +12,31 @@ struct test {
reactor r;
std::unique_ptr<pollable_fd> listener;
struct connection {
connection(reactor& r, std::unique_ptr<pollable_fd> fd) : r(r), fd(std::move(fd)) {}
reactor& r;
std::unique_ptr<pollable_fd> fd;
char buffer[8192];
void copy_data() {
r.read_some(*fd, buffer, sizeof(buffer)).then([this] (future<size_t> fut) {
auto n = fut.get();
if (n) {
r.write_all(*fd, buffer, n).then([this, n] (future<size_t> fut) {
if (fut.get() == n) {
copy_data();
}
});
} else {
delete this;
}
});
}
};
void new_connection(accept_result&& accepted) {
std::cout << "got connection\n";
copy_data(std::move(std::get<0>(accepted)));
}
void copy_data(std::unique_ptr<pollable_fd> fd) {
char buffer[8192];
r.read_some(*fd, buffer, sizeof(buffer)).then([this] (future<size_t> fut) {
auto n = fut.get();
std::cout << "got data: " << n << "\n";
});
auto c = new connection(r, std::move(std::get<0>(accepted)));
c->copy_data();
}
void start_accept() {
r.accept(*listener).then([this] (future<accept_result> fut) {
std::cout << "accept future returned\n";
new_connection(fut.get());
start_accept();
});