Commit b1a2e8d1 authored by octal's avatar octal

rusage and timers are now handled in the handler rather than the io layer itself

parent ce01689c
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
#pragma once #pragma once
#include <sys/timerfd.h>
#include <type_traits> #include <type_traits>
#include <stdexcept> #include <stdexcept>
#include <array> #include <array>
...@@ -98,6 +99,8 @@ public: ...@@ -98,6 +99,8 @@ public:
friend class RequestBuilder; friend class RequestBuilder;
friend class Client; friend class Client;
friend class ResponseWriter;
friend class Timeout;
Request(const Request& other) = default; Request(const Request& other) = default;
Request& operator=(const Request& other) = default; Request& operator=(const Request& other) = default;
...@@ -174,23 +177,46 @@ private: ...@@ -174,23 +177,46 @@ private:
}; };
class Handler; class Handler;
class ResponseWrite; class ResponseWriter;
class Timeout { class Timeout {
public: public:
friend class Handler; friend class ResponseWriter;
Timeout(Timeout&& other)
: handler(other.handler)
, request(std::move(other.request))
, peer(std::move(other.peer))
, transport(other.transport)
, armed(other.armed)
, timerFd(other.timerFd)
{
other.timerFd = -1;
}
Timeout& operator=(Timeout&& other) {
handler = other.handler;
request = std::move(other.request);
peer = std::move(other.peer);
transport = other.transport;
armed = other.armed;
timerFd = other.timerFd;
other.timerFd = -1;
}
template<typename Duration> template<typename Duration>
void arm(Duration duration) { void arm(Duration duration) {
Async::Promise<uint64_t> p([=](Async::Resolver& resolve, Async::Rejection& reject) { Async::Promise<uint64_t> p([=](Async::Resolver& resolve, Async::Rejection& reject) {
transport->io()->armTimer(duration, std::move(resolve), std::move(reject)); timerFd = TRY_RET(timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK));
transport->armTimer(timerFd, duration, std::move(resolve), std::move(reject));
}); });
p.then( p.then(
[=](uint64_t numWakeup) { [=](uint64_t numWakeup) {
this->armed = false; this->armed = false;
this->onTimeout(numWakeup); this->onTimeout(numWakeup);
close(timerFd);
}, },
[=](std::exception_ptr exc) { [=](std::exception_ptr exc) {
std::rethrow_exception(exc); std::rethrow_exception(exc);
...@@ -200,8 +226,9 @@ public: ...@@ -200,8 +226,9 @@ public:
} }
void disarm() { void disarm() {
transport->io()->disarmTimer(); if (armed) {
armed = false; transport->disarmTimer(timerFd);
}
} }
bool isArmed() const { bool isArmed() const {
...@@ -209,25 +236,37 @@ public: ...@@ -209,25 +236,37 @@ public:
} }
private: private:
Timeout()
: handler(nullptr)
, transport(nullptr)
{ }
Timeout(Tcp::Transport* transport, Timeout(Tcp::Transport* transport,
Handler* handler, Handler* handler,
const std::shared_ptr<Tcp::Peer>& peer, Request request)
const Request& request)
: transport(transport) : transport(transport)
, handler(handler) , handler(handler)
, peer(peer) , request(std::move(request))
, request(request)
, armed(false) , armed(false)
{ } , timerFd(-1)
{
}
template<typename Ptr>
void associatePeer(const Ptr& ptr) {
peer = ptr;
}
void onTimeout(uint64_t numWakeup); void onTimeout(uint64_t numWakeup);
Handler* handler; Handler* handler;
std::weak_ptr<Tcp::Peer> peer;
Request request; Request request;
std::weak_ptr<Tcp::Peer> peer;
Tcp::Transport* transport; Tcp::Transport* transport;
bool armed; bool armed;
Fd timerFd;
}; };
class ResponseStream : public Message { class ResponseStream : public Message {
...@@ -239,6 +278,7 @@ public: ...@@ -239,6 +278,7 @@ public:
, peer_(std::move(other.peer_)) , peer_(std::move(other.peer_))
, buf_(std::move(other.buf_)) , buf_(std::move(other.buf_))
, transport_(other.transport_) , transport_(other.transport_)
, timeout_(std::move(other.timeout_))
{ } { }
ResponseStream& operator=(ResponseStream&& other) { ResponseStream& operator=(ResponseStream&& other) {
...@@ -246,6 +286,7 @@ public: ...@@ -246,6 +286,7 @@ public:
peer_ = std::move(other.peer_); peer_ = std::move(other.peer_);
buf_ = std::move(other.buf_); buf_ = std::move(other.buf_);
transport_ = other.transport_; transport_ = other.transport_;
timeout_ = std::move(other.timeout_);
return *this; return *this;
} }
...@@ -274,6 +315,7 @@ private: ...@@ -274,6 +315,7 @@ private:
Message&& other, Message&& other,
std::weak_ptr<Tcp::Peer> peer, std::weak_ptr<Tcp::Peer> peer,
Tcp::Transport* transport, Tcp::Transport* transport,
Timeout timeout,
size_t streamSize); size_t streamSize);
std::shared_ptr<Tcp::Peer> peer() const { std::shared_ptr<Tcp::Peer> peer() const {
...@@ -286,6 +328,7 @@ private: ...@@ -286,6 +328,7 @@ private:
std::weak_ptr<Tcp::Peer> peer_; std::weak_ptr<Tcp::Peer> peer_;
DynamicStreamBuf buf_; DynamicStreamBuf buf_;
Tcp::Transport* transport_; Tcp::Transport* transport_;
Timeout timeout_;
}; };
inline ResponseStream& ends(ResponseStream &stream) { inline ResponseStream& ends(ResponseStream &stream) {
...@@ -303,7 +346,7 @@ ResponseStream& operator<<(ResponseStream& stream, const T& val) { ...@@ -303,7 +346,7 @@ ResponseStream& operator<<(ResponseStream& stream, const T& val) {
Net::Size<T> size; Net::Size<T> size;
std::ostream os(&stream.buf_); std::ostream os(&stream.buf_);
os << size(val) << crlf; os << std::hex << size(val) << crlf;
os << val << crlf; os << val << crlf;
return stream; return stream;
...@@ -379,12 +422,14 @@ public: ...@@ -379,12 +422,14 @@ public:
, peer_(other.peer_) , peer_(other.peer_)
, buf_(std::move(other.buf_)) , buf_(std::move(other.buf_))
, transport_(other.transport_) , transport_(other.transport_)
, timeout_(std::move(other.timeout_))
{ } { }
ResponseWriter& operator=(ResponseWriter&& other) { ResponseWriter& operator=(ResponseWriter&& other) {
Response::operator=(std::move(other)); Response::operator=(std::move(other));
peer_ = std::move(other.peer_); peer_ = std::move(other.peer_);
transport_ = other.transport_; transport_ = other.transport_;
buf_ = std::move(other.buf_); buf_ = std::move(other.buf_);
timeout_ = std::move(other.timeout_);
return *this; return *this;
} }
...@@ -441,7 +486,8 @@ public: ...@@ -441,7 +486,8 @@ public:
ResponseStream stream(Code code, size_t streamSize = DefaultStreamSize) { ResponseStream stream(Code code, size_t streamSize = DefaultStreamSize) {
code_ = code; code_ = code;
return ResponseStream(std::move(*this), peer_, transport_, streamSize); return ResponseStream(
std::move(*this), peer_, transport_, std::move(timeout_), streamSize);
} }
// Unsafe API // Unsafe API
...@@ -454,6 +500,15 @@ public: ...@@ -454,6 +500,15 @@ public:
throw std::domain_error("Unimplemented"); throw std::domain_error("Unimplemented");
} }
template<typename Duration>
void timeoutAfter(Duration duration) {
timeout_.arm(duration);
}
Timeout& timeout() {
return timeout_;
}
private: private:
ResponseWriter() ResponseWriter()
: Response() : Response()
...@@ -467,6 +522,13 @@ private: ...@@ -467,6 +522,13 @@ private:
, transport_(transport) , transport_(transport)
{ } { }
ResponseWriter(Tcp::Transport* transport, Request request, Handler* handler)
: Response()
, buf_(DefaultStreamSize)
, transport_(transport)
, timeout_(transport, handler, std::move(request))
{ }
std::shared_ptr<Tcp::Peer> peer() const { std::shared_ptr<Tcp::Peer> peer() const {
if (peer_.expired()) if (peer_.expired())
throw std::runtime_error("Write failed: Broken pipe"); throw std::runtime_error("Write failed: Broken pipe");
...@@ -480,6 +542,7 @@ private: ...@@ -480,6 +542,7 @@ private:
throw std::runtime_error("A peer was already associated to the response"); throw std::runtime_error("A peer was already associated to the response");
peer_ = peer; peer_ = peer;
timeout_.associatePeer(peer_);
} }
Async::Promise<ssize_t> putOnWire(const char* data, size_t len); Async::Promise<ssize_t> putOnWire(const char* data, size_t len);
...@@ -487,6 +550,7 @@ private: ...@@ -487,6 +550,7 @@ private:
std::weak_ptr<Tcp::Peer> peer_; std::weak_ptr<Tcp::Peer> peer_;
DynamicStreamBuf buf_; DynamicStreamBuf buf_;
Tcp::Transport *transport_; Tcp::Transport *transport_;
Timeout timeout_;
}; };
Async::Promise<ssize_t> serveFile( Async::Promise<ssize_t> serveFile(
...@@ -668,7 +732,7 @@ public: ...@@ -668,7 +732,7 @@ public:
void onConnection(const std::shared_ptr<Tcp::Peer>& peer); void onConnection(const std::shared_ptr<Tcp::Peer>& peer);
void onDisconnection(const std::shared_ptr<Tcp::Peer>& peer); void onDisconnection(const std::shared_ptr<Tcp::Peer>& peer);
virtual void onRequest(const Request& request, ResponseWriter response, Timeout timeout) = 0; virtual void onRequest(const Request& request, ResponseWriter response) = 0;
virtual void onTimeout(const Request& request, ResponseWriter response); virtual void onTimeout(const Request& request, ResponseWriter response);
......
...@@ -6,7 +6,6 @@ ...@@ -6,7 +6,6 @@
#pragma once #pragma once
#include "mailbox.h"
#include "flags.h" #include "flags.h"
#include "os.h" #include "os.h"
#include "async.h" #include "async.h"
...@@ -95,10 +94,6 @@ namespace Io { ...@@ -95,10 +94,6 @@ namespace Io {
class Service { class Service {
public: public:
PollableMailbox<Message> mailbox;
Service();
void registerFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level); 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 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 modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level);
...@@ -114,50 +109,13 @@ namespace Io { ...@@ -114,50 +109,13 @@ namespace Io {
std::thread::id thread() const { return thisId; } std::thread::id thread() const { return thisId; }
std::shared_ptr<Handler> handler() const { return handler_; } std::shared_ptr<Handler> handler() const { return handler_; }
Async::Promise<rusage> load() {
return Async::Promise<rusage>([=](Async::Resolver& resolve, Async::Rejection& reject) {
load_ = Some(Async::Holder(std::move(resolve), std::move(reject)));
notifier.notify();
});
}
template<typename Duration>
void armTimer(Duration timeout, Async::Resolver resolve, Async::Rejection reject) {
armTimerMs(std::chrono::duration_cast<std::chrono::milliseconds>(timeout),
std::move(resolve),
std::move(reject));
}
void disarmTimer();
private: private:
struct Timer {
Timer(std::chrono::milliseconds value,
Async::Resolver resolve,
Async::Rejection reject)
: value(value)
, resolve(std::move(resolve))
, reject(std::move(reject))
{ }
std::chrono::milliseconds value;
Async::Resolver resolve;
Async::Rejection reject;
};
void
armTimerMs(std::chrono::milliseconds value, Async::Resolver resolver, Async::Rejection reject);
void handleNotify();
void handleTimeout();
Fd timerFd;
std::thread::id thisId; std::thread::id thisId;
std::shared_ptr<Handler> handler_; std::shared_ptr<Handler> handler_;
Optional<Async::Holder> load_; std::atomic<bool> shutdown_;
Optional<Timer> timer; NotifyFd shutdownFd;
NotifyFd notifier;
Polling::Epoll poller; Polling::Epoll poller;
}; };
...@@ -190,9 +148,7 @@ namespace Io { ...@@ -190,9 +148,7 @@ namespace Io {
void init(const std::shared_ptr<Handler>& handler); void init(const std::shared_ptr<Handler>& handler);
Async::Promise<rusage> load() const { Async::Promise<rusage> load();
return service_->load();
}
void run(); void run();
void shutdown(); void shutdown();
...@@ -212,6 +168,10 @@ namespace Io { ...@@ -212,6 +168,10 @@ namespace Io {
virtual void onReady(const FdSet& fds) = 0; virtual void onReady(const FdSet& fds) = 0;
virtual void registerPoller(Polling::Epoll& poller) { } virtual void registerPoller(Polling::Epoll& poller) { }
virtual Async::Promise<rusage> load() {
return Async::Promise<rusage>::rejected(std::runtime_error("Unimplemented"));
}
Service* io() const { Service* io() const {
return io_; return io_;
} }
......
...@@ -135,8 +135,7 @@ namespace Private { ...@@ -135,8 +135,7 @@ namespace Private {
void onRequest( void onRequest(
const Http::Request& req, const Http::Request& req,
Http::ResponseWriter response, Http::ResponseWriter response);
Http::Timeout timeout);
private: private:
std::shared_ptr<Net::Tcp::Handler> clone() const { std::shared_ptr<Net::Tcp::Handler> clone() const {
......
...@@ -82,7 +82,7 @@ public: ...@@ -82,7 +82,7 @@ public:
: size(0) : size(0)
{ {
memset(bytes, 0, N); memset(bytes, 0, N);
Base::setg(bytes, bytes, bytes); Base::setg(bytes, bytes, bytes + N);
} }
template<size_t M> template<size_t M>
...@@ -99,7 +99,14 @@ public: ...@@ -99,7 +99,14 @@ public:
} }
memcpy(bytes + size, data, len); memcpy(bytes + size, data, len);
Base::setg(bytes, Base::gptr(), bytes + size + len); CharT *cur = nullptr;
if (this->gptr()) {
cur = this->gptr();
} else {
cur = bytes + size;
}
Base::setg(bytes, cur, bytes + size + len);
size += len; size += len;
return true; return true;
......
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
#pragma once #pragma once
#include "io.h" #include "io.h"
#include "mailbox.h"
#include "optional.h" #include "optional.h"
#include "async.h" #include "async.h"
#include "stream.h" #include "stream.h"
...@@ -33,7 +34,8 @@ public: ...@@ -33,7 +34,8 @@ public:
Async::Promise<ssize_t> asyncWrite(Fd fd, const Buf& buffer, int flags = 0) { Async::Promise<ssize_t> asyncWrite(Fd fd, const Buf& buffer, int flags = 0) {
// If the I/O operation has been initiated from an other thread, we queue it and we'll process // If the I/O operation has been initiated from an other thread, we queue it and we'll process
// it in our own thread so that we make sure that every I/O operation happens in the right thread // it in our own thread so that we make sure that every I/O operation happens in the right thread
if (std::this_thread::get_id() != io()->thread()) { const bool isInRightThread = std::this_thread::get_id() == io()->thread();
if (!isInRightThread) {
return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) { return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
BufferHolder holder(buffer); BufferHolder holder(buffer);
auto detached = holder.detach(); auto detached = holder.detach();
...@@ -56,6 +58,24 @@ public: ...@@ -56,6 +58,24 @@ public:
}); });
} }
Async::Promise<rusage> load() {
return Async::Promise<rusage>([=](Async::Resolver& resolve, Async::Rejection& reject) {
loadRequest_ = Some(Async::Holder(std::move(resolve), std::move(reject)));
notifier.notify();
});
}
template<typename Duration>
void armTimer(Fd fd, Duration timeout, Async::Resolver resolve, Async::Rejection reject) {
armTimerMs(
fd, std::chrono::duration_cast<std::chrono::milliseconds>(timeout),
std::move(resolve), std::move(reject));
}
void disarmTimer(Fd fd);
std::shared_ptr<Io::Handler> clone() const; std::shared_ptr<Io::Handler> clone() const;
private: private:
...@@ -146,6 +166,41 @@ private: ...@@ -146,6 +166,41 @@ private:
Fd peerFd; Fd peerFd;
}; };
struct TimerEntry {
TimerEntry(Fd fd, std::chrono::milliseconds value,
Async::Resolver resolve,
Async::Rejection reject)
: fd(fd)
, value(value)
, resolve(std::move(resolve))
, reject(std::move(reject))
{
active.store(true, std::memory_order_relaxed);
}
TimerEntry(TimerEntry&& other)
: fd(other.fd)
, value(other.value)
, resolve(std::move(other.resolve))
, reject(std::move(other.reject))
, active(other.active.load())
{ }
void disable() {
active.store(false, std::memory_order_relaxed);
}
bool isActive() {
return active.load(std::memory_order_relaxed);
}
Fd fd;
std::chrono::milliseconds value;
Async::Resolver resolve;
Async::Rejection reject;
std::atomic<bool> active;
};
mutable std::mutex peersMutex; mutable std::mutex peersMutex;
std::unordered_map<Fd, std::shared_ptr<Peer>> peers; std::unordered_map<Fd, std::shared_ptr<Peer>> peers;
/* @Incomplete: this should be a std::dequeue. /* @Incomplete: this should be a std::dequeue.
...@@ -153,14 +208,32 @@ private: ...@@ -153,14 +208,32 @@ private:
yet and some writes are still on-hold, writes should queue-up so that when the yet and some writes are still on-hold, writes should queue-up so that when the
fd becomes ready again, we can write everything fd becomes ready again, we can write everything
*/ */
std::unordered_map<Fd, OnHoldWrite> toWrite;
PollableQueue<OnHoldWrite> writesQueue; PollableQueue<OnHoldWrite> writesQueue;
std::unordered_map<Fd, OnHoldWrite> toWrite;
PollableQueue<TimerEntry> timersQueue;
std::unordered_map<Fd, TimerEntry> timers;
Optional<Async::Holder> loadRequest_;
NotifyFd notifier;
std::shared_ptr<Tcp::Handler> handler_; std::shared_ptr<Tcp::Handler> handler_;
bool isPeerFd(Fd fd) const;
bool isTimerFd(Fd fd) const;
bool isPeerFd(Polling::Tag tag) const;
bool isTimerFd(Polling::Tag tag) const;
std::shared_ptr<Peer>& getPeer(Fd fd); std::shared_ptr<Peer>& getPeer(Fd fd);
std::shared_ptr<Peer>& getPeer(Polling::Tag tag); std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
void
armTimerMs(Fd fd,
std::chrono::milliseconds value,
Async::Resolver resolver, Async::Rejection reject);
void armTimerMsImpl(TimerEntry entry);
void asyncWriteImpl(Fd fd, OnHoldWrite& entry, WriteStatus status = FirstTry); void asyncWriteImpl(Fd fd, OnHoldWrite& entry, WriteStatus status = FirstTry);
void asyncWriteImpl( void asyncWriteImpl(
Fd fd, int flags, const BufferHolder& buffer, Fd fd, int flags, const BufferHolder& buffer,
...@@ -170,6 +243,9 @@ private: ...@@ -170,6 +243,9 @@ private:
void handlePeerDisconnection(const std::shared_ptr<Peer>& peer); void handlePeerDisconnection(const std::shared_ptr<Peer>& peer);
void handleIncoming(const std::shared_ptr<Peer>& peer); void handleIncoming(const std::shared_ptr<Peer>& peer);
void handleWriteQueue(); void handleWriteQueue();
void handleTimerQueue();
void handleNotify();
void handleTimer(TimerEntry entry);
}; };
} // namespace Tcp } // namespace Tcp
......
...@@ -38,15 +38,15 @@ class MyHandler : public Net::Http::Handler { ...@@ -38,15 +38,15 @@ class MyHandler : public Net::Http::Handler {
void onRequest( void onRequest(
const Net::Http::Request& req, const Net::Http::Request& req,
Net::Http::ResponseWriter response, Net::Http::ResponseWriter response) {
Net::Http::Timeout timeout) {
if (req.resource() == "/ping") { if (req.resource() == "/ping") {
if (req.method() == Net::Http::Method::Get) { if (req.method() == Net::Http::Method::Get) {
using namespace Net::Http; using namespace Net::Http;
timeout.arm(std::chrono::seconds(2)); response.timeoutAfter(std::chrono::seconds(2));
//timeout.arm(std::chrono::seconds(2));
auto query = req.query(); auto query = req.query();
if (query.has("chunked")) { if (query.has("chunked")) {
...@@ -79,7 +79,7 @@ class MyHandler : public Net::Http::Handler { ...@@ -79,7 +79,7 @@ class MyHandler : public Net::Http::Handler {
throw std::runtime_error("Exception thrown in the handler"); throw std::runtime_error("Exception thrown in the handler");
} }
else if (req.resource() == "/timeout") { else if (req.resource() == "/timeout") {
timeout.arm(std::chrono::seconds(5)); response.timeoutAfter(std::chrono::seconds(2));
} }
#if 0 #if 0
else if (req.resource() == "/async") { else if (req.resource() == "/async") {
...@@ -288,7 +288,6 @@ private: ...@@ -288,7 +288,6 @@ private:
}; };
#endif #endif
/*
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
Net::Port port(9080); Net::Port port(9080);
...@@ -324,7 +323,7 @@ int main(int argc, char *argv[]) { ...@@ -324,7 +323,7 @@ int main(int argc, char *argv[]) {
LoadMonitor monitor(server); LoadMonitor monitor(server);
monitor.setInterval(std::chrono::seconds(1)); monitor.setInterval(std::chrono::seconds(1));
monitor.start(); //monitor.start();
server->serve(); server->serve();
...@@ -335,9 +334,8 @@ int main(int argc, char *argv[]) { ...@@ -335,9 +334,8 @@ int main(int argc, char *argv[]) {
server->shutdown(); server->shutdown();
monitor.shutdown(); monitor.shutdown();
} }
*/
#if 1 #if 0
int main() { int main() {
Net::Http::Client client("http://supnetwork.org:9080"); Net::Http::Client client("http://supnetwork.org:9080");
auto opts = Net::Http::Client::options() auto opts = Net::Http::Client::options()
......
...@@ -592,11 +592,13 @@ ResponseStream::ResponseStream( ...@@ -592,11 +592,13 @@ ResponseStream::ResponseStream(
Message&& other, Message&& other,
std::weak_ptr<Tcp::Peer> peer, std::weak_ptr<Tcp::Peer> peer,
Tcp::Transport* transport, Tcp::Transport* transport,
Timeout timeout,
size_t streamSize) size_t streamSize)
: Message(std::move(other)) : Message(std::move(other))
, peer_(std::move(peer)) , peer_(std::move(peer))
, buf_(streamSize) , buf_(streamSize)
, transport_(transport) , transport_(transport)
, timeout_(std::move(timeout))
{ {
if (!writeStatusLine(code_, buf_)) if (!writeStatusLine(code_, buf_))
throw Error("Response exceeded buffer size"); throw Error("Response exceeded buffer size");
...@@ -615,7 +617,7 @@ ResponseStream::ResponseStream( ...@@ -615,7 +617,7 @@ ResponseStream::ResponseStream(
void void
ResponseStream::flush() { ResponseStream::flush() {
transport_->io()->disarmTimer(); timeout_.disarm();
auto buf = buf_.buffer(); auto buf = buf_.buffer();
auto fd = peer()->fd(); auto fd = peer()->fd();
...@@ -665,7 +667,7 @@ ResponseWriter::putOnWire(const char* data, size_t len) ...@@ -665,7 +667,7 @@ ResponseWriter::putOnWire(const char* data, size_t len)
auto buffer = buf_.buffer(); auto buffer = buf_.buffer();
transport_->io()->disarmTimer(); timeout_.disarm();
#undef OUT #undef OUT
...@@ -756,15 +758,13 @@ Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer ...@@ -756,15 +758,13 @@ Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer
auto state = parser.parse(); auto state = parser.parse();
if (state == Private::State::Done) { if (state == Private::State::Done) {
ResponseWriter response(transport()); ResponseWriter response(transport(), parser.request, this);
response.associatePeer(peer); response.associatePeer(peer);
Timeout timeout(transport(), this, peer, parser.request);
#ifdef LIBSTDCPP_SMARTPTR_LOCK_FIXME #ifdef LIBSTDCPP_SMARTPTR_LOCK_FIXME
parser.request.associatePeer(peer); parser.request.associatePeer(peer);
#endif #endif
onRequest(parser.request, std::move(response), std::move(timeout)); onRequest(parser.request, std::move(response));
parser.reset(); parser.reset();
} }
} catch (const HttpError &err) { } catch (const HttpError &err) {
......
...@@ -12,29 +12,6 @@ ...@@ -12,29 +12,6 @@
namespace Io { namespace Io {
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());
}
Service::Service()
{
notifier.bind(poller);
}
void void
Service::registerFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode) { Service::registerFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode) {
poller.addFd(fd, interest, Polling::Tag(fd), mode); poller.addFd(fd, interest, Polling::Tag(fd), mode);
...@@ -75,7 +52,8 @@ Service::init(const std::shared_ptr<Handler>& handler) { ...@@ -75,7 +52,8 @@ Service::init(const std::shared_ptr<Handler>& handler) {
void void
Service::shutdown() { Service::shutdown() {
mailbox.post(new ShutdownMessage()); shutdown_.store(true);
shutdownFd.notify();
} }
void void
...@@ -83,10 +61,7 @@ Service::run() { ...@@ -83,10 +61,7 @@ Service::run() {
if (!handler_) if (!handler_)
throw std::runtime_error("You need to set a handler before running an io service"); throw std::runtime_error("You need to set a handler before running an io service");
timerFd = TRY_RET(timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK)); shutdownFd.bind(poller);
poller.addFd(timerFd, Polling::NotifyOn::Read, Polling::Tag(timerFd));
mailbox.bind(poller);
thisId = std::this_thread::get_id(); thisId = std::this_thread::get_id();
...@@ -103,122 +78,14 @@ Service::run() { ...@@ -103,122 +78,14 @@ Service::run() {
timeout = std::chrono::milliseconds(-1); timeout = std::chrono::milliseconds(-1);
break; break;
default: default:
std::vector<Polling::Event> evs; if (shutdown_) return;
for (auto& event: events) {
if (event.tag == mailbox.tag()) { FdSet set(std::move(events));
std::unique_ptr<Message> msg(mailbox.clear());
if (msg->type() == Message::Type::Shutdown) {
return;
}
}
else if (event.tag == notifier.tag()) {
handleNotify();
}
else {
if (event.flags.hasFlag(Polling::NotifyOn::Read)) {
auto fd = event.tag.value();
if (fd == timerFd) {
handleTimeout();
continue;
}
}
evs.push_back(std::move(event));
}
}
FdSet set(std::move(evs));
handler_->onReady(set); handler_->onReady(set);
} }
} }
} }
void
Service::armTimerMs(
std::chrono::milliseconds value,
Async::Resolver resolve, Async::Rejection reject)
{
itimerspec spec;
spec.it_interval.tv_sec = 0;
spec.it_interval.tv_nsec = 0;
if (value.count() < 1000) {
spec.it_value.tv_sec = 0;
spec.it_value.tv_nsec
= std::chrono::duration_cast<std::chrono::nanoseconds>(value).count();
} else {
spec.it_value.tv_sec
= std::chrono::duration_cast<std::chrono::seconds>(value).count();
spec.it_value.tv_nsec = 0;
}
int res = timerfd_settime(timerFd, 0, &spec, 0);
if (res == -1) {
reject(Net::Error::system("Could not set timer time"));
return;
}
timer = Some(Timer(value, std::move(resolve), std::move(reject)));
}
void
Service::disarmTimer()
{
if (!timer.isEmpty()) {
itimerspec spec;
spec.it_value.tv_sec = spec.it_value.tv_nsec = 0;
spec.it_interval.tv_sec = spec.it_interval.tv_nsec = 0;
int res = timerfd_settime(timerFd, 0, &spec, 0);
if (res == -1)
throw Net::Error::system("Could not set timer time");
timer = None();
}
}
void
Service::handleNotify() {
optionally_do(load_, [&](const Async::Holder& async) {
while (this->notifier.tryRead()) ;
rusage now;
auto res = getrusage(RUSAGE_THREAD, &now);
if (res == -1)
async.reject(std::runtime_error("Could not compute usage"));
async.resolve(now);
});
load_ = None();
}
void
Service::handleTimeout() {
optionally_do(timer, [=](const Timer& entry) {
uint64_t numWakeups;
int res = ::read(timerFd, &numWakeups, sizeof numWakeups);
if (res == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK)
return;
else
entry.reject(Net::Error::system("Could not read timerfd"));
} else {
if (res != sizeof(numWakeups)) {
entry.reject(Net::Error("Read invalid number of bytes for timer fd: "
+ std::to_string(timerFd)));
}
else {
entry.resolve(numWakeups);
}
}
});
}
void void
ServiceGroup::init( ServiceGroup::init(
size_t threads, size_t threads,
...@@ -284,6 +151,11 @@ ServiceGroup::Worker::init(const std::shared_ptr<Handler>& handler) { ...@@ -284,6 +151,11 @@ ServiceGroup::Worker::init(const std::shared_ptr<Handler>& handler) {
service_->init(handler); service_->init(handler);
} }
Async::Promise<rusage>
ServiceGroup::Worker::load() {
return service_->handler()->load();
}
void void
ServiceGroup::Worker::run() { ServiceGroup::Worker::run() {
thread_.reset(new std::thread([=]() { thread_.reset(new std::thread([=]() {
...@@ -293,7 +165,7 @@ ServiceGroup::Worker::run() { ...@@ -293,7 +165,7 @@ ServiceGroup::Worker::run() {
void void
ServiceGroup::Worker::shutdown() { ServiceGroup::Worker::shutdown() {
service_->mailbox.post(new ShutdownMessage); service_->shutdown();
} }
} }
...@@ -3,6 +3,7 @@ ...@@ -3,6 +3,7 @@
#include "tcp.h" #include "tcp.h"
#include "os.h" #include "os.h"
#include <sys/sendfile.h> #include <sys/sendfile.h>
#include <sys/timerfd.h>
using namespace Polling; using namespace Polling;
...@@ -28,6 +29,8 @@ Transport::clone() const { ...@@ -28,6 +29,8 @@ Transport::clone() const {
void void
Transport::registerPoller(Polling::Epoll& poller) { Transport::registerPoller(Polling::Epoll& poller) {
writesQueue.bind(poller); writesQueue.bind(poller);
timersQueue.bind(poller);
notifier.bind(poller);
} }
void void
...@@ -50,10 +53,28 @@ Transport::onReady(const Io::FdSet& fds) { ...@@ -50,10 +53,28 @@ Transport::onReady(const Io::FdSet& fds) {
if (entry.getTag() == writesQueue.tag()) { if (entry.getTag() == writesQueue.tag()) {
handleWriteQueue(); handleWriteQueue();
} }
else if (entry.getTag() == timersQueue.tag()) {
handleTimerQueue();
}
else if (entry.getTag() == notifier.tag()) {
handleNotify();
}
else if (entry.isReadable()) { else if (entry.isReadable()) {
auto& peer = getPeer(entry.getTag()); auto tag = entry.getTag();
handleIncoming(peer); if (isPeerFd(tag)) {
auto& peer = getPeer(tag);
handleIncoming(peer);
} else if (isTimerFd(tag)) {
auto it = timers.find(tag.value());
auto& entry = it->second;
handleTimer(std::move(entry));
timers.erase(it);
}
else {
throw std::runtime_error("Unknown fd");
}
} }
else if (entry.isWritable()) { else if (entry.isWritable()) {
auto tag = entry.getTag(); auto tag = entry.getTag();
...@@ -72,6 +93,16 @@ Transport::onReady(const Io::FdSet& fds) { ...@@ -72,6 +93,16 @@ Transport::onReady(const Io::FdSet& fds) {
} }
} }
void
Transport::disarmTimer(Fd fd) {
auto it = timers.find(fd);
if (it == std::end(timers))
throw std::runtime_error("Timer has not been armed");
auto &entry = it->second;
entry.disable();
}
void void
Transport::handleIncoming(const std::shared_ptr<Peer>& peer) { Transport::handleIncoming(const std::shared_ptr<Peer>& peer) {
char buffer[Const::MaxBuffer]; char buffer[Const::MaxBuffer];
...@@ -191,6 +222,53 @@ Transport::asyncWriteImpl( ...@@ -191,6 +222,53 @@ Transport::asyncWriteImpl(
} }
} }
void
Transport::armTimerMs(
Fd fd, std::chrono::milliseconds value,
Async::Resolver resolve, Async::Rejection reject) {
const bool isInRightThread = std::this_thread::get_id() == io()->thread();
TimerEntry entry(fd, value, std::move(resolve), std::move(reject));
if (!isInRightThread) {
auto *e = timersQueue.allocEntry(std::move(entry));
timersQueue.push(e);
} else {
armTimerMsImpl(std::move(entry));
}
}
void
Transport::armTimerMsImpl(TimerEntry entry) {
auto it = timers.find(entry.fd);
if (it != std::end(timers)) {
entry.reject(std::runtime_error("Timer is already armed"));
return;
}
itimerspec spec;
spec.it_interval.tv_sec = 0;
spec.it_interval.tv_nsec = 0;
if (entry.value.count() < 1000) {
spec.it_value.tv_sec = 0;
spec.it_value.tv_nsec
= std::chrono::duration_cast<std::chrono::nanoseconds>(entry.value).count();
} else {
spec.it_value.tv_sec
= std::chrono::duration_cast<std::chrono::seconds>(entry.value).count();
spec.it_value.tv_nsec = 0;
}
int res = timerfd_settime(entry.fd, 0, &spec, 0);
if (res == -1) {
entry.reject(Net::Error::system("Could not set timer time"));
return;
}
io()->registerFdOneShot(entry.fd, NotifyOn::Read, Polling::Mode::Edge);
timers.insert(std::make_pair(entry.fd, std::move(entry)));
}
void void
Transport::handleWriteQueue() { Transport::handleWriteQueue() {
// Let's drain the queue // Let's drain the queue
...@@ -203,6 +281,77 @@ Transport::handleWriteQueue() { ...@@ -203,6 +281,77 @@ Transport::handleWriteQueue() {
} }
} }
void
Transport::handleTimerQueue() {
for (;;) {
std::unique_ptr<PollableQueue<TimerEntry>::Entry> entry(timersQueue.pop());
if (!entry) break;
auto &timer = entry->data();
armTimerMsImpl(std::move(timer));
}
}
void
Transport::handleNotify() {
optionally_do(loadRequest_, [&](const Async::Holder& async) {
while (this->notifier.tryRead()) ;
rusage now;
auto res = getrusage(RUSAGE_THREAD, &now);
if (res == -1)
async.reject(std::runtime_error("Could not compute usage"));
async.resolve(now);
});
loadRequest_ = None();
}
void
Transport::handleTimer(TimerEntry entry) {
if (entry.isActive()) {
uint64_t numWakeups;
int res = ::read(entry.fd, &numWakeups, sizeof numWakeups);
if (res == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK)
return;
else
entry.reject(Net::Error::system("Could not read timerfd"));
} else {
if (res != sizeof(numWakeups)) {
entry.reject(Net::Error("Read invalid number of bytes for timer fd: "
+ std::to_string(entry.fd)));
}
else {
entry.resolve(numWakeups);
}
}
}
}
bool
Transport::isPeerFd(Fd fd) const {
std::unique_lock<std::mutex> guard(peersMutex);
return peers.find(fd) != std::end(peers);
}
bool
Transport::isTimerFd(Fd fd) const {
return timers.find(fd) != std::end(timers);
}
bool
Transport::isPeerFd(Polling::Tag tag) const {
return isPeerFd(tag.value());
}
bool
Transport::isTimerFd(Polling::Tag tag) const {
return isTimerFd(tag.value());
}
std::shared_ptr<Peer>& std::shared_ptr<Peer>&
Transport::getPeer(Fd fd) Transport::getPeer(Fd fd)
{ {
......
...@@ -218,8 +218,7 @@ HttpHandler::HttpHandler( ...@@ -218,8 +218,7 @@ HttpHandler::HttpHandler(
void void
HttpHandler::onRequest( HttpHandler::onRequest(
const Http::Request& req, const Http::Request& req,
Http::ResponseWriter response, Http::ResponseWriter response)
Http::Timeout timeout)
{ {
auto& r = routes[req.method()]; auto& r = routes[req.method()];
for (const auto& route: r) { for (const auto& route: r) {
......
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