Commit 1592db4a authored by Andrii Grynenko's avatar Andrii Grynenko Committed by Facebook GitHub Bot

Use alignas(folly::cacheline_align_v) to improve NotificationQueue perf

Summary: This also splits queue size counter into two to make sure we do fewer atomic operations on the consumer thread.

Differential Revision: D24490151

fbshipit-source-id: 69297a2fd0c5fddcd7355a28fe600b72320e81cd
parent a876ba8b
...@@ -98,8 +98,9 @@ AtomicNotificationQueue::AtomicQueue::~AtomicQueue() { ...@@ -98,8 +98,9 @@ AtomicNotificationQueue::AtomicQueue::~AtomicQueue() {
} }
bool AtomicNotificationQueue::AtomicQueue::push(Task&& value) { bool AtomicNotificationQueue::AtomicQueue::push(Task&& value) {
std::unique_ptr<Queue::Node> node(new Queue::Node(std::move(value))); pushCount_.fetch_add(1, std::memory_order_relaxed);
std::unique_ptr<Queue::Node> node(new Queue::Node(std::move(value)));
auto head = head_.load(std::memory_order_relaxed); auto head = head_.load(std::memory_order_relaxed);
while (true) { while (true) {
node->next = node->next =
...@@ -222,7 +223,9 @@ void AtomicNotificationQueue::setMaxReadAtOnce(uint32_t maxAtOnce) { ...@@ -222,7 +223,9 @@ void AtomicNotificationQueue::setMaxReadAtOnce(uint32_t maxAtOnce) {
} }
int32_t AtomicNotificationQueue::size() const { int32_t AtomicNotificationQueue::size() const {
return queueSize_.load(std::memory_order_relaxed); auto queueSize = atomicQueue_.getPushCount() - taskExecuteCount_;
DCHECK(queueSize >= 0);
return queueSize;
} }
bool AtomicNotificationQueue::empty() const { bool AtomicNotificationQueue::empty() const {
...@@ -230,7 +233,6 @@ bool AtomicNotificationQueue::empty() const { ...@@ -230,7 +233,6 @@ bool AtomicNotificationQueue::empty() const {
} }
void AtomicNotificationQueue::putMessage(Func&& func) { void AtomicNotificationQueue::putMessage(Func&& func) {
queueSize_.fetch_add(1, std::memory_order_relaxed);
if (atomicQueue_.push(Task{std::move(func), RequestContext::saveContext()})) { if (atomicQueue_.push(Task{std::move(func), RequestContext::saveContext()})) {
notifyFd(); notifyFd();
} }
...@@ -283,7 +285,7 @@ bool AtomicNotificationQueue::drive() { ...@@ -283,7 +285,7 @@ bool AtomicNotificationQueue::drive() {
break; break;
} }
} }
queueSize_.fetch_sub(1, std::memory_order_relaxed); ++taskExecuteCount_;
std::move(queue_.front()).execute(); std::move(queue_.front()).execute();
queue_.pop(); queue_.pop();
} }
......
...@@ -174,9 +174,18 @@ class AtomicNotificationQueue : private EventBase::LoopCallback, ...@@ -174,9 +174,18 @@ class AtomicNotificationQueue : private EventBase::LoopCallback,
return successfulArmCount_ - consumerDisarmCount_; return successfulArmCount_ - consumerDisarmCount_;
} }
/*
* Returns how many times push was called.
* Can be called from any thread.
*/
ssize_t getPushCount() const {
return pushCount_.load(std::memory_order_relaxed);
}
private: private:
std::atomic<Queue::Node*> head_{}; alignas(folly::cacheline_align_v) std::atomic<Queue::Node*> head_{};
ssize_t successfulArmCount_{0}; std::atomic<ssize_t> pushCount_{0};
alignas(folly::cacheline_align_v) ssize_t successfulArmCount_{0};
ssize_t consumerDisarmCount_{0}; ssize_t consumerDisarmCount_{0};
static constexpr intptr_t kQueueArmedTag = 1; static constexpr intptr_t kQueueArmedTag = 1;
}; };
...@@ -296,7 +305,7 @@ class AtomicNotificationQueue : private EventBase::LoopCallback, ...@@ -296,7 +305,7 @@ class AtomicNotificationQueue : private EventBase::LoopCallback,
AtomicQueue atomicQueue_; AtomicQueue atomicQueue_;
Queue queue_; Queue queue_;
std::atomic<int32_t> queueSize_{0}; ssize_t taskExecuteCount_{0};
int32_t maxReadAtOnce_{10}; int32_t maxReadAtOnce_{10};
int eventfd_{-1}; int eventfd_{-1};
int pipeFds_[2]{-1, -1}; // to fallback to on older/non-linux systems int pipeFds_[2]{-1, -1}; // to fallback to on older/non-linux systems
......
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