diff --git a/include/nets/net_node.h b/include/nets/net_node.h index b0d7ede..f45275f 100644 --- a/include/nets/net_node.h +++ b/include/nets/net_node.h @@ -25,6 +25,7 @@ typedef struct node_discovery node_discovery_t; #include #include #include +#include typedef struct { tcp_server_t* server; @@ -42,7 +43,10 @@ typedef struct { void* callbackUser; // Maintenance thread for periodic tasks (orphan attach, pruning, metrics) pthread_t maintenanceThread; - volatile int maintenanceRunning; + // Cross-thread stop flag: written by Node_Destroy on the main thread, read by the maintenance + // thread's loop condition. `volatile` stops the compiler hoisting the load but provides neither + // atomicity nor ordering, so this has to be a real atomic (and TSan rightly flagged it). + _Atomic int maintenanceRunning; int maintenanceIntervalMs; // UDP ping/pong daemon (latency oracle) and peer discovery state udp_node_t* udpNode; diff --git a/include/tcpd/tcpserver.h b/include/tcpd/tcpserver.h index 2d7fa40..2f97b83 100644 --- a/include/tcpd/tcpserver.h +++ b/include/tcpd/tcpserver.h @@ -7,12 +7,15 @@ #include #include +#include typedef struct { int sockFd; // IPv6 listening socket (-1 if IPv6 unavailable) int sockFdV4; // IPv4 listening socket (-1 on bind failure) int opt; - int isRunning; + // Cross-thread stop flag: cleared by TcpServer_Stop, read by both accept threads and by + // exiting client threads. Must be atomic, not a plain int. + _Atomic int isRunning; void* owner; // Called before the client thread runs diff --git a/include/udpd/udpnode.h b/include/udpd/udpnode.h index 139ba74..2196967 100644 --- a/include/udpd/udpnode.h +++ b/include/udpd/udpnode.h @@ -7,6 +7,7 @@ #include #include +#include #define UDP_LISTEN_PORT 9393 #define UDP_PING_RETRY_INTERVAL_MS 1000 @@ -25,7 +26,9 @@ typedef struct udp_node { int sockFd; // AF_INET6, IPV6_V6ONLY=1 int sockFdV4; // AF_INET - volatile int isRunning; + // Cross-thread stop flag: cleared by UdpNode_Stop, read by the recv and retry thread loops. + // See the note on net_node_t.maintenanceRunning -- volatile is not a substitute for atomic. + _Atomic int isRunning; pthread_t recvThreadV6; pthread_t recvThreadV4; diff --git a/src/nets/net_node.c b/src/nets/net_node.c index 04581b6..3104c7b 100644 --- a/src/nets/net_node.c +++ b/src/nets/net_node.c @@ -674,10 +674,43 @@ void Node_Destroy(net_node_t* node) { pthread_join(node->maintenanceThread, NULL); } + // Detach every outbound connection from its slot under outboundLock, then tear the connections + // down outside it -- the same pattern Node_ReapDeadOutbound uses, and for the same two reasons. + // + // Calling TcpClient_Destroy directly here instead raced with still-running inbound client + // threads: those read outboundClients[i].connection under outboundLock (via + // Node_HasLiveConnectionTo), while TcpClient_Disconnect cleared the same field with no lock + // held. The lock cannot simply be held across the destroy, because that path joins the io + // thread whose on_disconnect callback takes outboundLock itself. + tcp_connection_t* outbound[MAX_CONS]; + size_t outboundToClose = 0; + + pthread_mutex_lock(&node->outboundLock); for (size_t i = 0; i < MAX_CONS; ++i) { - TcpClient_Destroy(&node->outboundClients[i]); + tcp_connection_t* conn = node->outboundClients[i].connection; + if (!conn) continue; + node->outboundClients[i].connection = NULL; + node->outboundClients[i].peerBlockHeight = 0; + outbound[outboundToClose++] = conn; } node->outboundCount = 0; + pthread_mutex_unlock(&node->outboundLock); + + for (size_t i = 0; i < outboundToClose; ++i) { + tcp_connection_t* conn = outbound[i]; + + TcpConnection_RequestClose(conn); + if (!pthread_equal(conn->ioThread, pthread_self())) { + pthread_join(conn->ioThread, NULL); + } + if (!TcpConnection_IsDisconnectNotified(conn) && conn->on_disconnect) { + TcpConnection_MarkDisconnectNotified(conn); + conn->on_disconnect(conn); + } + + TcpConnection_Destroy(conn); + free(conn); + } if (node->server) { TcpServer_Stop(node->server); diff --git a/src/tcpd/tcpserver.c b/src/tcpd/tcpserver.c index e68d4eb..e04b383 100644 --- a/src/tcpd/tcpserver.c +++ b/src/tcpd/tcpserver.c @@ -16,15 +16,20 @@ typedef struct { int listenFd; } tcpaccept_thread_args_t; -static void TcpServer_RemoveClientByPtrUnlocked(tcp_server_t* svr, tcp_connection_t* cli) { +// Returns non-zero if `cli` was still registered (and has now been unregistered). A zero return +// means someone else already claimed the slot -- see the detach logic in the client thread. +static int TcpServer_RemoveClientByPtrUnlocked(tcp_server_t* svr, tcp_connection_t* cli) { if (!svr || !svr->clientsArrPtr || !cli) { - return; + return 0; } size_t idx = Generic_FindClientInArrayByPtr(svr->clientsArrPtr, cli, svr->maxClients); if (idx != SIZE_MAX) { svr->clientsArrPtr[idx] = NULL; + return 1; } + + return 0; } static void* TcpServer_clientthreadprocess(void* ptr) { @@ -65,13 +70,26 @@ static void* TcpServer_clientthreadprocess(void* ptr) { cli->on_disconnect(cli); } + // Unregister, decide who joins us, and free -- all under clientsMutex. + // + // The destroy/free used to happen after the lock was released, which left a window where + // TcpServer_Stop could be holding this very pointer and about to use it. Doing it under the + // same lock Stop uses to inspect the slots removes that window entirely. pthread_mutex_lock(&svr->clientsMutex); - TcpServer_RemoveClientByPtrUnlocked(svr, cli); - pthread_mutex_unlock(&svr->clientsMutex); + + // If our slot was still ours, TcpServer_Stop has not claimed us and never will (we are leaving + // the array now), so nobody is going to join this thread -- detach it or its resources leak. + // If the slot was already cleared, Stop took our handle and is waiting in pthread_join, so we + // must stay joinable. + if (TcpServer_RemoveClientByPtrUnlocked(svr, cli)) { + pthread_detach(pthread_self()); + } TcpConnection_Destroy(cli); free(cli); + pthread_mutex_unlock(&svr->clientsMutex); + return NULL; } @@ -359,30 +377,42 @@ void TcpServer_Stop(tcp_server_t* ptr) { } ptr->svrThreadV4 = 0; + // Ask every live client to close and copy out its thread handle, all under clientsMutex. + // + // This used to read the client slots with the lock released, which races with an exiting client + // thread clearing its own slot -- and worse, that thread destroys and frees the connection right + // afterwards, so the pointer read here could already be freed memory. Copying the pthread_t + // while holding the lock means the join below never dereferences the connection at all, and the + // client thread cannot free itself out from under us because it does that under the same lock. pthread_mutex_lock(&ptr->clientsMutex); size_t maxClients = ptr->maxClients; - tcp_connection_t** local = ptr->clientsArrPtr; + pthread_t* joinHandles = maxClients ? (pthread_t*)calloc(maxClients, sizeof(pthread_t)) : NULL; + size_t joinCount = 0; + + if (ptr->clientsArrPtr) { + for (size_t i = 0; i < maxClients; ++i) { + tcp_connection_t* cli = ptr->clientsArrPtr[i]; + if (!cli) { + continue; + } + + TcpConnection_RequestClose(cli); + + if (joinHandles && !pthread_equal(cli->ioThread, pthread_self())) { + joinHandles[joinCount++] = cli->ioThread; + // Claim the slot: the client thread checks whether it is still registered to decide + // whether to detach itself or stay joinable for the pthread_join below. + ptr->clientsArrPtr[i] = NULL; + } + } + } pthread_mutex_unlock(&ptr->clientsMutex); - for (size_t i = 0; i < maxClients; ++i) { - tcp_connection_t* cli = local[i]; - if (!cli) { - continue; - } - - TcpConnection_RequestClose(cli); - } - - for (size_t i = 0; i < maxClients; ++i) { - tcp_connection_t* cli = local[i]; - if (!cli) { - continue; - } - - if (!pthread_equal(cli->ioThread, pthread_self())) { - pthread_join(cli->ioThread, NULL); - } + // Join outside the lock: a client thread needs clientsMutex to finish unregistering itself. + for (size_t i = 0; i < joinCount; ++i) { + pthread_join(joinHandles[i], NULL); } + free(joinHandles); pthread_mutex_lock(&ptr->clientsMutex); free(ptr->clientsArrPtr);