#include <folly/concurrency/UnboundedQueue.h>
+#include <folly/concurrency/DynamicBoundedQueue.h>
+#include <folly/AtomicLinkedList.h>
+#include <folly/MPMCQueue.h>
#include <chrono>
#include <cassert>
namespace {
const char* kTestName = "EnqueueDequeue";
-size_t kEnqueueStride = 10000;
+// Unbounded queue
+size_t kUnboundedQueueEnqueueStride = 10000;
size_t kUSPSCQueueEnqueueCount = 1200000000;
-const char* kUSPSCQueueBenchmarkName = "FollyUSPSCQueue";
-
+const char* kUSPSCQueueBenchmarkName = "FollyUnboundedQueue_SPSC";
size_t kUMPSCQueueEnqueueCount = 320000000;
-const char* kUMPSCQueueBenchmarkName = "FollyUMPSCQueue";
-
+const char* kUMPSCQueueBenchmarkName = "FollyUnboundedQueue_MPSC";
size_t kUSPMCQueueEnqueueCount = 320000000;
-const char* kUSPMCQueueBenchmarkName = "FollyUSPMCQueue";
-
+const char* kUSPMCQueueBenchmarkName = "FollyUnboundedQueue_SPMC";
size_t kUMPMCQueueEnqueueCount = 320000000;
-const char* kUMPMCQueueBenchmarkName = "FollyMPMCQueue";
+const char* kUMPMCQueueBenchmarkName = "FollyUnboundedQueue_MPMC";
typedef folly::USPSCQueue<size_t, false> USPSCQueue;
typedef folly::UMPSCQueue<size_t, false> UMPSCQueue;
typedef folly::USPMCQueue<size_t, false> USPMCQueue;
typedef folly::UMPMCQueue<size_t, false> UMPMCQueue;
+// Dynamic bounded queue
+size_t kDynamicBoundedQueueEnqueueStride = 50000;
+size_t kDynamicBoundedQueueCapacity = 200000;
+size_t kDSPSCQueueEnqueueCount = 1200000000;
+const char* kDSPSCQueueBenchmarkName = "FollyDynamicBoundedQueue_SPSC";
+size_t kDMPSCQueueEnqueueCount = 320000000;
+const char* kDMPSCQueueBenchmarkName = "FollyDynamicBoundedQueue_MPSC";
+size_t kDSPMCQueueEnqueueCount = 320000000;
+const char* kDSPMCQueueBenchmarkName = "FollyDynamicBoundedQueue_SPMC";
+size_t kDMPMCQueueEnqueueCount = 320000000;
+const char* kDMPMCQueueBenchmarkName = "FollyDynamicBoundedQueue_MPMC";
+
+typedef folly::DSPSCQueue<size_t, false> DSPSCQueue;
+typedef folly::DMPSCQueue<size_t, false> DMPSCQueue;
+typedef folly::DSPMCQueue<size_t, false> DSPMCQueue;
+typedef folly::DMPMCQueue<size_t, false> DMPMCQueue;
+
+// AtomicLinkedList
+size_t kAtomicLinkedListSize = 50000;
+size_t kAtomicLinkedListPassCount = 10000;
+const char* kAtomicLinkedListBenchmarkName = "FollyAtomicLinkedList";
+typedef folly::AtomicLinkedList<size_t> AtomicLinkedList;
+
+// MPMC Queue (linearizable)
+size_t kMPMCQueueEnqueueStride = 10000;
+size_t kMPMCQueueCapacity = 50000;
+size_t kMPMCQueueEnqueueCount = 500000000;
+const char* kMPMCQueueBenchmarkName = "FollyMPMCQueue";
+typedef folly::MPMCQueue<size_t> MPMCQueue;
+
+}
+
+void run_atomic_linkedlist() {
+ std::cout << "[ RUN ] " << kTestName << "."
+ << kAtomicLinkedListBenchmarkName << std::endl;
+ auto start_time = std::chrono::system_clock::now();
+ for (size_t pass = 0; pass < kAtomicLinkedListPassCount; pass++) {
+ std::unique_ptr<AtomicLinkedList> list(new AtomicLinkedList());
+ bool in_order = true;
+ for (size_t i = 0; i < kAtomicLinkedListSize; i++) {
+ list->insertHead(i);
+ }
+ size_t nSum = 0;
+ auto func = [&nSum] (size_t elem) { nSum += elem; };
+ if (in_order) {
+ list->sweep(func);
+ } else {
+ list->reverseSweep(func);
+ }
+ in_order = !in_order;
+
+ size_t supposed_sum = kAtomicLinkedListSize * (kAtomicLinkedListSize - 1) / 2;
+ if (nSum != supposed_sum) {
+ std::cout << "Sequential linked list pop sum: " << nSum
+ << " != " << supposed_sum << "\n";
+ auto finish_time = std::chrono::system_clock::now();
+ auto dur = finish_time - start_time;
+ auto milisecs = std::chrono::duration_cast<std::chrono::milliseconds>(dur);
+ std::cout << "[ FAILED ] " << kTestName << "." << kAtomicLinkedListBenchmarkName
+ << " (" << milisecs.count() << " ms)" << std::endl;
+ assert(false && "Folly AtomicLinkedList ERROR");
+ }
+ }
+ auto finish_time = std::chrono::system_clock::now();
+ auto dur = finish_time - start_time;
+ auto milisecs = std::chrono::duration_cast<std::chrono::milliseconds>(dur);
+ std::cout << "[ OK ] " << kTestName << "." << kAtomicLinkedListBenchmarkName
+ << " (" << milisecs.count() << " ms)" << std::endl;
}
template <typename Queue>
-void run_queue(size_t enqueue_count, const char* bench_name) {
+void run_queue(Queue* q, size_t enqueue_count, const char* bench_name,
+ size_t enqueue_stride) {
std::cout << "[ RUN ] " << kTestName << "." << bench_name << std::endl;
auto start_time = std::chrono::system_clock::now();
size_t nNo = 0;
- size_t push_failure = 0;
size_t pop_sum = 0;
- std::unique_ptr<Queue> q(new Queue());
while (nNo < enqueue_count) {
size_t curr_push_count =
- std::min(enqueue_count - nNo, kEnqueueStride);
+ std::min(enqueue_count - nNo, enqueue_stride);
for (size_t i = 0; i < curr_push_count; i++) {
q->enqueue(nNo++);
}
<< " != " << supposed_sum << "\n";
std::cout << "[ FAILED ] " << kTestName << "." << bench_name
<< " (" << milisecs.count() << " ms)" << std::endl;
- assert(false && "Folly unbounded queue ERROR");
+ assert(false && "Folly concurrent queue ERROR");
} else {
std::cout << "[ OK ] " << kTestName << "." << bench_name
<< " (" << milisecs.count() << " ms)" << std::endl;
}
}
+// MPMC Specialization.
+template <>
+void run_queue(MPMCQueue* q, size_t enqueue_count, const char* bench_name,
+ size_t enqueue_stride) {
+ std::cout << "[ RUN ] " << kTestName << "." << bench_name << std::endl;
+ auto start_time = std::chrono::system_clock::now();
+
+ size_t nNo = 0;
+ size_t push_sum = 0;
+ size_t pop_sum = 0;
+ while (nNo < enqueue_count) {
+ size_t curr_push_count =
+ std::min(enqueue_count - nNo, enqueue_stride);
+ for (size_t i = 0; i < curr_push_count; i++) {
+ if (q->write(nNo)) {
+ push_sum += nNo;
+ nNo++;
+ }
+ }
+ size_t res;
+ while (q->read(res)) {
+ pop_sum += res;
+ }
+ }
+
+ auto finish_time = std::chrono::system_clock::now();
+ auto dur = finish_time - start_time;
+ auto milisecs = std::chrono::duration_cast<std::chrono::milliseconds>(dur);
+
+ size_t supposed_sum = enqueue_count * (enqueue_count - 1) / 2;
+ if (pop_sum != supposed_sum) {
+ std::cout << "Sequential queue pop sum: " << pop_sum
+ << " != " << supposed_sum << "\n";
+ std::cout << "[ FAILED ] " << kTestName << "." << bench_name
+ << " (" << milisecs.count() << " ms)" << std::endl;
+ assert(false && "Folly concurrent queue ERROR");
+ } else {
+ std::cout << "[ OK ] " << kTestName << "." << bench_name
+ << " (" << milisecs.count() << " ms)" << std::endl;
+ }
+}
+
+template <typename Queue>
+void run_without_initial_capacity(size_t enqueue_count, const char* bench_name,
+ size_t enqueue_stride) {
+ std::unique_ptr<Queue> q(new Queue());
+ run_queue(q.get(), enqueue_count, bench_name, enqueue_stride);
+}
+
+template <typename Queue>
+void run_with_initial_capacity(size_t queue_capacity, size_t enqueue_count,
+ const char* bench_name, size_t enqueue_stride) {
+ std::unique_ptr<Queue> q(new Queue(queue_capacity));
+ run_queue(q.get(), enqueue_count, bench_name, enqueue_stride);
+}
+
int main() {
- run_queue<USPSCQueue>(kUSPSCQueueEnqueueCount, kUSPSCQueueBenchmarkName);
- run_queue<UMPSCQueue>(kUMPSCQueueEnqueueCount, kUMPSCQueueBenchmarkName);
- run_queue<USPMCQueue>(kUSPMCQueueEnqueueCount, kUSPMCQueueBenchmarkName);
- run_queue<UMPMCQueue>(kUMPMCQueueEnqueueCount, kUMPMCQueueBenchmarkName);
+ // MPMCQueue
+ run_with_initial_capacity<MPMCQueue>(
+ kMPMCQueueCapacity ,
+ kMPMCQueueEnqueueCount,
+ kMPMCQueueBenchmarkName,
+ kMPMCQueueEnqueueStride);
+
+ // AtomicLinkedList
+ run_atomic_linkedlist();
+
+ // UnboundedQueue
+ run_without_initial_capacity<USPSCQueue>(
+ kUSPSCQueueEnqueueCount,
+ kUSPSCQueueBenchmarkName,
+ kUnboundedQueueEnqueueStride);
+ run_without_initial_capacity<UMPSCQueue>(
+ kUMPSCQueueEnqueueCount,
+ kUMPSCQueueBenchmarkName,
+ kUnboundedQueueEnqueueStride);
+ run_without_initial_capacity<USPMCQueue>(
+ kUSPMCQueueEnqueueCount,
+ kUSPMCQueueBenchmarkName,
+ kUnboundedQueueEnqueueStride);
+ run_without_initial_capacity<UMPMCQueue>(
+ kUMPMCQueueEnqueueCount,
+ kUMPMCQueueBenchmarkName,
+ kUnboundedQueueEnqueueStride);
+
+ // DynamicBoundedQueue
+ run_with_initial_capacity<DSPSCQueue>(
+ kDynamicBoundedQueueCapacity ,
+ kDSPSCQueueEnqueueCount, kDSPSCQueueBenchmarkName,
+ kDynamicBoundedQueueEnqueueStride);
+ run_with_initial_capacity<DMPSCQueue>(
+ kDynamicBoundedQueueCapacity,
+ kDMPSCQueueEnqueueCount,
+ kDMPSCQueueBenchmarkName,
+ kDynamicBoundedQueueEnqueueStride);
+ run_with_initial_capacity<DSPMCQueue>(
+ kDynamicBoundedQueueCapacity,
+ kDSPMCQueueEnqueueCount,
+ kDSPMCQueueBenchmarkName,
+ kDynamicBoundedQueueEnqueueStride);
+ run_with_initial_capacity<DMPMCQueue>(
+ kDynamicBoundedQueueCapacity,
+ kDMPMCQueueEnqueueCount,
+ kDMPMCQueueBenchmarkName,
+ kDynamicBoundedQueueEnqueueStride);
return 0;
}
projects / folly.git / blobdiff