Initial commit

Makefile

@@ -0,0 +1,16 @@
+
+
+CXXFLAGS = -std=gnu++1y -g -Wall -O2 -MD -MT $@ -MP -flto
+
+tests = test-reactor
+
+all: seastar $(tests)
+
+seastar: main.o reactor.o
+	$(CXX) $(CXXFLAGS) -o $@ $^
+
+test-reactor: test-reactor.o reactor.o
+	$(CXX) $(CXXFLAGS) -o $@ $^
+
+
+-include *.d

main.cc

@@ -0,0 +1,18 @@
+/*
+ * main.cc
+ *
+ *  Created on: Aug 1, 2014
+ *      Author: avi
+ */
+
+
+#include "reactor.hh"
+
+int main(int ac, char** av)
+{
+    reactor r;
+    r.run();
+    return 0;
+}
+
+

reactor.cc

@@ -0,0 +1,85 @@
+/*
+ * reactor.cc
+ *
+ *  Created on: Aug 1, 2014
+ *      Author: avi
+ */
+
+#include "reactor.hh"
+#include <cassert>
+#include <unistd.h>
+#include <fcntl.h>
+
+reactor::reactor()
+    : _epollfd(epoll_create1(EPOLL_CLOEXEC)) {
+    assert(_epollfd != -1);
+}
+
+reactor::~reactor() {
+    ::close(_epollfd);
+}
+
+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;
+    assert(!pfd.pollin);
+    pfd.pollin = std::move(t);
+    ::epoll_event eevt;
+    eevt.events = pfd.events;
+    eevt.data.ptr = &pfd;
+    int r = ::epoll_ctl(_epollfd, ctl, pfd.fd, &eevt);
+    assert(r == 0);
+}
+
+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;
+    assert(!pfd.pollout);
+    pfd.pollout = std::move(t);
+    ::epoll_event eevt;
+    eevt.events = pfd.events;
+    eevt.data.ptr = &pfd;
+    int r = ::epoll_ctl(_epollfd, ctl, pfd.fd, &eevt);
+    assert(r == 0);
+}
+
+std::unique_ptr<pollable_fd>
+reactor::listen(socket_address sa)
+{
+    int fd = ::socket(sa.u.sa.sa_family, SOCK_STREAM, SOCK_NONBLOCK | SOCK_CLOEXEC);
+    assert(fd != -1);
+    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));
+}
+
+void reactor::run() {
+    while (true) {
+        std::array<epoll_event, 128> eevt;
+        int nr = ::epoll_wait(_epollfd, eevt.data(), eevt.size(), -1);
+        assert(nr != -1);
+        for (int i = 0; i < nr; ++i) {
+            auto& evt = eevt[i];
+            auto pfd = reinterpret_cast<pollable_fd*>(evt.data.ptr);
+            auto events = evt.events;
+            if (events & EPOLLIN) {
+                auto t = std::move(pfd->pollin);
+                t->run();
+            }
+            if (events & EPOLLOUT) {
+                auto t = std::move(pfd->pollout);
+                t->run();
+            }
+        }
+    }
+}
+
+socket_address make_ipv4_address(ipv4_addr addr) {
+    socket_address sa;
+    sa.u.in.sin_family = AF_INET;
+    sa.u.in.sin_port = htons(addr.port);
+    std::memcpy(&sa.u.in.sin_addr, addr.host, 4);
+    return sa;
+}
+

reactor.hh

