Commit fca12460 authored by octal's avatar octal

Async: added a typed Deferred<T> interface to avoid storing untyped Resolver...

Async: added a typed Deferred<T> interface to avoid storing untyped Resolver and Rejection objects and thus loosing type information
parent a5ee9e46
......@@ -541,6 +541,8 @@ namespace Async {
template<typename Arg>
bool operator()(Arg&& arg) const {
if (!core_) return false;
typedef typename std::remove_reference<Arg>::type Type;
if (core_->state != State::Pending)
......@@ -565,6 +567,8 @@ namespace Async {
}
bool operator()() const {
if (!core_) return false;
if (core_->state != State::Pending)
throw Error("Attempt to resolve a fulfilled promise");
......@@ -580,6 +584,10 @@ namespace Async {
return true;
}
void clear() {
core_ = nullptr;
}
private:
std::shared_ptr<Private::Core> core_;
};
......@@ -593,6 +601,8 @@ namespace Async {
template<typename Exc>
bool operator()(Exc exc) const {
if (!core_) return false;
if (core_->state != State::Pending)
throw Error("Attempt to reject a fulfilled promise");
......@@ -606,28 +616,90 @@ namespace Async {
return true;
}
void clear() {
core_ = nullptr;
}
private:
std::shared_ptr<Private::Core> core_;
};
struct Holder {
Holder(Resolver resolver, Rejection rejection)
template<typename T>
class Deferred {
public:
Deferred()
: resolver(nullptr)
, rejection(nullptr)
{ }
Deferred(const Deferred& other) = delete;
Deferred& operator=(const Deferred& other) = delete;
Deferred(Deferred&& other) = default;
Deferred& operator=(Deferred&& other) = default;
Deferred(Resolver resolver, Rejection reject)
: resolver(std::move(resolver))
, rejection(std::move(rejection))
, rejection(std::move(reject))
{ }
template<typename Arg>
void resolve(Arg&& arg) const {
resolver(std::forward<Arg>(arg));
template<typename U>
bool resolve(U&& arg) {
typedef typename std::remove_reference<U>::type CleanU;
static_assert(std::is_same<T, CleanU>::value || std::is_convertible<U, T>::value,
"Types mismatch");
return resolver(std::forward<U>(arg));
}
template<typename... Args>
void emplaceResolve(Args&& ...args) {
}
template<typename Exc>
bool reject(Exc exc) {
return rejection(std::move(exc));
}
void resolve() const {
void clear() {
resolver.clear();
rejection.clear();
}
private:
Resolver resolver;
Rejection rejection;
};
template<>
class Deferred<void> {
public:
Deferred()
: resolver(nullptr)
, rejection(nullptr)
{ }
Deferred(const Deferred& other) = delete;
Deferred& operator=(const Deferred& other) = delete;
Deferred(Deferred&& other) = default;
Deferred& operator=(Deferred&& other) = default;
Deferred(Resolver resolver, Rejection reject)
: resolver(std::move(resolver))
, rejection(std::move(reject))
{ }
void resolve() {
resolver();
}
template<typename Exc>
void reject(Exc exc) const {
void reject(Exc exc) {
rejection(std::move(exc));
}
......@@ -636,27 +708,40 @@ namespace Async {
Rejection rejection;
};
static constexpr Private::IgnoreException IgnoreException{};
static constexpr Private::NoExcept NoExcept{};
static constexpr Private::Throw Throw{};
namespace details {
template<typename T, typename Func>
typename std::result_of<Func(Resolver&, Rejection&)>::type
callAsync(Func func, Resolver& resolver, Rejection& rejection) {
func(resolver, rejection);
}
template<typename T, typename Func>
typename std::result_of<Func(Deferred<T>)>::type
callAsync(Func func, Resolver& resolver, Rejection& rejection) {
func(Deferred<T>(std::move(resolver), std::move(rejection)));
}
};
template<typename T>
class Promise : public PromiseBase
{
public:
template<typename U> friend class Promise;
typedef std::function<void (Resolver&, Rejection&)> ResolveFunc;
typedef Private::CoreT<T> Core;
Promise(ResolveFunc func)
template<typename Func>
Promise(Func func)
: core_(std::make_shared<Core>())
, resolver_(core_)
, rejection_(core_)
{
func(resolver_, rejection_);
//func(resolver_, rejection_);
details::callAsync<T>(func, resolver_, rejection_);
}
Promise(const Promise<T>& other) = delete;
......
......@@ -203,9 +203,9 @@ public:
template<typename Duration>
void arm(Duration duration) {
Async::Promise<uint64_t> p([=](Async::Resolver& resolve, Async::Rejection& reject) {
Async::Promise<uint64_t> p([=](Async::Deferred<uint64_t> deferred) {
timerFd = TRY_RET(timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK));
transport->armTimer(timerFd, duration, std::move(resolve), std::move(reject));
transport->armTimer(timerFd, duration, std::move(deferred));
});
p.then(
......
......@@ -36,10 +36,10 @@ public:
// it in our own thread so that we make sure that every I/O operation happens in the right 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::Deferred<ssize_t> deferred) {
BufferHolder holder(buffer);
auto detached = holder.detach();
WriteEntry write(std::move(resolve), std::move(reject), detached, flags);
WriteEntry write(std::move(deferred), detached, flags);
write.peerFd = fd;
auto *e = writesQueue.allocEntry(std::move(write));
writesQueue.push(e);
......@@ -53,24 +53,24 @@ public:
return;
}
asyncWriteImpl(fd, flags, BufferHolder(buffer), std::move(resolve), std::move(reject));
asyncWriteImpl(fd, flags, BufferHolder(buffer), Async::Deferred<ssize_t>(std::move(resolve), std::move(reject)));
});
}
Async::Promise<rusage> load() {
return Async::Promise<rusage>([=](Async::Resolver& resolve, Async::Rejection& reject) {
loadRequest_ = Some(Async::Holder(std::move(resolve), std::move(reject)));
return Async::Promise<rusage>([=](Async::Deferred<rusage> deferred) {
loadRequest_ = std::move(deferred);
notifier.notify();
});
}
template<typename Duration>
void armTimer(Fd fd, Duration timeout, Async::Resolver resolve, Async::Rejection reject) {
void armTimer(Fd fd, Duration timeout, Async::Deferred<uint64_t> deferred) {
armTimerMs(
fd, std::chrono::duration_cast<std::chrono::milliseconds>(timeout),
std::move(resolve), std::move(reject));
std::move(deferred));
}
......@@ -150,17 +150,15 @@ private:
};
struct WriteEntry {
WriteEntry(Async::Resolver resolve, Async::Rejection reject,
WriteEntry(Async::Deferred<ssize_t> deferred,
BufferHolder buffer, int flags = 0)
: resolve(std::move(resolve))
, reject(std::move(reject))
: deferred(std::move(deferred))
, buffer(std::move(buffer))
, flags(flags)
, peerFd(-1)
{ }
Async::Resolver resolve;
Async::Rejection reject;
Async::Deferred<ssize_t> deferred;
BufferHolder buffer;
int flags;
Fd peerFd;
......@@ -168,12 +166,10 @@ private:
struct TimerEntry {
TimerEntry(Fd fd, std::chrono::milliseconds value,
Async::Resolver resolve,
Async::Rejection reject)
Async::Deferred<uint64_t> deferred)
: fd(fd)
, value(value)
, resolve(std::move(resolve))
, reject(std::move(reject))
, deferred(std::move(deferred))
{
active.store(true, std::memory_order_relaxed);
}
......@@ -181,8 +177,7 @@ private:
TimerEntry(TimerEntry&& other)
: fd(other.fd)
, value(other.value)
, resolve(std::move(other.resolve))
, reject(std::move(other.reject))
, deferred(std::move(other.deferred))
, active(other.active.load())
{ }
......@@ -196,8 +191,7 @@ private:
Fd fd;
std::chrono::milliseconds value;
Async::Resolver resolve;
Async::Rejection reject;
Async::Deferred<uint64_t> deferred;
std::atomic<bool> active;
};
......@@ -223,7 +217,7 @@ private:
PollableQueue<PeerEntry> peersQueue;
std::unordered_map<Fd, std::shared_ptr<Peer>> peers;
Optional<Async::Holder> loadRequest_;
Async::Deferred<rusage> loadRequest_;
NotifyFd notifier;
std::shared_ptr<Tcp::Handler> handler_;
......@@ -239,14 +233,14 @@ private:
void
armTimerMs(Fd fd,
std::chrono::milliseconds value,
Async::Resolver resolver, Async::Rejection reject);
Async::Deferred<uint64_t> deferred);
void armTimerMsImpl(TimerEntry entry);
void asyncWriteImpl(Fd fd, WriteEntry& entry, WriteStatus status = FirstTry);
void asyncWriteImpl(
Fd fd, int flags, const BufferHolder& buffer,
Async::Resolver resolve, Async::Rejection reject,
Async::Deferred<ssize_t> deferred,
WriteStatus status = FirstTry);
void handlePeerDisconnection(const std::shared_ptr<Peer>& peer);
......
......@@ -164,13 +164,13 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) {
void
Transport::asyncWriteImpl(Fd fd, Transport::WriteEntry& entry, WriteStatus status) {
asyncWriteImpl(fd, entry.flags, entry.buffer, std::move(entry.resolve), std::move(entry.reject), status);
asyncWriteImpl(fd, entry.flags, entry.buffer, std::move(entry.deferred), status);
}
void
Transport::asyncWriteImpl(
Fd fd, int flags, const BufferHolder& buffer,
Async::Resolver resolve, Async::Rejection reject, WriteStatus status)
Async::Deferred<ssize_t> deferred, WriteStatus status)
{
auto cleanUp = [&]() {
if (buffer.isRaw()) {
......@@ -200,13 +200,13 @@ Transport::asyncWriteImpl(
if (status == FirstTry) {
toWrite.insert(
std::make_pair(fd,
WriteEntry(std::move(resolve), std::move(reject), buffer.detach(totalWritten), flags)));
WriteEntry(std::move(deferred), buffer.detach(totalWritten), flags)));
}
io()->modifyFd(fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
}
else {
cleanUp();
reject(Net::Error::system("Could not write data"));
deferred.reject(Net::Error::system("Could not write data"));
}
break;
}
......@@ -214,7 +214,7 @@ Transport::asyncWriteImpl(
totalWritten += bytesWritten;
if (totalWritten == len) {
cleanUp();
resolve(totalWritten);
deferred.resolve(totalWritten);
break;
}
}
......@@ -224,9 +224,10 @@ Transport::asyncWriteImpl(
void
Transport::armTimerMs(
Fd fd, std::chrono::milliseconds value,
Async::Resolver resolve, Async::Rejection reject) {
Async::Deferred<uint64_t> deferred) {
const bool isInRightThread = std::this_thread::get_id() == io()->thread();
TimerEntry entry(fd, value, std::move(resolve), std::move(reject));
TimerEntry entry(fd, value, std::move(deferred));
if (!isInRightThread) {
auto *e = timersQueue.allocEntry(std::move(entry));
timersQueue.push(e);
......@@ -240,7 +241,7 @@ Transport::armTimerMsImpl(TimerEntry entry) {
auto it = timers.find(entry.fd);
if (it != std::end(timers)) {
entry.reject(std::runtime_error("Timer is already armed"));
entry.deferred.reject(std::runtime_error("Timer is already armed"));
return;
}
......@@ -260,7 +261,7 @@ Transport::armTimerMsImpl(TimerEntry entry) {
int res = timerfd_settime(entry.fd, 0, &spec, 0);
if (res == -1) {
entry.reject(Net::Error::system("Could not set timer time"));
entry.deferred.reject(Net::Error::system("Could not set timer time"));
return;
}
......@@ -315,19 +316,16 @@ Transport::handlePeer(const std::shared_ptr<Peer>& peer) {
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_.reject(std::runtime_error("Could not compute usage"));
loadRequest_ = None();
loadRequest_.resolve(now);
loadRequest_.clear();
}
void
......@@ -339,14 +337,14 @@ Transport::handleTimer(TimerEntry entry) {
if (errno == EAGAIN || errno == EWOULDBLOCK)
return;
else
entry.reject(Net::Error::system("Could not read timerfd"));
entry.deferred.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: "
entry.deferred.reject(Net::Error("Read invalid number of bytes for timer fd: "
+ std::to_string(entry.fd)));
}
else {
entry.resolve(numWakeups);
entry.deferred.resolve(numWakeups);
}
}
}
......
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