Skip to content

P
pistache
Commit c9a8e990 authored by octal's avatar octal
Browse files
Options

Squashed commit of the following: 

commit e60bcb0bc15fbba453340d69d106c5850b9a6d67
Author: octal <mathieu.stefani@supinfo.com>
Date:   Thu Nov 12 14:14:53 2015 +0100

    Added a NoExcept concept that will call std::terminate if a promise has been rejected

commit bae200046427b93e6f124c848ab36d246a7580c3
Author: octal <mathieu.stefani@supinfo.com>
Date:   Wed Nov 11 14:50:27 2015 +0100

    Updated Optional to be able to move the result out when chaining a Promise

commit da60bc8b02a1fb89a91985a0da9ddf4b7219f5b3
Author: octal <mathieu.stefani@supinfo.com>
Date:   Fri Nov 6 17:43:36 2015 +0100

    Started working on a proof-of-concept implementation of an asynchronous primitive based on the Promises/A+ specification
parent 0d8215d1
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 947 additions and 276 deletions
main.cc
View file @ c9a8e990
......@@ -18,7 +18,9 @@ class MyHandler : public Net::Http::Handler {
.add(std::make_shared<Header::Server>("lys"))
.add(std::make_shared<Header::ContentType>(MIME(Text, Plain)));
response.send(Net::Http::Code::Ok, "PONG");
response.send(Net::Http::Code::Ok, "PONG").then([](ssize_t bytes) {
std::cout << "Sent total of " << bytes << " bytes" << std::endl;
}, Async::IgnoreException);
}
}
......
src/CMakeLists.txt
View file @ c9a8e990
set(SOURCE_FILES
tcp.cc
io.cc
listener.cc
net.cc
peer.cc
......
src/async.h 0 → 100644
View file @ c9a8e990
This diff is collapsed.
src/http.cc
View file @ c9a8e990
......@@ -361,12 +361,12 @@ Response::associatePeer(const std::shared_ptr<Tcp::Peer>& peer)
peer_ = peer;
}
ssize_t
Async::Promise<ssize_t>
Response::send(Code code) {
return send(code, "");
}
ssize_t
Async::Promise<ssize_t>
Response::send(Code code, const std::string& body, const Mime::MediaType& mime)
{
char buffer[Const::MaxBuffer];
......@@ -461,10 +461,6 @@ Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer
}
}
void
Handler::onOutput() {
}
void
Handler::onConnection(const std::shared_ptr<Tcp::Peer>& peer) {
peer->putData(ParserData, std::make_shared<Private::Parser>());
......
src/http.h
View file @ c9a8e990
......@@ -15,6 +15,7 @@
#include "http_defs.h"
#include "stream.h"
#include "mime.h"
#include "async.h"
namespace Net {
......@@ -96,8 +97,8 @@ public:
void setMime(const Mime::MediaType& mime);
ssize_t send(Code code);
ssize_t send(Code code, const std::string& body, const Mime::MediaType &mime = Mime::MediaType());
Async::Promise<ssize_t> send(Code code);
Async::Promise<ssize_t> send(Code code, const std::string& body, const Mime::MediaType &mime = Mime::MediaType());
private:
Response();
......@@ -203,7 +204,6 @@ namespace Private {
class Handler : public Net::Tcp::Handler {
public:
void onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer);
void onOutput();
void onConnection(const std::shared_ptr<Tcp::Peer>& peer);
void onDisconnection(const std::shared_ptr<Tcp::Peer>& peer);
......
src/io.cc 0 → 100644
View file @ c9a8e990
/* io.cc
Mathieu Stefani, 05 novembre 2015
I/O handling
*/
#include <thread>
#include "io.h"
#include "listener.h"
#include "peer.h"
#include "os.h"
namespace Net {
namespace Tcp {
using namespace Polling;
struct Message {
virtual ~Message() { }
enum class Type { Shutdown };
virtual Type type() const = 0;
};
struct ShutdownMessage : public Message {
Type type() const { return Type::Shutdown; }
};
template<typename To>
To *message_cast(const std::unique_ptr<Message>& from)
{
return static_cast<To *>(from.get());
}
IoWorker::IoWorker() {
}
IoWorker::~IoWorker() {
if (thread && thread->joinable()) {
thread->join();
}
}
void
IoWorker::start(const std::shared_ptr<Handler>& handler, Flags<Options> options) {
handler_ = handler;
options_ = options;
thread.reset(new std::thread([this]() {
this->run();
}));
if (pins.count() > 0) {
auto cpuset = pins.toPosix();
auto handle = thread->native_handle();
pthread_setaffinity_np(handle, sizeof (cpuset), &cpuset);
}
}
void
IoWorker::pin(const CpuSet& set) {
pins = set;
if (thread) {
auto cpuset = set.toPosix();
auto handle = thread->native_handle();
pthread_setaffinity_np(handle, sizeof (cpuset), &cpuset);
}
}
void
IoWorker::shutdown() {
mailbox.post(new ShutdownMessage());
}
std::shared_ptr<Peer>&
IoWorker::getPeer(Fd fd)
{
std::unique_lock<std::mutex> guard(peersMutex);
auto it = peers.find(fd);
if (it == std::end(peers))
{
throw std::runtime_error("No peer found for fd: " + std::to_string(fd));
}
return it->second;
}
std::shared_ptr<Peer>&
IoWorker::getPeer(Polling::Tag tag)
{
return getPeer(tag.value());
}
void
IoWorker::handleIncoming(const std::shared_ptr<Peer>& peer) {
char buffer[Const::MaxBuffer];
memset(buffer, 0, sizeof buffer);
ssize_t totalBytes = 0;
int fd = peer->fd();
for (;;) {
ssize_t bytes;
bytes = recv(fd, buffer + totalBytes, Const::MaxBuffer - totalBytes, 0);
if (bytes == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
if (totalBytes > 0) {
handler_->onInput(buffer, totalBytes, peer);
}
} else {
if (errno == ECONNRESET) {
handler_->onDisconnection(peer);
close(fd);
}
else {
throw std::runtime_error(strerror(errno));
}
}
break;
}
else if (bytes == 0) {
handler_->onDisconnection(peer);
close(fd);
break;
}
else {
totalBytes += bytes;
if (totalBytes >= Const::MaxBuffer) {
std::cerr << "Too long packet" << std::endl;
break;
}
}
}
}
void
IoWorker::handleNewPeer(const std::shared_ptr<Peer>& peer)
{
int fd = peer->fd();
{
std::unique_lock<std::mutex> guard(peersMutex);
peers.insert(std::make_pair(fd, peer));
}
peer->io = this;
handler_->onConnection(peer);
poller.addFd(fd, NotifyOn::Read, Polling::Tag(fd), Polling::Mode::Edge);
}
void
IoWorker::run() {
if (pins.count() > 0) {
}
mailbox.bind(poller);
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:
for (const auto& event: events) {
if (event.tag == mailbox.tag()) {
std::unique_ptr<Message> msg(mailbox.clear());
if (msg->type() == Message::Type::Shutdown) {
return;
}
} else {
auto& peer = getPeer(event.tag);
if (event.flags.hasFlag(NotifyOn::Read)) {
handleIncoming(peer);
}
else if (event.flags.hasFlag(NotifyOn::Write)) {
}
}
}
timeout = std::chrono::milliseconds(0);
break;
}
}
}
Async::Promise<ssize_t>
IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) {
return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
ssize_t bytes = ::send(fd, buf, len, 0);
if (bytes < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
toWrite.insert(
std::make_pair(fd,
OnHoldWrite(std::move(resolve), std::move(reject), buf, len)));
poller.addFdOneShot(fd, NotifyOn::Write, Polling::Tag(fd));
}
else {
reject(std::runtime_error("send"));
}
}
else {
resolve(bytes);
}
});
}
} // namespace Tcp
} // namespace Net
src/io.h 0 → 100644
View file @ c9a8e990
/* io.h
Mathieu Stefani, 05 novembre 2015
I/O handling
*/
#pragma once
#include "mailbox.h"
#include "flags.h"
#include "os.h"
#include "tcp.h"
#include "async.h"
#include <thread>
#include <mutex>
#include <memory>
#include <unordered_map>
namespace Net {
namespace Tcp {
class Peer;
class Message;
class Handler;
class IoWorker {
public:
friend class Peer;
PollableMailbox<Message> mailbox;
IoWorker();
~IoWorker();
void start(const std::shared_ptr<Handler> &handler, Flags<Options> options);
void handleNewPeer(const std::shared_ptr<Peer>& peer);
void pin(const CpuSet& set);
void shutdown();
private:
struct OnHoldWrite {
OnHoldWrite(Async::Resolver resolve, Async::Rejection reject,
const void *buf, size_t len)
: resolve(std::move(resolve))
, reject(std::move(reject))
, buf(buf)
, len(len)
{ }
Async::Resolver resolve;
Async::Rejection reject;
const void *buf;
size_t len;
};
Polling::Epoll poller;
std::unique_ptr<std::thread> thread;
mutable std::mutex peersMutex;
std::unordered_map<Fd, std::shared_ptr<Peer>> peers;
std::unordered_map<Fd, OnHoldWrite> toWrite;
std::shared_ptr<Handler> handler_;
Flags<Options> options_;
CpuSet pins;
std::shared_ptr<Peer>& getPeer(Fd fd);
std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
Async::Promise<ssize_t> asyncWrite(Fd fd, const void *buf, size_t len);
void handleIncoming(const std::shared_ptr<Peer>& peer);
void run();
};
} // namespace Tcp
} // namespace Net
src/listener.cc
View file @ c9a8e990
......@@ -3,7 +3,6 @@
*/
#include <thread>
#include <iostream>
#include <sys/socket.h>
#include <unistd.h>
......@@ -40,23 +39,6 @@ namespace {
using Polling::NotifyOn;
struct Message {
virtual ~Message() { }
enum class Type { Shutdown };
virtual Type type() const = 0;
};
struct ShutdownMessage : public Message {
Type type() const { return Type::Shutdown; }
};
template<typename To>
To *message_cast(const std::unique_ptr<Message>& from)
{
return static_cast<To *>(from.get());
}
void setSocketOptions(Fd fd, Flags<Options> options) {
if (options.hasFlag(Options::ReuseAddr)) {
......@@ -82,179 +64,6 @@ void setSocketOptions(Fd fd, Flags<Options> options) {
}
IoWorker::IoWorker() {
}
IoWorker::~IoWorker() {
if (thread && thread->joinable()) {
thread->join();
}
}
void
IoWorker::start(const std::shared_ptr<Handler>& handler, Flags<Options> options) {
handler_ = handler;
options_ = options;
thread.reset(new std::thread([this]() {
this->run();
}));
if (pins.count() > 0) {
auto cpuset = pins.toPosix();
auto handle = thread->native_handle();
pthread_setaffinity_np(handle, sizeof (cpuset), &cpuset);
}
}
void
IoWorker::pin(const CpuSet& set) {
pins = set;
if (thread) {
auto cpuset = set.toPosix();
auto handle = thread->native_handle();
pthread_setaffinity_np(handle, sizeof (cpuset), &cpuset);
}
}
std::shared_ptr<Peer>&
IoWorker::getPeer(Fd fd)
{
std::unique_lock<std::mutex> guard(peersMutex);
auto it = peers.find(fd);
if (it == std::end(peers))
{
throw std::runtime_error("No peer found for fd: " + to_string(fd));
}
return it->second;
}
std::shared_ptr<Peer>&
IoWorker::getPeer(Polling::Tag tag)
{
return getPeer(tag.value());
}
void
IoWorker::handleIncoming(const std::shared_ptr<Peer>& peer) {
char buffer[Const::MaxBuffer];
memset(buffer, 0, sizeof buffer);
ssize_t totalBytes = 0;
int fd = peer->fd();
for (;;) {
ssize_t bytes;
bytes = recv(fd, buffer + totalBytes, Const::MaxBuffer - totalBytes, 0);
if (bytes == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
if (totalBytes > 0) {
handler_->onInput(buffer, totalBytes, peer);
}
} else {
if (errno == ECONNRESET) {
handler_->onDisconnection(peer);
close(fd);
}
else {
throw std::runtime_error(strerror(errno));
}
}
break;
}
else if (bytes == 0) {
handler_->onDisconnection(peer);
close(fd);
break;
}
else {
totalBytes += bytes;
if (totalBytes >= Const::MaxBuffer) {
cerr << "Too long packet" << endl;
break;
}
}
}
}
void
IoWorker::handleNewPeer(const std::shared_ptr<Peer>& peer)
{
int fd = peer->fd();
{
std::unique_lock<std::mutex> guard(peersMutex);
peers.insert(std::make_pair(fd, peer));
}
handler_->onConnection(peer);
poller.addFd(fd, NotifyOn::Read, Polling::Tag(fd), Polling::Mode::Edge);
}
void
IoWorker::run() {
if (pins.count() > 0) {
}
mailbox.bind(poller);
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:
for (const auto& event: events) {
if (event.tag == mailbox.tag()) {
std::unique_ptr<Message> msg(mailbox.clear());
if (msg->type() == Message::Type::Shutdown) {
return;
}
} else {
if (event.flags.hasFlag(NotifyOn::Read)) {
auto& peer = getPeer(event.tag);
handleIncoming(peer);
}
}
}
timeout = std::chrono::milliseconds(0);
break;
}
}
}
void
IoWorker::handleMailbox() {
}
Handler::Handler()
{ }
Handler::~Handler()
{ }
void
Handler::onConnection(const std::shared_ptr<Tcp::Peer>& peer) {
}
void
Handler::onDisconnection(const std::shared_ptr<Tcp::Peer>& peer) {
}
Listener::Listener()
: listen_fd(-1)
, backlog_(Const::MaxBacklog)
......@@ -267,7 +76,6 @@ Listener::Listener(const Address& address)
{
}
void
Listener::init(size_t workers, Flags<Options> options, int backlog)
{
......@@ -400,7 +208,7 @@ Listener::run() {
void
Listener::shutdown() {
for (auto &worker: ioGroup) {
worker->mailbox.post(new ShutdownMessage());
worker->shutdown();
}
}
......
src/listener.h
View file @ c9a8e990
......@@ -6,76 +6,26 @@
#pragma once
#include "tcp.h"
#include "net.h"
#include "mailbox.h"
#include "os.h"
#include "flags.h"
#include "io.h"
#include <vector>
#include <memory>
#include <thread>
#include <unordered_map>
#include <mutex>
namespace Net {
namespace Tcp {
class Peer;
class Message;
class Handler;
enum class Options : uint64_t {
None = 0,
NoDelay = 1,
Linger = NoDelay << 1,
FastOpen = Linger << 1,
QuickAck = FastOpen << 1,
ReuseAddr = QuickAck << 1,
ReverseLookup = ReuseAddr << 1,
InstallSignalHandler = ReverseLookup << 1
};
DECLARE_FLAGS_OPERATORS(Options)
void setSocketOptions(Fd fd, Flags<Options> options);
class IoWorker {
public:
PollableMailbox<Message> mailbox;
IoWorker();
~IoWorker();
void start(const std::shared_ptr<Handler> &handler, Flags<Options> options);
void handleNewPeer(const std::shared_ptr<Peer>& peer);
void pin(const CpuSet& set);
private:
Polling::Epoll poller;
std::unique_ptr<std::thread> thread;
mutable std::mutex peersMutex;
std::unordered_map<Fd, std::shared_ptr<Peer>> peers;
std::shared_ptr<Handler> handler_;
Flags<Options> options_;
CpuSet pins;
std::shared_ptr<Peer>& getPeer(Fd fd);
std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
void handleIncoming(const std::shared_ptr<Peer>& peer);
void handleMailbox();
void run();
};
class Listener {
public:
friend class IoWorker;
friend class Peer;
Listener();
Listener(const Address& address);
......@@ -106,20 +56,6 @@ private:
void dispatchPeer(const std::shared_ptr<Peer>& peer);
};
class Handler {
public:
Handler();
~Handler();
virtual void onInput(const char *buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer) = 0;
virtual void onOutput() = 0;
virtual void onConnection(const std::shared_ptr<Tcp::Peer>& peer);
virtual void onDisconnection(const std::shared_ptr<Tcp::Peer>& peer);
};
} // namespace Tcp
} // namespace Net
src/optional.h
View file @ c9a8e990
......@@ -208,12 +208,17 @@ public:
return *constData();
}
T& unsafeGet() const {
return *data();
}
~Optional() {
if (!isEmpty()) {
data()->~T();
}
}
private:
T *const constData() const {
return const_cast<T *const>(reinterpret_cast<const T *const>(bytes));
......@@ -293,8 +298,22 @@ namespace details {
void static_checks() {
do_static_checks<T, Func>(std::is_bind_expression<Func>());
}
}
template<typename Func>
struct IsArgMovable : public IsArgMovable<decltype(&Func::operator())> { };
template<typename R, typename Class, typename Arg>
struct IsArgMovable<R (Class::*)(Arg) const> : public std::is_rvalue_reference<Arg> { };
template<typename Func, typename Arg>
typename std::conditional<
IsArgMovable<Func>::value,
Arg&&,
const Arg&
>::type tryMove(Arg& arg) {
return std::move(arg);
}
}
template<typename T, typename Func>
const Optional<T>&
......@@ -303,7 +322,7 @@ optionally_do(const Optional<T> &option, Func func) {
static_assert(std::is_same<typename types::callable_trait<Func>::ReturnType, void>::value,
"Use optionally_map if you want to return a value");
if (!option.isEmpty()) {
func(option.get());
func(details::tryMove<Func>(option.unsafeGet()));
}
return option;
......@@ -317,7 +336,7 @@ optionally_map(const Optional<T> &option, Func func)
{
details::static_checks<T, Func>();
if (!option.isEmpty()) {
return Some(func(option.get()));
return Some(func(details::tryMove<Func>(option.unsafeGet())));
}
return None();
......@@ -330,7 +349,7 @@ optionally_fmap(const Optional<T> &option, Func func)
{
details::static_checks<T, Func>();
if (!option.isEmpty()) {
const auto &ret = func(option.get());
const auto &ret = func(details::tryMove<Func>(option.unsafeGet()));
if (!ret.isEmpty()) {
return Some(ret.get());
}
......
src/peer.cc
View file @ c9a8e990
......@@ -4,6 +4,8 @@
*/
#include "peer.h"
#include "async.h"
#include "io.h"
#include <iostream>
#include <stdexcept>
#include <sys/socket.h>
......@@ -75,9 +77,9 @@ Peer::tryGetData(std::string(name)) const {
return it->second;
}
ssize_t
Async::Promise<ssize_t>
Peer::send(const void* buf, size_t len) {
return ::send(fd_, buf, len, 0);
return io->asyncWrite(fd_, buf, len);
}
......
src/peer.h
View file @ c9a8e990
......@@ -8,6 +8,7 @@
#include "net.h"
#include "os.h"
#include "async.h"
#include <string>
#include <iostream>
#include <memory>
......@@ -17,8 +18,12 @@ namespace Net {
namespace Tcp {
class IoWorker;
class Peer {
public:
friend class IoWorker;
Peer();
Peer(const Address& addr);
......@@ -45,9 +50,10 @@ public:
return std::static_pointer_cast<T>(data);
}
ssize_t send(const void* buf, size_t len);
Async::Promise<ssize_t> send(const void* buf, size_t len);
private:
IoWorker* io;
Address addr;
std::string hostname_;
......
src/tcp.cc 0 → 100644
View file @ c9a8e990
/* tcp.cc
Mathieu Stefani, 05 novembre 2015
TCP
*/
#include "tcp.h"
#include "peer.h"
namespace Net {
namespace Tcp {
Handler::Handler()
{ }
Handler::~Handler()
{ }
void
Handler::onConnection(const std::shared_ptr<Tcp::Peer>& peer) {
}
void
Handler::onDisconnection(const std::shared_ptr<Tcp::Peer>& peer) {
}
} // namespace Tcp
} // namespace Net
src/tcp.h 0 → 100644
View file @ c9a8e990
/* tcp.h
Mathieu Stefani, 05 novembre 2015
TCP
*/
#pragma once
#include "flags.h"
#include <memory>
namespace Net {
namespace Tcp {
class Peer;
enum class Options : uint64_t {
None = 0,
NoDelay = 1,
Linger = NoDelay << 1,
FastOpen = Linger << 1,
QuickAck = FastOpen << 1,
ReuseAddr = QuickAck << 1,
ReverseLookup = ReuseAddr << 1,
InstallSignalHandler = ReverseLookup << 1
};
DECLARE_FLAGS_OPERATORS(Options)
class Handler {
public:
Handler();
~Handler();
virtual void onInput(const char *buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer) = 0;
virtual void onConnection(const std::shared_ptr<Tcp::Peer>& peer);
virtual void onDisconnection(const std::shared_ptr<Tcp::Peer>& peer);
};
} // namespace Tcp
} // namespace Net
tests/CMakeLists.txt
View file @ c9a8e990
......@@ -5,3 +5,7 @@ add_test( mime_test run_mime_test )
add_executable( run_headers_test headers_test.cc )
target_link_libraries(run_headers_test gtest gtest_main net)
add_test( headers_test run_headers_test )
add_executable( run_async_test async_test.cc )
target_link_libraries(run_async_test gtest gtest_main net)
add_test( async_test run_async_test )
tests/async_test.cc 0 → 100644
View file @ c9a8e990
#include "gtest/gtest.h"
#include "async.h"
#include <thread>
Async::Promise<int> doAsync(int N)
{
Async::Promise<int> promise(
[=](Async::Resolver& resolve, Async::Rejection& reject) {
std::thread thr([=]() mutable {
std::this_thread::sleep_for(std::chrono::seconds(1));
resolve(N * 2);
});
thr.detach();
});
return promise;
}
TEST(async_test, basic_test) {
Async::Promise<int> p1(
[](Async::Resolver& resolv, Async::Rejection& reject) {
resolv(10);
});
ASSERT_TRUE(p1.isFulfilled());
int val { 0 };
p1.then([&](int v) { val = v; }, Async::NoExcept);
ASSERT_EQ(val, 10);
Async::Promise<int> p2 = doAsync(10);
p2.then([](int result) { ASSERT_EQ(result, 20); },
Async::NoExcept);
std::this_thread::sleep_for(std::chrono::seconds(2));
Async::Promise<int> p3(
[](Async::Resolver& resolv, Async::Rejection& reject) {
reject(std::runtime_error("Because I decided"));
});
ASSERT_TRUE(p3.isRejected());
p3.then([](int) { ASSERT_TRUE(false); }, [](std::exception_ptr eptr) {
ASSERT_THROW(std::rethrow_exception(eptr), std::runtime_error);
});
}
TEST(async_test, chain_test) {
Async::Promise<int> p1(
[](Async::Resolver& resolve, Async::Rejection& reject) {
resolve(10);
});
p1
.then([](int result) { return result * 2; }, Async::NoExcept)
->chain()
.then([](int result) { std::cout << "Result = " << result << std::endl; },
Async::NoExcept);
Async::Promise<int> p2(
[](Async::Resolver& resolve, Async::Rejection& reject) {
resolve(10);
});
#if 0
p2
.then([](int result) { return result * 2.0; }, Async::IgnoreException)
->chain()
.then([](double result) { std::cout << "Result = " << result << std::endl; },
Async::IgnoreException);
#endif
}
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