| … | … | @@ -385,22 +385,18 @@ void CogChannel<Client, Data>::reply_handler(const Msg& msg) |
| 385 | 385 | template<typename Client, typename Data> |
| 386 | 386 | void CogChannel<Client, Data>::barrier() |
| 387 | 387 | { |
| 388 | | // Drain work queues. |
| 389 | | _msg_buffer.barrier(); |
| 390 | | |
| 391 | | // Server can complete barrier only after it receives it on |
| 392 | | // all of the open sockets; the random string is the uuid to |
| 393 | | // disambiguate this barrier from any others that might get |
| 394 | | // issued. The barrier is retired after all N of them are received. |
| 388 | // Generate a unique barrier ID |
| 395 | 389 | static thread_local std::minstd_rand rng(std::random_device{}()); |
| 396 | 390 | uint64_t rnd = (uint64_t(rng()) << 32) | rng(); |
| 397 | 391 | |
| 398 | | std::lock_guard<std::mutex> lck(_sock_set_mtx); |
| 392 | // Build the barrier message. Each worker thread will send this, |
| 393 | // opening its socket if needed. Server completes when all N arrive. |
| 399 | 394 | char msg[64]; |
| 400 | | snprintf(msg, sizeof(msg), "(cog-barrier %zu \"%016lx\")\n", |
| 401 | | _open_socks.size(), rnd); |
| 402 | | for (int sockfd : _open_socks) |
| 403 | | send(sockfd, msg, strlen(msg), MSG_NOSIGNAL); |
| 395 | snprintf(msg, sizeof(msg), "(cog-barrier %u \"%016lx\")\n", NTHREADS, rnd); |
| 396 | |
| 397 | Data dummy = Data(); |
| 398 | Msg block{nullptr, nullptr, true, msg, dummy}; |
| 399 | _msg_buffer.barrier(block); |
| 404 | 400 | } |
| 405 | 401 | |
| 406 | 402 | template<typename Client, typename Data> |