X-Git-Url: http://plrg.eecs.uci.edu/git/?p=folly.git;a=blobdiff_plain;f=folly%2Fstress-test%2Fstress-sequential-folly-queue.cpp;h=d3c3906a5dcdb16213e4307ebc959a0dc599f696;hp=eade97fa420700246f327fdb7fd308bc81020d97;hb=2d16fa08610b9c37dc13cfce06561ff0f57de7f4;hpb=9eadcfcdb4d766a4a10b3828d4a574c71a4ce70f diff --git a/folly/stress-test/stress-sequential-folly-queue.cpp b/folly/stress-test/stress-sequential-folly-queue.cpp index eade97fa..d3c3906a 100644 --- a/folly/stress-test/stress-sequential-folly-queue.cpp +++ b/folly/stress-test/stress-sequential-folly-queue.cpp @@ -1,4 +1,7 @@ #include +#include +#include +#include #include #include @@ -8,39 +11,104 @@ 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 USPSCQueue; typedef folly::UMPSCQueue UMPSCQueue; typedef folly::USPMCQueue USPMCQueue; typedef folly::UMPMCQueue 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 DSPSCQueue; +typedef folly::DMPSCQueue DMPSCQueue; +typedef folly::DSPMCQueue DSPMCQueue; +typedef folly::DMPMCQueue DMPMCQueue; + +// AtomicLinkedList +size_t kAtomicLinkedListSize = 50000; +size_t kAtomicLinkedListPassCount = 10000; +const char* kAtomicLinkedListBenchmarkName = "FollyAtomicLinkedList"; +typedef folly::AtomicLinkedList AtomicLinkedList; + +// MPMC Queue (linearizable) +size_t kMPMCQueueEnqueueStride = 10000; +size_t kMPMCQueueCapacity = 50000; +size_t kMPMCQueueEnqueueCount = 500000000; +const char* kMPMCQueueBenchmarkName = "FollyMPMCQueue"; +typedef folly::MPMCQueue 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 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(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(dur); + std::cout << "[ OK ] " << kTestName << "." << kAtomicLinkedListBenchmarkName + << " (" << milisecs.count() << " ms)" << std::endl; } template -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 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++); } @@ -61,18 +129,118 @@ void run_queue(size_t enqueue_count, const char* bench_name) { << " != " << 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(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 +void run_without_initial_capacity(size_t enqueue_count, const char* bench_name, + size_t enqueue_stride) { + std::unique_ptr q(new Queue()); + run_queue(q.get(), enqueue_count, bench_name, enqueue_stride); +} + +template +void run_with_initial_capacity(size_t queue_capacity, size_t enqueue_count, + const char* bench_name, size_t enqueue_stride) { + std::unique_ptr q(new Queue(queue_capacity)); + run_queue(q.get(), enqueue_count, bench_name, enqueue_stride); +} + int main() { - run_queue(kUSPSCQueueEnqueueCount, kUSPSCQueueBenchmarkName); - run_queue(kUMPSCQueueEnqueueCount, kUMPSCQueueBenchmarkName); - run_queue(kUSPMCQueueEnqueueCount, kUSPMCQueueBenchmarkName); - run_queue(kUMPMCQueueEnqueueCount, kUMPMCQueueBenchmarkName); + // MPMCQueue + run_with_initial_capacity( + kMPMCQueueCapacity , + kMPMCQueueEnqueueCount, + kMPMCQueueBenchmarkName, + kMPMCQueueEnqueueStride); + + // AtomicLinkedList + run_atomic_linkedlist(); + + // UnboundedQueue + run_without_initial_capacity( + kUSPSCQueueEnqueueCount, + kUSPSCQueueBenchmarkName, + kUnboundedQueueEnqueueStride); + run_without_initial_capacity( + kUMPSCQueueEnqueueCount, + kUMPSCQueueBenchmarkName, + kUnboundedQueueEnqueueStride); + run_without_initial_capacity( + kUSPMCQueueEnqueueCount, + kUSPMCQueueBenchmarkName, + kUnboundedQueueEnqueueStride); + run_without_initial_capacity( + kUMPMCQueueEnqueueCount, + kUMPMCQueueBenchmarkName, + kUnboundedQueueEnqueueStride); + + // DynamicBoundedQueue + run_with_initial_capacity( + kDynamicBoundedQueueCapacity , + kDSPSCQueueEnqueueCount, kDSPSCQueueBenchmarkName, + kDynamicBoundedQueueEnqueueStride); + run_with_initial_capacity( + kDynamicBoundedQueueCapacity, + kDMPSCQueueEnqueueCount, + kDMPSCQueueBenchmarkName, + kDynamicBoundedQueueEnqueueStride); + run_with_initial_capacity( + kDynamicBoundedQueueCapacity, + kDSPMCQueueEnqueueCount, + kDSPMCQueueBenchmarkName, + kDynamicBoundedQueueEnqueueStride); + run_with_initial_capacity( + kDynamicBoundedQueueCapacity, + kDMPMCQueueEnqueueCount, + kDMPMCQueueBenchmarkName, + kDynamicBoundedQueueEnqueueStride); return 0; }