| 59 | |
| 60 | template <typename Queue> |
| 61 | void ProducerThread(Queue* queue) { |
| 62 | int max_inprogress = FLAGS_max_queue_size - FLAGS_num_producers; |
| 63 | while (true) { |
| 64 | while (inprogress > max_inprogress) { |
| 65 | base::subtle::PauseCPU(); |
| 66 | } |
| 67 | inprogress++; |
| 68 | InboundCall* call = new InboundCall(nullptr); |
| 69 | std::optional<InboundCall*> evicted; |
| 70 | auto status = queue->Put(call, &evicted); |
| 71 | if (status == QUEUE_FULL) { |
| 72 | LOG(INFO) << "queue full: producer exiting"; |
| 73 | delete call; |
| 74 | break; |
| 75 | } |
| 76 | |
| 77 | if (PREDICT_TRUE(evicted)) { |
| 78 | LOG(INFO) << "call evicted: producer exiting"; |
| 79 | delete *evicted; |
| 80 | break; |
| 81 | } |
| 82 | |
| 83 | if (PREDICT_TRUE(status == QUEUE_SHUTDOWN)) { |
| 84 | delete call; |
| 85 | break; |
| 86 | } |
| 87 | } |
| 88 | } |
| 89 | |
| 90 | template <typename Queue> |
| 91 | void ConsumerThread(Queue* queue) { |