| 187 | |
| 188 | template <typename Queue> |
| 189 | void ConcurrentQueueBasicTest(Queue& queue) { |
| 190 | #ifndef ARROW_ENABLE_THREADING |
| 191 | GTEST_SKIP() << "Test requires threading enabled"; |
| 192 | #endif |
| 193 | ASSERT_TRUE(queue.Empty()); |
| 194 | queue.Push(1); |
| 195 | ASSERT_FALSE(queue.Empty()); |
| 196 | ASSERT_EQ(queue.TryPop(), std::make_optional(1)); |
| 197 | ASSERT_TRUE(queue.Empty()); |
| 198 | |
| 199 | auto fut_pop = std::async(std::launch::async, [&]() { return queue.WaitAndPop(); }); |
| 200 | ASSERT_EQ(fut_pop.wait_for(std::chrono::milliseconds(10)), std::future_status::timeout); |
| 201 | queue.Push(2); |
| 202 | queue.Push(3); |
| 203 | queue.Push(4); |
| 204 | // Note we should use wait() which guarantees the future is ready, but this will make |
| 205 | // the test hang forever if the code is broken. Thus we in turn use wait_for() with a |
| 206 | // large enough timeout which should be enough in practice. |
| 207 | ASSERT_EQ(fut_pop.wait_for(std::chrono::seconds(5)), std::future_status::ready); |
| 208 | ASSERT_EQ(fut_pop.get(), 2); |
| 209 | fut_pop = std::async(std::launch::async, [&]() { return queue.WaitAndPop(); }); |
| 210 | // Ditto. |
| 211 | ASSERT_EQ(fut_pop.wait_for(std::chrono::seconds(5)), std::future_status::ready); |
| 212 | ASSERT_EQ(fut_pop.get(), 3); |
| 213 | ASSERT_FALSE(queue.Empty()); |
| 214 | ASSERT_EQ(queue.TryPop(), std::make_optional(4)); |
| 215 | ASSERT_EQ(queue.TryPop(), std::nullopt); |
| 216 | queue.Push(5); |
| 217 | ASSERT_FALSE(queue.Empty()); |
| 218 | ASSERT_EQ(queue.Front(), 5); |
| 219 | ASSERT_FALSE(queue.Empty()); |
| 220 | queue.Clear(); |
| 221 | ASSERT_TRUE(queue.Empty()); |
| 222 | } |
| 223 | |
| 224 | TEST(ConcurrentQueue, BasicTest) { |
| 225 | ConcurrentQueue<int> queue; |