Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Modify class name and other small changes

Browse Source
pull/19358/head
Yunjia Wang 6 years ago
parent d957083bb8
commit fdd856312a
3 changed files with 107 additions and 149 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 97
      src/core/lib/iomgr/executor/mpmcqueue.cc
  2. 45
      src/core/lib/iomgr/executor/mpmcqueue.h
  3. 114
      test/core/iomgr/mpmcqueue_test.cc

97
src/core/lib/iomgr/executor/mpmcqueue.cc
Unescape View File

@ -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 =
gpr_time_sub(gpr_now(GPR_CLOCK_PRECISE), head_to_remove->insert_time);
gpr_free(head_to_remove);
// 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);
count_.FetchSub(1, MemoryOrder::RELAXED);
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 (GRPC_TRACE_FLAG_ENABLED(thread_pool)) {
gpr_timespec wait_time =
gpr_time_sub(gpr_now(GPR_CLOCK_PRECISE), head_to_remove->insert_time);
if (GRPC_TRACE_FLAG_ENABLED(thread_pool_trace)) {
PrintStats();
// 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);
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));
}
Delete(head_to_remove);
// Singal waiting thread
if (count_.Load(MemoryOrder::RELAXED) > 0 && num_waiters_ > 0) {
wait_nonempty_.Signal();
@ -71,37 +79,34 @@ inline void* MPMCQueue::PopFront() {
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::Put(void* elem) {
}
}
void* MPMCQueue::Get() {
void* InfLenFIFOQueue::Get() {
MutexLock l(&mu_);
if (count_.Load(MemoryOrder::RELAXED) == 0) {
num_waiters_++;
@ -118,20 +123,8 @@ void* MPMCQueue::Get() {
} 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

45
src/core/lib/iomgr/executor/mpmcqueue.h
Unescape View File

@ -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 @@ class MPMCQueue : public MPMCQueueInterface {
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 @@ class MPMCQueue : public MPMCQueueInterface {
}
};
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 @@ class MPMCQueue : public MPMCQueueInterface {
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 */

114
test/core/iomgr/mpmcqueue_test.cc
Unescape View File

@ -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 @@ struct WorkItem {
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 @@ static void test_get_empty(void) {
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 @@ static void test_large_queue(void) {
}
// 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 @@ class WorkThread {
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 @@ int main(int argc, char** argv) {
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;

Write Preview
Loading…
Cancel
Save