@@ -0,0 +1,168 @@
+/*
+ * reactor.hh
+ *
+ *  Created on: Aug 1, 2014
+ *      Author: avi
+ */
+
+#ifndef REACTOR_HH_
+#define REACTOR_HH_
+
+#include <memory>
+#include <libaio.h>
+#include <sys/epoll.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <unordered_map>
+#include <netinet/ip.h>
+#include <cstring>
+#include <cassert>
+
+class socket_address;
+class reactor;
+class pollable_fd;
+
+struct ipv4_addr {
+    uint8_t host[4];
+    uint16_t port;
+};
+
+socket_address make_ipv4_address(ipv4_addr addr);
+
+class socket_address {
+private:
+    union {
+        ::sockaddr_storage sas;
+        ::sockaddr sa;
+        ::sockaddr_in in;
+    } u;
+    friend socket_address make_ipv4_address(ipv4_addr addr);
+    friend class reactor;
+};
+
+template <typename T>
+class promise;
+
+template <typename T>
+struct future_state {
+    virtual ~future_state();
+    promise<T>* promise;
+    bool value_valid = false;
+    bool ex_valid = false;
+    union {
+        T value;
+        std::exception_ptr ex;
+    } u;
+    void set(const T& value);
+    void set(T&& value);
+    void set_exception(std::exception_ptr ex);
+    T get() {
+        while (promise) {
+            promise->wait();
+        }
+        if (ex) {
+            std::rethrow_exception(ex);
+        }
+        return std::move(u.value);
+    }
+};
+
+template <typename T>
+class future {
+    std::unique_ptr<future_state<T>> _state;
+public:
+    T get() {
+        return _state.get();
+    }
+    template <typename Func>
+    void then(Func func) {
+
+    }
+};
+
+class reactor {
+    class task;
+public:
+    int _epollfd;
+    io_context_t _io_context;
+private:
+    class task {
+    public:
+        virtual ~task() {}
+        virtual void run() = 0;
+    };
+    template <typename Func>
+    class lambda_task : public task {
+        Func _func;
+    public:
+        lambda_task(Func func) : _func(func) {}
+        virtual void run() { _func(); }
+    };
+
+    template <typename Func>
+    std::unique_ptr<task>
+    make_task(Func func) {
+        return std::make_unique<lambda_task<Func>>(func);
+    }
+
+    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 abort_on_error(int ret);
+public:
+    reactor();
+    ~reactor();
+
+    std::unique_ptr<pollable_fd> listen(socket_address sa);
+
+    template <typename Func>
+    void accept(pollable_fd& listenfd, Func with_pfd_sockaddr);
+
+    future<std::unique_ptr<pollable_fd>> accept(pollable_fd& listen_fd)
+
+    future<size_t> read_some(pollable_fd& fd, void* buffer, size_t size);
+    template <typename Func>
+    void read_some(pollable_fd& fd, void* buffer, size_t len, Func with_len);
+
+    void run();
+
+    friend class pollable_fd;
+};
+
+class pollable_fd {
+protected:
+    explicit pollable_fd(int fd) : fd(fd) {}
+    pollable_fd(const pollable_fd&) = delete;
+    void operator=(const pollable_fd&) = delete;
+    int fd;
+    int events = 0;
+    std::unique_ptr<reactor::task> pollin;
+    std::unique_ptr<reactor::task> pollout;
+    friend class reactor;
+};
+
+template <typename Func>
+inline
+void reactor::accept(pollable_fd& listenfd, Func with_pfd_sockaddr) {
+    auto lfd = listenfd.fd;
+    epoll_add_in(listenfd, make_task([=] {
+        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);
+        auto pfd = std::unique_ptr<pollable_fd>(new pollable_fd(fd));
+        with_pfd_sockaddr(std::move(pfd), sa);
+    }));
+}
+
+template <typename Func>
+void reactor::read_some(pollable_fd& fd, void* buffer, size_t len, Func with_len) {
+    auto rfd = fd.fd;
+    epoll_add_in(fd, make_task([=] {
+        ssize_t r = ::recv(rfd, buffer, len, 0);
+        assert(r != -1);
+        with_len(len);
+    }));
+}
+
+
+#endif /* REACTOR_HH_ */

sstring.hh

