Addressing comments

src/core/lib/iomgr/executor/mpmcqueue.cc

@@ -27,7 +27,7 @@ DebugOnlyTraceFlag grpc_thread_pool_trace(false, "thread_pool");
 inline void* InfLenFIFOQueue::PopFront() {
   // Caller should already check queue is not empty and has already held the
   // mutex. This function will assume that there is at least one element in the
-  // queue (aka queue_head_ content is valid).
+  // queue (i.e. queue_head_->content is valid).
   void* result = queue_head_->content;
   count_.Store(count_.Load(MemoryOrder::RELAXED) - 1, MemoryOrder::RELAXED);
 
@@ -65,6 +65,7 @@ inline void* InfLenFIFOQueue::PopFront() {
 }
 
 InfLenFIFOQueue::Node* InfLenFIFOQueue::AllocateNodes(int num) {
+  num_nodes_  = num_nodes_ + num;
   Node* new_chunk = static_cast<Node*>(gpr_zalloc(sizeof(Node) * num));
   new_chunk[0].next = &new_chunk[1];
   new_chunk[num - 1].prev = &new_chunk[num - 2];
@@ -76,11 +77,11 @@ InfLenFIFOQueue::Node* InfLenFIFOQueue::AllocateNodes(int num) {
 }
 
 InfLenFIFOQueue::InfLenFIFOQueue() {
-  delete_list_size_ = 1024;
+  delete_list_size_ = kDeleteListInitSize;
   delete_list_ =
       static_cast<Node**>(gpr_zalloc(sizeof(Node*) * delete_list_size_));
 
-  Node* new_chunk = AllocateNodes(1024);
+  Node* new_chunk = AllocateNodes(kDeleteListInitSize);
   delete_list_[delete_list_count_++] = new_chunk;
   queue_head_ = queue_tail_ = new_chunk;
   new_chunk[0].prev = &new_chunk[1023];
@@ -126,10 +127,11 @@ void InfLenFIFOQueue::Put(void* elem) {
     stats_.num_started++;
     gpr_log(GPR_INFO, "[InfLenFIFOQueue Put] num_started:        %" PRIu64,
             stats_.num_started);
+    auto current_time = gpr_now(GPR_CLOCK_MONOTONIC);
     if (curr_count == 0) {
-      busy_time = gpr_now(GPR_CLOCK_MONOTONIC);
+      busy_time = current_time;
     }
-    queue_tail_->insert_time = gpr_now(GPR_CLOCK_MONOTONIC);
+    queue_tail_->insert_time = current_time;
   }
 
   count_.Store(curr_count + 1, MemoryOrder::RELAXED);
@@ -163,14 +165,6 @@ void* InfLenFIFOQueue::Get(gpr_timespec* wait_time) {
   return PopFront();
 }
 
-size_t InfLenFIFOQueue::num_node() {
-  size_t num = 1024;
-  for (size_t i = 1; i < delete_list_count_; ++i) {
-    num = num * 2;
-  }
-  return num;
-}
-
 void InfLenFIFOQueue::PushWaiter(Waiter* waiter) {
   waiter->next = waiters_.next;
   waiter->prev = &waiters_;

src/core/lib/iomgr/executor/mpmcqueue.h

@@ -89,7 +89,10 @@ class InfLenFIFOQueue : public MPMCQueueInterface {
 
   // For test purpose only. Returns number of nodes allocated in queue.
   // All allocated nodes will not be free until destruction of queue.
-  size_t num_node();
+  int num_nodes() const { return num_nodes_; }
+
+  // For test purpose only. Returns the initial number of nodes in queue.
+  int init_num_nodes() const { return kQueueInitNumNodes; }
 
  private:
   // For Internal Use Only.
@@ -144,6 +147,9 @@ class InfLenFIFOQueue : public MPMCQueueInterface {
   Mutex mu_;        // Protecting lock
   Waiter waiters_;  // Head of waiting thread queue
 
+  const int kDeleteListInitSize = 1024;  // Initial size for delete list
+  const int kQueueInitNumNodes = 1024;   // Initial number of nodes allocated
+
   Node** delete_list_ = nullptr;  // Keeps track of all allocated array entries
                                   // for deleting on destruction
   size_t delete_list_count_ = 0;  // Number of entries in list
@@ -153,6 +159,7 @@ class InfLenFIFOQueue : public MPMCQueueInterface {
   Node* queue_head_ = nullptr;  // Head of the queue, remove position
   Node* queue_tail_ = nullptr;  // End of queue, insert position
   Atomic<int> count_{0};        // Number of elements in queue
+  int num_nodes_ = 0;        // Number of nodes allocated
 
   Stats stats_;            // Stats info
   gpr_timespec busy_time;  // Start time of busy queue

test/core/iomgr/mpmcqueue_test.cc

@@ -119,47 +119,53 @@ static void test_FIFO(void) {
   }
 }
 
+// Test if queue's behavior of expanding is correct. (Only does expansion when
+// it gets full, and each time expands to doubled size).
 static void test_space_efficiency(void) {
   gpr_log(GPR_INFO, "test_space_efficiency");
   grpc_core::InfLenFIFOQueue queue;
-  for (int i = 0; i < 1024; ++i) {
+  for (int i = 0; i < queue.init_num_nodes(); ++i) {
     queue.Put(static_cast<void*>(grpc_core::New<WorkItem>(i)));
   }
-  GPR_ASSERT(queue.num_node() == 1024);
-  for (int i = 0; i < 1024; ++i) {
+  // List should not have been expanded at this time.
+  GPR_ASSERT(queue.num_nodes() == queue.init_num_nodes());
+  for (int i = 0; i < queue.init_num_nodes(); ++i) {
     WorkItem* item = static_cast<WorkItem*>(queue.Get());
     queue.Put(item);
   }
-  GPR_ASSERT(queue.num_node() == 1024);
-  for (int i = 0; i < 1024; ++i) {
+  GPR_ASSERT(queue.num_nodes() == queue.init_num_nodes());
+  for (int i = 0; i < queue.init_num_nodes(); ++i) {
     WorkItem* item = static_cast<WorkItem*>(queue.Get());
     grpc_core::Delete(item);
   }
-  GPR_ASSERT(queue.num_node() == 1024);
+  GPR_ASSERT(queue.num_nodes() == queue.init_num_nodes());
   GPR_ASSERT(queue.count() == 0);
   // queue empty now
-  for (int i = 0; i < 4000; ++i) {
+  for (int i = 0; i < queue.init_num_nodes() * 2; ++i) {
     queue.Put(static_cast<void*>(grpc_core::New<WorkItem>(i)));
   }
-  GPR_ASSERT(queue.count() == 4000);
-  GPR_ASSERT(queue.num_node() == 4096);
-  for (int i = 0; i < 2000; ++i) {
+  GPR_ASSERT(queue.count() == queue.init_num_nodes() * 2);
+  // List should have been expanded once.
+  GPR_ASSERT(queue.num_nodes() == queue.init_num_nodes() * 2);
+  for (int i = 0; i < queue.init_num_nodes(); ++i) {
     WorkItem* item = static_cast<WorkItem*>(queue.Get());
     grpc_core::Delete(item);
   }
-  GPR_ASSERT(queue.count() == 2000);
-  GPR_ASSERT(queue.num_node() == 4096);
-  for (int i = 0; i < 1000; ++i) {
+  GPR_ASSERT(queue.count() == queue.init_num_nodes());
+  // List will never shrink, should keep same number of node as before.
+  GPR_ASSERT(queue.num_nodes() == queue.init_num_nodes() * 2);
+  for (int i = 0; i < queue.init_num_nodes() + 1; ++i) {
     queue.Put(static_cast<void*>(grpc_core::New<WorkItem>(i)));
   }
-  GPR_ASSERT(queue.count() == 3000);
-  GPR_ASSERT(queue.num_node() == 4096);
-  for (int i = 0; i < 3000; ++i) {
+  GPR_ASSERT(queue.count() == queue.init_num_nodes() * 2 + 1);
+  // List should have been expanded twice.
+  GPR_ASSERT(queue.num_nodes() == queue.init_num_nodes() * 4);
+  for (int i = 0; i < queue.init_num_nodes() * 2 + 1; ++i) {
     WorkItem* item = static_cast<WorkItem*>(queue.Get());
     grpc_core::Delete(item);
   }
   GPR_ASSERT(queue.count() == 0);
-  GPR_ASSERT(queue.num_node() == 4096);
+  GPR_ASSERT(queue.num_nodes() == queue.init_num_nodes() * 4);
   gpr_log(GPR_DEBUG, "Done.");
 }