Modify class name and other small changes
diff --git a/src/core/lib/iomgr/executor/mpmcqueue.cc b/src/core/lib/iomgr/executor/mpmcqueue.cc
index 7dc7be8..6deba05 100644
--- a/src/core/lib/iomgr/executor/mpmcqueue.cc
+++ b/src/core/lib/iomgr/executor/mpmcqueue.cc
@@ -29,40 +29,48 @@
#include <string.h>
#include "src/core/lib/debug/stats.h"
+#include "src/core/lib/gprpp/memory.h"
#include "src/core/lib/gprpp/sync.h"
namespace grpc_core {
-DebugOnlyTraceFlag thread_pool_trace(false, "thread_pool_trace");
+DebugOnlyTraceFlag thread_pool(false, "thread_pool_trace");
-inline void* MPMCQueue::PopFront() {
+inline void* InfLenFIFOQueue::PopFront() {
void* result = queue_head_->content;
Node* head_to_remove = queue_head_;
queue_head_ = queue_head_->next;
- count_.Store(count_.Load(MemoryOrder::RELAXED) - 1, MemoryOrder::RELAXED);
- gpr_timespec wait_time =
+ count_.FetchSub(1, MemoryOrder::RELAXED);
+
+ if (GRPC_TRACE_FLAG_ENABLED(thread_pool)) {
+ gpr_timespec wait_time =
gpr_time_sub(gpr_now(GPR_CLOCK_PRECISE), head_to_remove->insert_time);
- gpr_free(head_to_remove);
+ // Updates Stats info
+ stats_.num_completed++;
+ stats_.total_queue_cycles =
+ gpr_time_add(stats_.total_queue_cycles, wait_time);
+ stats_.max_queue_cycles = gpr_time_max(
+ gpr_convert_clock_type(stats_.max_queue_cycles, GPR_TIMESPAN),
+ wait_time);
- // Update Stats info
- stats_.num_completed++;
- stats_.total_queue_cycles =
- gpr_time_add(stats_.total_queue_cycles, wait_time);
- stats_.max_queue_cycles = gpr_time_max(
- gpr_convert_clock_type(stats_.max_queue_cycles, GPR_TIMESPAN), wait_time);
+ if (count_.Load(MemoryOrder::RELAXED) == 0) {
+ stats_.busy_time_cycles =
+ gpr_time_add(stats_.busy_time_cycles,
+ gpr_time_sub(gpr_now(GPR_CLOCK_PRECISE), busy_time));
+ }
- if (count_.Load(MemoryOrder::RELAXED) == 0) {
- stats_.busy_time_cycles =
- gpr_time_add(stats_.busy_time_cycles,
- gpr_time_sub(gpr_now(GPR_CLOCK_PRECISE), busy_time));
+ gpr_log(GPR_INFO,
+ "[InfLenFIFOQueue Get] num_completed: %" PRIu64
+ " total_queue_cycles: %" PRId32 " max_queue_cycles: %" PRId32
+ " busy_time_cycles: %" PRId32,
+ stats_.num_completed, gpr_time_to_millis(stats_.total_queue_cycles),
+ gpr_time_to_millis(stats_.max_queue_cycles),
+ gpr_time_to_millis(stats_.busy_time_cycles));
}
- if (GRPC_TRACE_FLAG_ENABLED(thread_pool_trace)) {
- PrintStats();
- }
-
+ Delete(head_to_remove);
// Singal waiting thread
if (count_.Load(MemoryOrder::RELAXED) > 0 && num_waiters_ > 0) {
wait_nonempty_.Signal();
@@ -71,37 +79,34 @@
return result;
}
-MPMCQueue::MPMCQueue()
- : num_waiters_(0), queue_head_(nullptr), queue_tail_(nullptr) {
- count_.Store(0, MemoryOrder::RELAXED);
-}
+InfLenFIFOQueue::InfLenFIFOQueue()
+ : num_waiters_(0), queue_head_(nullptr), queue_tail_(nullptr) {}
-MPMCQueue::~MPMCQueue() {
+InfLenFIFOQueue::~InfLenFIFOQueue() {
GPR_ASSERT(count_.Load(MemoryOrder::RELAXED) == 0);
- MutexLock l(&mu_);
GPR_ASSERT(num_waiters_ == 0);
}
-void MPMCQueue::Put(void* elem) {
+void InfLenFIFOQueue::Put(void* elem) {
MutexLock l(&mu_);
- Node* new_node = static_cast<Node*>(gpr_malloc(sizeof(Node)));
- new_node->next = nullptr;
- new_node->content = elem;
- new_node->insert_time = gpr_now(GPR_CLOCK_PRECISE);
+ Node* new_node = New<Node>(elem);
if (count_.Load(MemoryOrder::RELAXED) == 0) {
- busy_time = gpr_now(GPR_CLOCK_PRECISE);
+ if (GRPC_TRACE_FLAG_ENABLED(thread_pool)) {
+ busy_time = gpr_now(GPR_CLOCK_PRECISE);
+ }
queue_head_ = queue_tail_ = new_node;
} else {
queue_tail_->next = new_node;
queue_tail_ = queue_tail_->next;
}
- count_.Store(count_.Load(MemoryOrder::RELAXED) + 1, MemoryOrder::RELAXED);
+ count_.FetchAdd(1, MemoryOrder::RELAXED);
- // Update Stats info
- stats_.num_started++;
- if (GRPC_TRACE_FLAG_ENABLED(thread_pool_trace)) {
- PrintStats();
+ // Updates Stats info
+ if (GRPC_TRACE_FLAG_ENABLED(thread_pool)) {
+ stats_.num_started++;
+ gpr_log(GPR_INFO, "[InfLenFIFOQueue Put] num_started: %" PRIu64,
+ stats_.num_started);
}
if (num_waiters_ > 0) {
@@ -109,7 +114,7 @@
}
}
-void* MPMCQueue::Get() {
+void* InfLenFIFOQueue::Get() {
MutexLock l(&mu_);
if (count_.Load(MemoryOrder::RELAXED) == 0) {
num_waiters_++;
@@ -118,20 +123,8 @@
} while (count_.Load(MemoryOrder::RELAXED) == 0);
num_waiters_--;
}
- GPR_ASSERT(count_.Load(MemoryOrder::RELAXED) > 0);
+ GPR_DEBUG_ASSERT(count_.Load(MemoryOrder::RELAXED) > 0);
return PopFront();
}
-void MPMCQueue::PrintStats() {
- gpr_log(GPR_INFO, "STATS INFO:");
- gpr_log(GPR_INFO, "num_started: %" PRIu64, stats_.num_started);
- gpr_log(GPR_INFO, "num_completed: %" PRIu64, stats_.num_completed);
- gpr_log(GPR_INFO, "total_queue_cycles: %" PRId32,
- gpr_time_to_millis(stats_.total_queue_cycles));
- gpr_log(GPR_INFO, "max_queue_cycles: %" PRId32,
- gpr_time_to_millis(stats_.max_queue_cycles));
- gpr_log(GPR_INFO, "busy_time_cycles: %" PRId32,
- gpr_time_to_millis(stats_.busy_time_cycles));
-}
-
} // namespace grpc_core
diff --git a/src/core/lib/iomgr/executor/mpmcqueue.h b/src/core/lib/iomgr/executor/mpmcqueue.h
index aec8dd3..8b59b05 100644
--- a/src/core/lib/iomgr/executor/mpmcqueue.h
+++ b/src/core/lib/iomgr/executor/mpmcqueue.h
@@ -16,8 +16,8 @@
*
*/
-#ifndef GRPC_CORE_LIB_IOMGR_EXECUTOR_MPMCQUEUE_H
-#define GRPC_CORE_LIB_IOMGR_EXECUTOR_MPMCQUEUE_H
+#ifndef GRPC_CORE_LIB_IOMGR_EXECUTOR_INFLENFIFOQUEUE_H
+#define GRPC_CORE_LIB_IOMGR_EXECUTOR_INFLENFIFOQUEUE_H
#include <grpc/support/port_platform.h>
@@ -31,46 +31,47 @@
namespace grpc_core {
-extern DebugOnlyTraceFlag thread_pool_trace;
+extern DebugOnlyTraceFlag thread_pool;
-// Abstract base class of a MPMC queue interface
+// Abstract base class of a Multiple-Producer-Multiple-Consumer(MPMC) queue
+// interface
class MPMCQueueInterface {
public:
virtual ~MPMCQueueInterface() {}
- // Put elem into queue immediately at the end of queue.
+ // Puts elem into queue immediately at the end of queue.
// This might cause to block on full queue depending on implementation.
virtual void Put(void* elem) GRPC_ABSTRACT;
- // Remove the oldest element from the queue and return it.
+ // Removes the oldest element from the queue and return it.
// This might cause to block on empty queue depending on implementation.
virtual void* Get() GRPC_ABSTRACT;
- // Return number of elements in the queue currently
+ // Returns number of elements in the queue currently
virtual int count() const GRPC_ABSTRACT;
GRPC_ABSTRACT_BASE_CLASS
};
-class MPMCQueue : public MPMCQueueInterface {
+class InfLenFIFOQueue : public MPMCQueueInterface {
public:
- // Create a new Multiple-Producer-Multiple-Consumer Queue. The queue created
+ // Creates a new MPMC Queue. The queue created
// will have infinite length.
- MPMCQueue();
+ InfLenFIFOQueue();
- // Release all resources hold by the queue. The queue must be empty, and no
+ // Releases all resources hold by the queue. The queue must be empty, and no
// one waiting on conditional variables.
- ~MPMCQueue();
+ ~InfLenFIFOQueue();
- // Put elem into queue immediately at the end of queue. Since the queue has
+ // Puts elem into queue immediately at the end of queue. Since the queue has
// infinite length, this routine will never block and should never fail.
void Put(void* elem);
- // Remove the oldest element from the queue and return it.
+ // Removes the oldest element from the queue and returns it.
// This routine will cause the thread to block if queue is currently empty.
void* Get();
- // Return number of elements in queue currently.
+ // Returns number of elements in queue currently.
// There might be concurrently add/remove on queue, so count might change
// quickly.
int count() const { return count_.Load(MemoryOrder::RELAXED); }
@@ -78,11 +79,11 @@
GRPC_ABSTRACT_BASE_CLASS
private:
+ // For Internal Use Only.
+ // Removes the oldest element from the queue and returns it. This routine
+ // will NOT check whether queue is empty, and it will NOT acquire mutex.
void* PopFront();
- // Print out Stats. Time measurement are printed in millisecond.
- void PrintStats();
-
struct Node {
Node* next; // Linking
void* content; // Points to actual element
@@ -94,7 +95,9 @@
}
};
- struct Stats { // Stats of queue
+ // Stats of queue. This will only be collect when debug trace mode is on.
+ // All printed stats info will have time measurement in millisecond.
+ struct Stats {
uint64_t num_started; // Number of elements have been added to queue
uint64_t num_completed; // Number of elements have been removed from
// the queue
@@ -120,11 +123,11 @@
Node* queue_head_; // Head of the queue, remove position
Node* queue_tail_; // End of queue, insert position
- Atomic<uint64_t> count_; // Number of elements in queue
+ Atomic<uint64_t> count_{0}; // Number of elements in queue
Stats stats_; // Stats info
gpr_timespec busy_time; // Start time of busy queue
};
} // namespace grpc_core
-#endif /* GRPC_CORE_LIB_IOMGR_EXECUTOR_MPMCQUEUE_H */
+#endif /* GRPC_CORE_LIB_IOMGR_EXECUTOR_INFLENFIFOQUEUE_H */
diff --git a/test/core/iomgr/mpmcqueue_test.cc b/test/core/iomgr/mpmcqueue_test.cc
index 12799d7..ebbf060 100644
--- a/test/core/iomgr/mpmcqueue_test.cc
+++ b/test/core/iomgr/mpmcqueue_test.cc
@@ -29,14 +29,6 @@
#define THREAD_SMALL_ITERATION 100
#define THREAD_LARGE_ITERATION 10000
-static void test_no_op(void) {
- gpr_log(GPR_DEBUG, "test_no_op");
- grpc_core::MPMCQueue mpmcqueue;
- gpr_log(GPR_DEBUG, "Checking count...");
- GPR_ASSERT(mpmcqueue.count() == 0);
- gpr_log(GPR_DEBUG, "Done.");
-}
-
// Testing items for queue
struct WorkItem {
int index;
@@ -45,59 +37,44 @@
WorkItem(int i) : index(i) { done = false; }
};
-static void test_small_queue(void) {
- gpr_log(GPR_DEBUG, "test_small_queue");
- grpc_core::MPMCQueue small_queue;
- for (int i = 0; i < THREAD_SMALL_ITERATION; ++i) {
- small_queue.Put(static_cast<void*>(new WorkItem(i)));
- }
- GPR_ASSERT(small_queue.count() == THREAD_SMALL_ITERATION);
- // Get items out in FIFO order
- for (int i = 0; i < THREAD_SMALL_ITERATION; ++i) {
- WorkItem* item = static_cast<WorkItem*>(small_queue.Get());
- GPR_ASSERT(i == item->index);
- delete item;
- }
- gpr_log(GPR_DEBUG, "Done.");
-}
-
-static void test_get_thd(void* args) {
- grpc_core::MPMCQueue* mpmcqueue = static_cast<grpc_core::MPMCQueue*>(args);
+static void ConsumerThread(void* args) {
+ grpc_core::InfLenFIFOQueue* queue =
+ static_cast<grpc_core::InfLenFIFOQueue*>(args);
// count number of Get() called in this thread
int count = 0;
- int last_index = -1;
+
WorkItem* item;
- while ((item = static_cast<WorkItem*>(mpmcqueue->Get())) != nullptr) {
+ while ((item = static_cast<WorkItem*>(queue->Get())) != nullptr) {
count++;
- GPR_ASSERT(item->index > last_index);
- last_index = item->index;
GPR_ASSERT(!item->done);
- delete item;
+ item->done = true;
}
- gpr_log(GPR_DEBUG, "test_get_thd: %d times of Get() called.", count);
+ gpr_log(GPR_DEBUG, "ConsumerThread: %d times of Get() called.", count);
}
static void test_get_empty(void) {
- gpr_log(GPR_DEBUG, "test_get_empty");
- grpc_core::MPMCQueue mpmcqueue;
+ gpr_log(GPR_INFO, "test_get_empty");
+ grpc_core::InfLenFIFOQueue queue;
+ GPR_ASSERT(queue.count() == 0);
const int num_threads = 10;
grpc_core::Thread thds[num_threads];
// Fork threads. Threads should block at the beginning since queue is empty.
for (int i = 0; i < num_threads; ++i) {
- thds[i] = grpc_core::Thread("mpmcq_test_ge_thd", test_get_thd, &mpmcqueue);
+ thds[i] =
+ grpc_core::Thread("mpmcq_test_ge_thd", ConsumerThread, &queue);
thds[i].Start();
}
for (int i = 0; i < THREAD_LARGE_ITERATION; ++i) {
- mpmcqueue.Put(static_cast<void*>(new WorkItem(i)));
+ queue.Put(static_cast<void*>(new WorkItem(i)));
}
gpr_log(GPR_DEBUG, "Terminating threads...");
for (int i = 0; i < num_threads; ++i) {
- mpmcqueue.Put(nullptr);
+ queue.Put(nullptr);
}
for (int i = 0; i < num_threads; ++i) {
thds[i].Join();
@@ -105,9 +82,9 @@
gpr_log(GPR_DEBUG, "Done.");
}
-static void test_large_queue(void) {
- gpr_log(GPR_DEBUG, "test_large_queue");
- grpc_core::MPMCQueue large_queue;
+static void test_FIFO(void) {
+ gpr_log(GPR_INFO, "test_large_queue");
+ grpc_core::InfLenFIFOQueue large_queue;
for (int i = 0; i < THREAD_LARGE_ITERATION; ++i) {
large_queue.Put(static_cast<void*>(new WorkItem(i)));
}
@@ -120,18 +97,19 @@
}
// Thread for put items into queue
-class WorkThread {
+class ProducerThread {
public:
- WorkThread(grpc_core::MPMCQueue* mpmcqueue, int start_index, int num_items)
+ ProducerThread(grpc_core::InfLenFIFOQueue* queue, int start_index,
+ int num_items)
: start_index_(start_index),
num_items_(num_items),
- mpmcqueue_(mpmcqueue) {
+ queue_(queue) {
items_ = nullptr;
thd_ = grpc_core::Thread(
"mpmcq_test_mt_put_thd",
- [](void* th) { static_cast<WorkThread*>(th)->Run(); }, this);
+ [](void* th) { static_cast<ProducerThread*>(th)->Run(); }, this);
}
- ~WorkThread() {
+ ~ProducerThread() {
for (int i = 0; i < num_items_; ++i) {
GPR_ASSERT(items_[i]->done);
delete items_[i];
@@ -147,63 +125,49 @@
items_ = new WorkItem*[num_items_];
for (int i = 0; i < num_items_; ++i) {
items_[i] = new WorkItem(start_index_ + i);
- mpmcqueue_->Put(items_[i]);
+ queue_->Put(items_[i]);
}
}
int start_index_;
int num_items_;
- grpc_core::MPMCQueue* mpmcqueue_;
+ grpc_core::InfLenFIFOQueue* queue_;
grpc_core::Thread thd_;
WorkItem** items_;
};
-static void test_many_get_thd(void* args) {
- grpc_core::MPMCQueue* mpmcqueue = static_cast<grpc_core::MPMCQueue*>(args);
- // count number of Get() called in this thread
- int count = 0;
-
- WorkItem* item;
- while ((item = static_cast<WorkItem*>(mpmcqueue->Get())) != nullptr) {
- count++;
- GPR_ASSERT(!item->done);
- item->done = true;
- }
-
- gpr_log(GPR_DEBUG, "test_many_get_thd: %d times of Get() called.", count);
-}
static void test_many_thread(void) {
- gpr_log(GPR_DEBUG, "test_many_thread");
+ gpr_log(GPR_INFO, "test_many_thread");
const int num_work_thd = 10;
const int num_get_thd = 20;
- grpc_core::MPMCQueue mpmcqueue;
- WorkThread** work_thds = new WorkThread*[num_work_thd];
+ grpc_core::InfLenFIFOQueue queue;
+ ProducerThread** work_thds = new ProducerThread*[num_work_thd];
grpc_core::Thread get_thds[num_get_thd];
- gpr_log(GPR_DEBUG, "Fork WorkThread...");
+ gpr_log(GPR_DEBUG, "Fork ProducerThread...");
for (int i = 0; i < num_work_thd; ++i) {
- work_thds[i] = new WorkThread(&mpmcqueue, i * THREAD_LARGE_ITERATION,
+ work_thds[i] = new ProducerThread(&queue, i * THREAD_LARGE_ITERATION,
THREAD_LARGE_ITERATION);
work_thds[i]->Start();
}
- gpr_log(GPR_DEBUG, "WorkThread Started.");
- gpr_log(GPR_DEBUG, "For Getter Thread...");
+ gpr_log(GPR_DEBUG, "ProducerThread Started.");
+ gpr_log(GPR_DEBUG, "Fork Getter Thread...");
for (int i = 0; i < num_get_thd; ++i) {
- get_thds[i] = grpc_core::Thread("mpmcq_test_mt_get_thd", test_many_get_thd,
- &mpmcqueue);
+ get_thds[i] = grpc_core::Thread("mpmcq_test_mt_get_thd", ConsumerThread,
+ &queue);
get_thds[i].Start();
}
gpr_log(GPR_DEBUG, "Getter Thread Started.");
- gpr_log(GPR_DEBUG, "Waiting WorkThread to finish...");
+ gpr_log(GPR_DEBUG, "Waiting ProducerThread to finish...");
for (int i = 0; i < num_work_thd; ++i) {
work_thds[i]->Join();
}
- gpr_log(GPR_DEBUG, "All WorkThread Terminated.");
+ gpr_log(GPR_DEBUG, "All ProducerThread Terminated.");
gpr_log(GPR_DEBUG, "Terminating Getter Thread...");
for (int i = 0; i < num_get_thd; ++i) {
- mpmcqueue.Put(nullptr);
+ queue.Put(nullptr);
}
for (int i = 0; i < num_get_thd; ++i) {
get_thds[i].Join();
@@ -221,10 +185,8 @@
grpc::testing::TestEnvironment env(argc, argv);
grpc_init();
gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
- test_no_op();
- test_small_queue();
test_get_empty();
- test_large_queue();
+ test_FIFO();
test_many_thread();
grpc_shutdown();
return 0;