@@ -0,0 +1,124 @@
+/*
+ * sstring.hh
+ *
+ *  Created on: Jul 31, 2014
+ *      Author: avi
+ */
+
+#ifndef SSTRING_HH_
+#define SSTRING_HH_
+
+#include <stdint.h>
+#include <algorithm>
+#include <string>
+#include <cstring>
+#include <stdexcept>
+
+template <typename char_type, typename size_type, size_type max_size>
+class basic_sstring {
+    union {
+        struct external_type {
+            char* str;
+            size_type size;
+        } external;
+        struct internal_type {
+            char str[max_size - 1];
+            int8_t size;
+        } internal;
+        static_assert(sizeof(external_type) < sizeof(internal_type), "max_size too small");
+        static_assert(max_size <= 127, "max_size too large");
+    } u;
+    bool is_internal() const noexcept {
+        return u.internal.size >= 0;
+    }
+    bool is_external() const noexcept {
+        return !is_internal();
+    }
+    const char* str() const {
+        return is_internal() ? u.internal.str : u.external.str;
+    }
+    char* str() {
+        return is_internal() ? u.internal.str : u.external.str;
+    }
+public:
+    basic_sstring() noexcept {
+        u.internal.size = 0;
+        u.internal.str[0] = '\0';
+    }
+    basic_sstring(const basic_sstring& x) {
+        if (x.is_internal()) {
+            u.internal = x.u.internal;
+        } else {
+            u.external.str = new char[x.u.external.size + 1];
+            std::copy(x.u.str, x.u.str + x.u.extenal.size + 1, u.external.str);
+            u.external.size = x.u.external.size;
+        }
+    }
+    basic_sstring(basic_sstring&& x) noexcept {
+        u = x.u;
+        x.u.internal.size = 0;
+        x.u.internal.str[0] = '\0';
+    }
+    basic_sstring(const char* x, size_t len) {
+        if (size_type(size) != size) {
+            throw std::overflow_error("sstring overflow");
+        }
+        if (size + 1 <= sizeof(u.internal.str)) {
+            std::copy(x, x + size + 1, u.internal.str);
+            u.internal.size = size;
+        } else {
+            u.internal.size = -1;
+            u.external.str = new char[size + 1];
+            u.external.size = size;
+            std::copy(x, x + size + 1, u.external.str);
+        }
+    }
+    basic_sstring(const char* x) : basic_sstring(x, std::strlen(x)) {}
+    basic_sstring(std::string& x) : basic_sstring(x.c_str(), x.size()) {}
+    ~basic_sstring() noexcept {
+        if (!is_external()) {
+            delete[] u.external.str;
+        }
+    }
+    basic_sstring& operator=(const basic_sstring& x) {
+        basic_sstring tmp(x);
+        swap(tmp);
+    }
+    basic_sstring& operator=(basic_sstring&& x) noexcept {
+        reset();
+        swap(x);
+    }
+    operator std::string() const {
+        return str();
+    }
+    size_t size() const noexcept {
+        return is_internal() ? u.internal.size : u.external.size;
+    }
+    bool empty() const noexcept {
+        return u.internal.size == 0;
+    }
+    void reset() noexcept {
+        if (is_external()) {
+            delete u.external.str;
+        }
+        u.internal.size = 0;
+    }
+    void swap(basic_sstring& x) noexcept {
+        basic_sstring tmp;
+        tmp.u = x.u;
+        x.u = u;
+        u = tmp.u;
+    }
+};
+
+template <typename char_type, typename size_type, size_type max_size>
+inline
+void swap(basic_sstring<char_type, size_type, max_size>& x,
+          basic_sstring<char_type, size_type, max_size>& y) noexcept
+{
+    return x.swap(y);
+}
+
+using sstring = basic_sstring<char, uint32_t, 15>;
+
+#endif /* SSTRING_HH_ */

test-reactor.cc

@@ -0,0 +1,29 @@
+/*
+ * test-reactor.cc
+ *
+ *  Created on: Aug 2, 2014
+ *      Author: avi
+ */
+
+#include "reactor.hh"
+#include <iostream>
+
+struct test {
+    reactor r;
+    std::unique_ptr<pollable_fd> listener;
+    void on_accept(std::unique_ptr<pollable_fd> pfd, socket_address sa) {
+        std::cout << "got connection\n";
+        r.accept(*listener, [=] (std::unique_ptr<pollable_fd> pfd, socket_address sa) {
+            on_accept(std::move(pfd), sa);
+        });
+    }
+
+int main(int ac, char** av)
+{
+    test t;
+    ipv4_addr addr{{}, 10000};
+    t.listener = r.listen(make_ipv4_address(addr));
+    r.accept(*listener, [&]
+    r.run();
+}
+