Commit d7bba510 authored by jcastro's avatar jcastro

Fix destruction Queue::Entry storage for sentinel entries

Removed allocEntry member function and moved Queue::Entry
allocation to Queue::push, which now constructs and adds the
Entry to the queue.

See #379
parent a8c09c80
...@@ -155,6 +155,22 @@ public: ...@@ -155,6 +155,22 @@ public:
struct Entry { struct Entry {
friend class Queue; friend class Queue;
Entry() :
storage(),
next(nullptr)
{
}
template < class U >
Entry( U&& u ) :
storage(),
next(nullptr)
{
new (&storage) T(std::forward<U>(u));
}
~Entry() = default;
const T& data() const { const T& data() const {
return *reinterpret_cast<const T*>(&storage); return *reinterpret_cast<const T*>(&storage);
} }
...@@ -162,11 +178,6 @@ public: ...@@ -162,11 +178,6 @@ public:
T& data() { T& data() {
return *reinterpret_cast<T*>(&storage); return *reinterpret_cast<T*>(&storage);
} }
~Entry() {
auto *d = reinterpret_cast<T *>(&storage);
d->~T();
}
private: private:
typedef typename std::aligned_storage<sizeof(T), alignof(T)>::type Storage; typedef typename std::aligned_storage<sizeof(T), alignof(T)>::type Storage;
Storage storage; Storage storage;
...@@ -184,18 +195,17 @@ public: ...@@ -184,18 +195,17 @@ public:
} }
virtual ~Queue() { virtual ~Queue() {
while (auto *e = pop()) delete e; while (!empty()) {
Entry* e = pop();
e->data().~T();
delete e;
}
delete tail;
} }
template<typename U> template<typename U>
Entry* allocEntry(U&& u) const { void push( U&& u ) {
auto *e = new Entry; Entry* entry = new Entry(std::forward<U>(u));
new (&e->storage) T(std::forward<U>(u));
return e;
}
void push(Entry *entry) {
entry->next = nullptr;
// @Note: we're using SC atomics here (exchange will issue a full fence), // @Note: we're using SC atomics here (exchange will issue a full fence),
// but I don't think we should bother relaxing them for now // but I don't think we should bother relaxing them for now
auto *prev = head.exchange(entry); auto *prev = head.exchange(entry);
...@@ -215,9 +225,7 @@ public: ...@@ -215,9 +225,7 @@ public:
} }
bool empty() { bool empty() {
auto *res = tail; return head == tail;
auto* next = res->next.load(std::memory_order_acquire);
return next != nullptr;
} }
std::unique_ptr<Entry> popSafe() { std::unique_ptr<Entry> popSafe() {
...@@ -261,8 +269,9 @@ public: ...@@ -261,8 +269,9 @@ public:
return tag; return tag;
} }
void push(Entry* entry) { template < class U >
Queue<T>::push(entry); void push( U&& u ) {
Queue<T>::push(std::forward<U>(u));
if (isBound()) { if (isBound()) {
uint64_t val = 1; uint64_t val = 1;
......
...@@ -46,8 +46,7 @@ public: ...@@ -46,8 +46,7 @@ public:
auto detached = holder.detach(); auto detached = holder.detach();
WriteEntry write(std::move(deferred), detached, flags); WriteEntry write(std::move(deferred), detached, flags);
write.peerFd = fd; write.peerFd = fd;
auto *e = writesQueue.allocEntry(std::move(write)); writesQueue.push(std::move(write));
writesQueue.push(e);
}); });
} }
......
...@@ -184,8 +184,7 @@ Transport::asyncConnect(const std::shared_ptr<Connection>& connection, const str ...@@ -184,8 +184,7 @@ Transport::asyncConnect(const std::shared_ptr<Connection>& connection, const str
{ {
return Async::Promise<void>([=](Async::Resolver& resolve, Async::Rejection& reject) { return Async::Promise<void>([=](Async::Resolver& resolve, Async::Rejection& reject) {
ConnectionEntry entry(std::move(resolve), std::move(reject), connection, address, addr_len); ConnectionEntry entry(std::move(resolve), std::move(reject), connection, address, addr_len);
auto *e = connectionsQueue.allocEntry(std::move(entry)); connectionsQueue.push(std::move(entry));
connectionsQueue.push(e);
}); });
} }
...@@ -199,8 +198,7 @@ Transport::asyncSendRequest( ...@@ -199,8 +198,7 @@ Transport::asyncSendRequest(
auto ctx = context(); auto ctx = context();
RequestEntry req(std::move(resolve), std::move(reject), connection, std::move(timer), std::move(buffer)); RequestEntry req(std::move(resolve), std::move(reject), connection, std::move(timer), std::move(buffer));
if (std::this_thread::get_id() != ctx.thread()) { if (std::this_thread::get_id() != ctx.thread()) {
auto *e = requestsQueue.allocEntry(std::move(req)); requestsQueue.push(std::move(req));
requestsQueue.push(e);
} else { } else {
asyncSendRequestImpl(req); asyncSendRequestImpl(req);
} }
...@@ -480,14 +478,13 @@ Connection::perform( ...@@ -480,14 +478,13 @@ Connection::perform(
Connection::OnDone onDone) { Connection::OnDone onDone) {
return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) { return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) {
if (!isConnected()) { if (!isConnected()) {
auto* entry = requestsQueue.allocEntry( requestsQueue.push(
RequestData( RequestData(
std::move(resolve), std::move(resolve),
std::move(reject), std::move(reject),
request, request,
timeout, timeout,
std::move(onDone))); std::move(onDone)));
requestsQueue.push(entry);
} else { } else {
performImpl(request, timeout, std::move(resolve), std::move(reject), std::move(onDone)); performImpl(request, timeout, std::move(resolve), std::move(reject), std::move(onDone));
} }
......
...@@ -49,8 +49,7 @@ Transport::handleNewPeer(const std::shared_ptr<Tcp::Peer>& peer) { ...@@ -49,8 +49,7 @@ Transport::handleNewPeer(const std::shared_ptr<Tcp::Peer>& peer) {
const bool isInRightThread = std::this_thread::get_id() == ctx.thread(); const bool isInRightThread = std::this_thread::get_id() == ctx.thread();
if (!isInRightThread) { if (!isInRightThread) {
PeerEntry entry(peer); PeerEntry entry(peer);
auto *e = peersQueue.allocEntry(entry); peersQueue.push(std::move(entry));
peersQueue.push(e);
} else { } else {
handlePeer(peer); handlePeer(peer);
} }
...@@ -278,8 +277,7 @@ Transport::armTimerMs( ...@@ -278,8 +277,7 @@ Transport::armTimerMs(
TimerEntry entry(fd, value, std::move(deferred)); TimerEntry entry(fd, value, std::move(deferred));
if (!isInRightThread) { if (!isInRightThread) {
auto *e = timersQueue.allocEntry(std::move(entry)); timersQueue.push(std::move(entry));
timersQueue.push(e);
} else { } else {
armTimerMsImpl(std::move(entry)); armTimerMsImpl(std::move(entry));
} }
......
...@@ -25,8 +25,7 @@ TEST(queue_test, destructor_test) { ...@@ -25,8 +25,7 @@ TEST(queue_test, destructor_test) {
EXPECT_TRUE(queue.empty()); EXPECT_TRUE(queue.empty());
for( int i = 0; i < 5; i++ ) { for( int i = 0; i < 5; i++ ) {
auto* e = queue.allocEntry(Data()); queue.push(Data());
queue.push(e);
} }
// Should call Data::~Data 5 times and not 6 (placeholder entry) // Should call Data::~Data 5 times and not 6 (placeholder entry)
} }
......
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