| … | … | @@ -146,7 +146,7 @@ void UdxTransport::stop() { |
| 146 | 146 | _running.store(false); |
| 147 | 147 | |
| 148 | 148 | // Wake up the event loop so it can exit. |
| 149 | | uv_async_send(&_async_handle); |
| 149 | _wake(); |
| 150 | 150 | |
| 151 | 151 | if (_thread.joinable()) { |
| 152 | 152 | _thread.join(); |
| … | … | @@ -187,7 +187,7 @@ void UdxTransport::send(int32_t target_peer, const uint8_t *data, int32_t size, |
| 187 | 187 | cmd.mode = mode; |
| 188 | 188 | cmd.channel = channel; |
| 189 | 189 | _queue_command(std::move(cmd)); |
| 190 | | uv_async_send(&_async_handle); |
| 190 | _wake(); |
| 191 | 191 | } |
| 192 | 192 | |
| 193 | 193 | void UdxTransport::disconnect_peer(int32_t peer_id, bool force) { |
| … | … | @@ -196,7 +196,7 @@ void UdxTransport::disconnect_peer(int32_t peer_id, bool force) { |
| 196 | 196 | cmd.peer_id = peer_id; |
| 197 | 197 | cmd.force = force; |
| 198 | 198 | _queue_command(std::move(cmd)); |
| 199 | | uv_async_send(&_async_handle); |
| 199 | _wake(); |
| 200 | 200 | } |
| 201 | 201 | |
| 202 | 202 | bool UdxTransport::pop_peer_event(PeerEvent &out) { |
| … | … | @@ -270,7 +270,7 @@ void UdxTransport::start_dht(const Ed25519KeyPair &kp, |
| 270 | 270 | _running.store(true); |
| 271 | 271 | _thread = std::thread(&UdxTransport::_thread_func, this); |
| 272 | 272 | } else { |
| 273 | | uv_async_send(&_async_handle); |
| 273 | _wake(); |
| 274 | 274 | } |
| 275 | 275 | } |
| 276 | 276 | |
| … | … | @@ -279,14 +279,14 @@ void UdxTransport::dht_connect(const uint8_t *remote_public_key) { |
| 279 | 279 | cmd.type = CMD_DHT_CONNECT; |
| 280 | 280 | memcpy(cmd.remote_public_key, remote_public_key, 32); |
| 281 | 281 | _queue_command(std::move(cmd)); |
| 282 | | uv_async_send(&_async_handle); |
| 282 | _wake(); |
| 283 | 283 | } |
| 284 | 284 | |
| 285 | 285 | void UdxTransport::dht_listen() { |
| 286 | 286 | Command cmd; |
| 287 | 287 | cmd.type = CMD_DHT_LISTEN; |
| 288 | 288 | _queue_command(std::move(cmd)); |
| 289 | | uv_async_send(&_async_handle); |
| 289 | _wake(); |
| 290 | 290 | } |
| 291 | 291 | |
| 292 | 292 | void UdxTransport::dht_announce(const uint8_t *topic) { |
| … | … | @@ -294,7 +294,7 @@ void UdxTransport::dht_announce(const uint8_t *topic) { |
| 294 | 294 | cmd.type = CMD_DHT_ANNOUNCE; |
| 295 | 295 | cmd.data.assign(topic, topic + 32); |
| 296 | 296 | _queue_command(std::move(cmd)); |
| 297 | | uv_async_send(&_async_handle); |
| 297 | _wake(); |
| 298 | 298 | } |
| 299 | 299 | |
| 300 | 300 | void UdxTransport::dht_lookup(const uint8_t *topic) { |
| … | … | @@ -302,7 +302,7 @@ void UdxTransport::dht_lookup(const uint8_t *topic) { |
| 302 | 302 | cmd.type = CMD_DHT_LOOKUP; |
| 303 | 303 | cmd.data.assign(topic, topic + 32); |
| 304 | 304 | _queue_command(std::move(cmd)); |
| 305 | | uv_async_send(&_async_handle); |
| 305 | _wake(); |
| 306 | 306 | } |
| 307 | 307 | |
| 308 | 308 | void UdxTransport::dht_join_topic(const uint8_t *topic, bool server, bool client) { |
| … | … | @@ -312,7 +312,7 @@ void UdxTransport::dht_join_topic(const uint8_t *topic, bool server, bool client |
| 312 | 312 | cmd.server = server; |
| 313 | 313 | cmd.client = client; |
| 314 | 314 | _queue_command(std::move(cmd)); |
| 315 | | uv_async_send(&_async_handle); |
| 315 | _wake(); |
| 316 | 316 | } |
| 317 | 317 | |
| 318 | 318 | void UdxTransport::dht_leave_topic(const uint8_t *topic) { |
| … | … | @@ -320,7 +320,7 @@ void UdxTransport::dht_leave_topic(const uint8_t *topic) { |
| 320 | 320 | cmd.type = CMD_DHT_LEAVE_TOPIC; |
| 321 | 321 | cmd.data.assign(topic, topic + 32); |
| 322 | 322 | _queue_command(std::move(cmd)); |
| 323 | | uv_async_send(&_async_handle); |
| 323 | _wake(); |
| 324 | 324 | } |
| 325 | 325 | |
| 326 | 326 | void UdxTransport::dht_immutable_put(const uint8_t *value, size_t value_len) { |
| … | … | @@ -328,7 +328,7 @@ void UdxTransport::dht_immutable_put(const uint8_t *value, size_t value_len) { |
| 328 | 328 | cmd.type = CMD_DHT_IMMUTABLE_PUT; |
| 329 | 329 | cmd.data.assign(value, value + value_len); |
| 330 | 330 | _queue_command(std::move(cmd)); |
| 331 | | uv_async_send(&_async_handle); |
| 331 | _wake(); |
| 332 | 332 | } |
| 333 | 333 | |
| 334 | 334 | void UdxTransport::dht_immutable_get(const uint8_t *key) { |
| … | … | @@ -336,7 +336,7 @@ void UdxTransport::dht_immutable_get(const uint8_t *key) { |
| 336 | 336 | cmd.type = CMD_DHT_IMMUTABLE_GET; |
| 337 | 337 | cmd.data.assign(key, key + 32); |
| 338 | 338 | _queue_command(std::move(cmd)); |
| 339 | | uv_async_send(&_async_handle); |
| 339 | _wake(); |
| 340 | 340 | } |
| 341 | 341 | |
| 342 | 342 | void UdxTransport::dht_mutable_put(const uint8_t *value, size_t value_len, uint64_t seq) { |
| … | … | @@ -345,7 +345,7 @@ void UdxTransport::dht_mutable_put(const uint8_t *value, size_t value_len, uint6 |
| 345 | 345 | cmd.data.assign(value, value + value_len); |
| 346 | 346 | cmd.seq = seq; |
| 347 | 347 | _queue_command(std::move(cmd)); |
| 348 | | uv_async_send(&_async_handle); |
| 348 | _wake(); |
| 349 | 349 | } |
| 350 | 350 | |
| 351 | 351 | void UdxTransport::dht_mutable_get(const uint8_t *public_key, uint64_t min_seq) { |
| … | … | @@ -354,7 +354,7 @@ void UdxTransport::dht_mutable_get(const uint8_t *public_key, uint64_t min_seq) |
| 354 | 354 | cmd.data.assign(public_key, public_key + 32); |
| 355 | 355 | cmd.seq = min_seq; |
| 356 | 356 | _queue_command(std::move(cmd)); |
| 357 | | uv_async_send(&_async_handle); |
| 357 | _wake(); |
| 358 | 358 | } |
| 359 | 359 | |
| 360 | 360 | // ---- Background Thread ---- |
| … | … | @@ -369,6 +369,29 @@ static void _tick_cb(uv_timer_t *handle) { |
| 369 | 369 | transport->_process_commands(); |
| 370 | 370 | } |
| 371 | 371 | |
| 372 | // ⛔ THE ONLY WAY TO WAKE THE LOOP, because the guard was previously written at |
| 373 | // exactly ONE of twenty call sites. |
| 374 | // |
| 375 | // `_async_handle` is a uv_async_t that does not exist until `_thread_func` has |
| 376 | // run `uv_async_init` on it. Signalling it before that reads a null callback |
| 377 | // pointer and jumps to address zero. run_on_loop checked `_async_ready`; |
| 378 | // swarm_join, swarm_leave, the peer variants, every dht_* command, send, |
| 379 | // disconnect_peer, start_dht and stop did not — nineteen unguarded senders and |
| 380 | // one guarded one. |
| 381 | // |
| 382 | // It was reachable the moment a client issued a command promptly after start: |
| 383 | // measured as a SEGV in 7 runs of 20 of swarmd's test_client, with the UDX |
| 384 | // thread still inside udx_socket_bind and thread 4 at `0x0 in ?? ()`. Nothing |
| 385 | // before it called anything that fast — the e2e gates all wait for dht=up |
| 386 | // first, which is why a race this plain survived. |
| 387 | // |
| 388 | // Commands are queued regardless, and `_thread_func` sends to itself the |
| 389 | // instant the handle exists, so a wake dropped here is a wake deferred by |
| 390 | // microseconds and not a command lost. |
| 391 | void UdxTransport::_wake() { |
| 392 | if (_async_ready) uv_async_send(&_async_handle); |
| 393 | } |
| 394 | |
| 372 | 395 | void UdxTransport::_thread_func() { |
| 373 | 396 | uv_loop_init(&_loop); |
| 374 | 397 | udx_init(&_loop, &_udx, nullptr); |
| … | … | @@ -378,7 +401,7 @@ void UdxTransport::_thread_func() { |
| 378 | 401 | // Only now may anyone signal it. Publish AFTER init, and flush whatever was |
| 379 | 402 | // queued while the thread was starting. |
| 380 | 403 | _async_ready = true; |
| 381 | | uv_async_send(&_async_handle); |
| 404 | _wake(); |
| 382 | 405 | |
| 383 | 406 | // Periodic timer for processing commands. |
| 384 | 407 | uv_timer_init(&_loop, &_tick_timer); |
| … | … | @@ -450,8 +473,8 @@ void UdxTransport::run_on_loop(std::function<void()> task) { |
| 450 | 473 | std::lock_guard<std::mutex> lock(_tasks_mutex); |
| 451 | 474 | _tasks.push_back(std::move(task)); |
| 452 | 475 | } |
| 453 | | // Queue always; wake only once the handle exists. See _async_ready. |
| 454 | | if (_async_ready) uv_async_send(&_async_handle); |
| 476 | // Queue always; wake only once the handle exists. See _wake(). |
| 477 | _wake(); |
| 455 | 478 | } |
| 456 | 479 | |
| 457 | 480 | void UdxTransport::_process_commands() { |
| … | … | @@ -1222,7 +1245,7 @@ void UdxTransport::swarm_join(const uint8_t *topic, bool server, bool client) { |
| 1222 | 1245 | cmd.server = server; |
| 1223 | 1246 | cmd.client = client; |
| 1224 | 1247 | _queue_command(std::move(cmd)); |
| 1225 | | uv_async_send(&_async_handle); |
| 1248 | _wake(); |
| 1226 | 1249 | } |
| 1227 | 1250 | |
| 1228 | 1251 | void UdxTransport::swarm_leave(const uint8_t *topic) { |
| … | … | @@ -1230,7 +1253,7 @@ void UdxTransport::swarm_leave(const uint8_t *topic) { |
| 1230 | 1253 | cmd.type = CMD_SWARM_LEAVE; |
| 1231 | 1254 | cmd.data.assign(topic, topic + 32); |
| 1232 | 1255 | _queue_command(std::move(cmd)); |
| 1233 | | uv_async_send(&_async_handle); |
| 1256 | _wake(); |
| 1234 | 1257 | } |
| 1235 | 1258 | |
| 1236 | 1259 | void UdxTransport::swarm_join_peer(const uint8_t *pk) { |
| … | … | @@ -1238,7 +1261,7 @@ void UdxTransport::swarm_join_peer(const uint8_t *pk) { |
| 1238 | 1261 | cmd.type = CMD_SWARM_JOIN_PEER; |
| 1239 | 1262 | memcpy(cmd.remote_public_key, pk, 32); |
| 1240 | 1263 | _queue_command(std::move(cmd)); |
| 1241 | | uv_async_send(&_async_handle); |
| 1264 | _wake(); |
| 1242 | 1265 | } |
| 1243 | 1266 | |
| 1244 | 1267 | void UdxTransport::swarm_leave_peer(const uint8_t *pk) { |
| … | … | @@ -1246,7 +1269,7 @@ void UdxTransport::swarm_leave_peer(const uint8_t *pk) { |
| 1246 | 1269 | cmd.type = CMD_SWARM_LEAVE_PEER; |
| 1247 | 1270 | memcpy(cmd.remote_public_key, pk, 32); |
| 1248 | 1271 | _queue_command(std::move(cmd)); |
| 1249 | | uv_async_send(&_async_handle); |
| 1272 | _wake(); |
| 1250 | 1273 | } |
| 1251 | 1274 | |
| 1252 | 1275 | bool UdxTransport::pop_swarm_event(SwarmEvent &out) { |