Reap dead outbound connection slots; strike disconnected peers from discovery. Two related connection-lifecycle fixes on top of the SIGPIPE fix.
Reclaim outbound slots on peer disconnect: - TcpClient_ThreadProc fired on_disconnect but never cleared the outbound slot, closed the fd, joined the io thread, or freed the connection, so a dead peer permanently held its outboundClients[] slot. After MAX_CONS (32) churned connections the node could make no new outbound connections, and leaked fds/threads/memory. (The inbound side already self-reclaimed.) - Add a reaper (Node_ReapDeadOutbound) on the maintenance thread: under outboundLock it detaches dead (disconnect-notified) slots, then joins the io thread and destroys/frees each connection outside the lock. - Guard against use-after-free with a pin count on tcp_connection_t (TcpConnection_Pin/Unpin). The only cross-thread consumer holding a raw connection pointer across a blocking op is the `sync` command (via Node_GetBestOutboundPeer); it now pins the peer and unpins when done, and the reaper skips pinned connections. Discovery's snapshots run on the reaper's own thread, so they need no pin. - Node_GetBestOutboundPeer/GetClientList/GetPeerEndpoints skip disconnect-notified connections so a dead peer is never handed out. - Node_Destroy stops+joins the maintenance thread before tearing down outbound clients, so the reaper can't race shutdown. Strike disconnected peers from the discovery peer list: - Add NodeDiscovery_RemovePeer + Node_HandlePeerDisconnect: on disconnect, remove the peer from the known-peer table once no live connection (inbound or outbound) to its listen endpoint remains (Node_HasLiveConnectionTo; disconnect-notified conns don't count, so both directions dropping at once is handled). Wired into Node_Server_OnDisconnect and Node_Client_OnDisconnect.
This commit is contained in:
@@ -23,6 +23,10 @@ void NodeDiscovery_OnGetPeers(node_discovery_t* disc, tcp_connection_t* fromConn
|
||||
// Decode a received PEERS payload and fold a couple of its endpoints into the known-peer table.
|
||||
void NodeDiscovery_OnPeersReceived(node_discovery_t* disc, tcp_connection_t* fromConn, const unsigned char* payload, size_t payloadLen);
|
||||
|
||||
// Strike a peer (by its listen endpoint) from the known-peer table. Called when a peer becomes
|
||||
// logically disconnected (no remaining connection to it).
|
||||
void NodeDiscovery_RemovePeer(node_discovery_t* disc, const struct sockaddr_storage* endpoint);
|
||||
|
||||
// Dump the known-peer table to stdout (for the CLI `peers` command).
|
||||
void NodeDiscovery_PrintPeers(node_discovery_t* disc);
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <pthread.h>
|
||||
#include <stdatomic.h>
|
||||
#include <stdbool.h>
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
@@ -37,6 +38,11 @@ struct tcp_connection_t {
|
||||
bool closing;
|
||||
bool disconnectedNotified;
|
||||
|
||||
// Non-zero while another thread holds a raw pointer to this connection taken from a
|
||||
// lock-protected snapshot and used after releasing the lock. The reaper must not free a
|
||||
// pinned connection. See TcpConnection_Pin/Unpin.
|
||||
atomic_int pinCount;
|
||||
|
||||
unsigned char* dataBuf;
|
||||
size_t dataBufLen;
|
||||
size_t dataBufCap;
|
||||
@@ -75,4 +81,9 @@ void TcpConnection_RequestClose(tcp_connection_t* conn);
|
||||
void TcpConnection_MarkDisconnectNotified(tcp_connection_t* conn);
|
||||
bool TcpConnection_IsDisconnectNotified(tcp_connection_t* conn);
|
||||
|
||||
// Pin/unpin a connection so a background reaper won't free it while a caller still holds a raw
|
||||
// pointer to it (e.g. across a blocking operation after releasing the collection lock).
|
||||
void TcpConnection_Pin(tcp_connection_t* conn);
|
||||
void TcpConnection_Unpin(tcp_connection_t* conn);
|
||||
|
||||
#endif
|
||||
|
||||
@@ -1308,6 +1308,9 @@ int main(int argc, char* argv[]) {
|
||||
if ((uint64_t)Chain_Size(chain) >= peerHeight) break;
|
||||
continue;
|
||||
}
|
||||
// Sync loop finished with this peer; release the pin taken by Node_GetBestOutboundPeer so
|
||||
// the reaper may reclaim the slot if the peer has since disconnected.
|
||||
TcpConnection_Unpin(peerConn);
|
||||
|
||||
if (strcmp(cmd, "txpooldetail") == 0) {
|
||||
char* hashStr = strtok(NULL, " \t");
|
||||
|
||||
+100
-14
@@ -119,6 +119,47 @@ static int Node_HasOtherInboundFrom(net_node_t* node, const tcp_connection_t* se
|
||||
return found;
|
||||
}
|
||||
|
||||
// Returns non-zero if a connection OTHER than `exclude` to `endpoint` is still live. A connection
|
||||
// that is itself mid-disconnect (disconnectedNotified) does not count as live - this is what lets
|
||||
// us decide a peer is fully gone even when both its inbound and outbound drop simultaneously.
|
||||
static int Node_HasLiveConnectionTo(net_node_t* node, const struct sockaddr_storage* endpoint, const tcp_connection_t* exclude) {
|
||||
int found = 0;
|
||||
|
||||
pthread_mutex_lock(&node->outboundLock);
|
||||
for (size_t i = 0; i < MAX_CONS && !found; ++i) {
|
||||
tcp_connection_t* c = node->outboundClients[i].connection;
|
||||
if (!c || c == exclude || TcpConnection_IsDisconnectNotified(c)) continue;
|
||||
struct sockaddr_storage ep;
|
||||
if (Node_ConnListenEndpoint(c, &ep) && NetNode_EndpointEqual(&ep, endpoint)) found = 1;
|
||||
}
|
||||
pthread_mutex_unlock(&node->outboundLock);
|
||||
if (found) return 1;
|
||||
|
||||
if (node->server) {
|
||||
pthread_mutex_lock(&node->server->clientsMutex);
|
||||
for (size_t i = 0; i < node->server->maxClients && !found; ++i) {
|
||||
tcp_connection_t* c = node->server->clientsArrPtr ? node->server->clientsArrPtr[i] : NULL;
|
||||
if (!c || c == exclude || TcpConnection_IsDisconnectNotified(c)) continue;
|
||||
struct sockaddr_storage ep;
|
||||
if (Node_ConnListenEndpoint(c, &ep) && NetNode_EndpointEqual(&ep, endpoint)) found = 1;
|
||||
}
|
||||
pthread_mutex_unlock(&node->server->clientsMutex);
|
||||
}
|
||||
return found;
|
||||
}
|
||||
|
||||
// Called when a connection to a peer drops. Strikes the peer from the discovery table, but only
|
||||
// once it is logically disconnected - i.e. no other live connection (inbound or outbound) to the
|
||||
// same listen endpoint remains. Must be called from the disconnect callback while `conn` is still
|
||||
// valid and outside outboundLock/clientsMutex.
|
||||
static void Node_HandlePeerDisconnect(net_node_t* node, tcp_connection_t* conn) {
|
||||
if (!node || !node->discovery || !conn) return;
|
||||
struct sockaddr_storage ep;
|
||||
if (!Node_ConnListenEndpoint(conn, &ep)) return; // never advertised an endpoint -> not tracked
|
||||
if (Node_HasLiveConnectionTo(node, &ep, conn)) return; // still reachable via another connection
|
||||
NodeDiscovery_RemovePeer(node->discovery, &ep);
|
||||
}
|
||||
|
||||
int Node_ConnListenEndpoint(const tcp_connection_t* conn, struct sockaddr_storage* out) {
|
||||
if (!conn || !out) return 0;
|
||||
memset(out, 0, sizeof(*out));
|
||||
@@ -170,7 +211,7 @@ size_t Node_GetPeerEndpoints(net_node_t* node, struct sockaddr_storage* outEndpo
|
||||
pthread_mutex_lock(&node->outboundLock);
|
||||
for (size_t i = 0; i < MAX_CONS && count < maxOut; ++i) {
|
||||
tcp_connection_t* c = node->outboundClients[i].connection;
|
||||
if (!c) continue;
|
||||
if (!c || TcpConnection_IsDisconnectNotified(c)) continue; // ignore connections that are tearing down
|
||||
struct sockaddr_storage ep;
|
||||
if (!Node_ConnListenEndpoint(c, &ep)) continue;
|
||||
int dup = 0;
|
||||
@@ -186,7 +227,7 @@ size_t Node_GetPeerEndpoints(net_node_t* node, struct sockaddr_storage* outEndpo
|
||||
pthread_mutex_lock(&node->server->clientsMutex);
|
||||
for (size_t i = 0; i < node->server->maxClients && count < maxOut; ++i) {
|
||||
tcp_connection_t* c = node->server->clientsArrPtr ? node->server->clientsArrPtr[i] : NULL;
|
||||
if (!c) continue;
|
||||
if (!c || TcpConnection_IsDisconnectNotified(c)) continue; // ignore connections that are tearing down
|
||||
struct sockaddr_storage ep;
|
||||
if (!Node_ConnListenEndpoint(c, &ep)) continue;
|
||||
int dup = 0;
|
||||
@@ -226,6 +267,41 @@ typedef enum {
|
||||
NODE_BLOCK_ACCEPTED = 2
|
||||
} node_block_accept_result_t;
|
||||
|
||||
// Reclaims outbound slots whose peer has disconnected. Mirrors the inbound self-reclaim in
|
||||
// TcpServer_clientthreadprocess: detach dead connections from their slots under outboundLock, then
|
||||
// join their io threads and destroy/free them outside the lock. Pinned connections (a raw pointer
|
||||
// is still held elsewhere, e.g. by an in-progress sync) are skipped and retried on a later tick.
|
||||
static void Node_ReapDeadOutbound(net_node_t* node) {
|
||||
if (!node) return;
|
||||
|
||||
tcp_connection_t* dead[MAX_CONS];
|
||||
size_t deadCount = 0;
|
||||
|
||||
pthread_mutex_lock(&node->outboundLock);
|
||||
for (size_t i = 0; i < MAX_CONS; ++i) {
|
||||
tcp_connection_t* c = node->outboundClients[i].connection;
|
||||
if (!c) continue;
|
||||
if (!TcpConnection_IsDisconnectNotified(c)) continue; // still live
|
||||
if (atomic_load(&c->pinCount) != 0) continue; // someone holds a raw pointer; retry later
|
||||
// Detach the dead connection from its slot and reset the slot to a clean free state.
|
||||
node->outboundClients[i].connection = NULL;
|
||||
node->outboundClients[i].peerBlockHeight = 0;
|
||||
dead[deadCount++] = c;
|
||||
}
|
||||
pthread_mutex_unlock(&node->outboundLock);
|
||||
|
||||
// Join + destroy outside the lock: the io thread's on_disconnect callback itself takes
|
||||
// outboundLock, so joining under it would deadlock.
|
||||
for (size_t i = 0; i < deadCount; ++i) {
|
||||
tcp_connection_t* c = dead[i];
|
||||
if (!pthread_equal(c->ioThread, pthread_self())) {
|
||||
pthread_join(c->ioThread, NULL);
|
||||
}
|
||||
TcpConnection_Destroy(c);
|
||||
free(c);
|
||||
}
|
||||
}
|
||||
|
||||
static void* Node_MaintenanceThread(void* arg) {
|
||||
net_node_t* n = (net_node_t*)arg;
|
||||
if (!n) return NULL;
|
||||
@@ -238,6 +314,8 @@ static void* Node_MaintenanceThread(void* arg) {
|
||||
BalanceSheet_SaveToFile(chainDataDir);
|
||||
}
|
||||
}
|
||||
// Reclaim outbound slots whose peer has disconnected so they can be reused.
|
||||
Node_ReapDeadOutbound(n);
|
||||
// Peer discovery tick: ping/query connected peers and connect to the best-ping discoveries.
|
||||
if (n->discovery) {
|
||||
NodeDiscovery_Iterate(n->discovery);
|
||||
@@ -496,6 +574,14 @@ void Node_Destroy(net_node_t* node) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Stop the maintenance thread first: it runs the outbound reaper (which touches outboundClients
|
||||
// and outboundLock) and the discovery tick, so it must not run concurrently with the teardown
|
||||
// below or against soon-to-be-destroyed state.
|
||||
if (node->maintenanceRunning) {
|
||||
node->maintenanceRunning = 0;
|
||||
pthread_join(node->maintenanceThread, NULL);
|
||||
}
|
||||
|
||||
for (size_t i = 0; i < MAX_CONS; ++i) {
|
||||
TcpClient_Destroy(&node->outboundClients[i]);
|
||||
}
|
||||
@@ -506,12 +592,6 @@ void Node_Destroy(net_node_t* node) {
|
||||
TcpServer_Destroy(node->server);
|
||||
}
|
||||
|
||||
// Stop maintenance thread (no more discovery ticks after this)
|
||||
if (node->maintenanceRunning) {
|
||||
node->maintenanceRunning = 0;
|
||||
pthread_join(node->maintenanceThread, NULL);
|
||||
}
|
||||
|
||||
// Tear down UDP + discovery. Stop UDP first so no pong/timeout callback races the destroy.
|
||||
if (node->udpNode) {
|
||||
UdpNode_Stop(node->udpNode);
|
||||
@@ -983,6 +1063,7 @@ void Node_Server_OnDisconnect(tcp_connection_t* client) {
|
||||
net_node_t* node = Node_FromConnection(client);
|
||||
Node_ForwardDisconnect(node, client);
|
||||
printf("Inbound node disconnected: %u\n", client ? client->connectionId : 0U);
|
||||
Node_HandlePeerDisconnect(node, client);
|
||||
}
|
||||
|
||||
void Node_Client_OnConnect(tcp_connection_t* client) {
|
||||
@@ -1202,6 +1283,7 @@ void Node_Client_OnDisconnect(tcp_connection_t* client) {
|
||||
|
||||
Node_ForwardDisconnect(node, client);
|
||||
printf("Outbound node disconnected: %u\n", client ? client->connectionId : 0U);
|
||||
Node_HandlePeerDisconnect(node, client);
|
||||
}
|
||||
|
||||
int Node_GetBestOutboundPeer(net_node_t* node, tcp_connection_t** outConn, uint64_t* outHeight) {
|
||||
@@ -1212,13 +1294,16 @@ int Node_GetBestOutboundPeer(net_node_t* node, tcp_connection_t** outConn, uint6
|
||||
|
||||
pthread_mutex_lock(&node->outboundLock);
|
||||
for (size_t i = 0; i < MAX_CONS; ++i) {
|
||||
if (node->outboundClients[i].connection) {
|
||||
if (node->outboundClients[i].peerBlockHeight > bestH || best == NULL) {
|
||||
best = node->outboundClients[i].connection;
|
||||
tcp_connection_t* c = node->outboundClients[i].connection;
|
||||
if (!c || TcpConnection_IsDisconnectNotified(c)) continue; // don't hand out a dead peer
|
||||
if (best == NULL || node->outboundClients[i].peerBlockHeight > bestH) {
|
||||
best = c;
|
||||
bestH = node->outboundClients[i].peerBlockHeight;
|
||||
}
|
||||
}
|
||||
}
|
||||
// Pin the winner while still holding outboundLock so the reaper cannot free it out from under
|
||||
// the caller (which uses the raw pointer after this lock is released). Caller must Unpin.
|
||||
if (best) TcpConnection_Pin(best);
|
||||
pthread_mutex_unlock(&node->outboundLock);
|
||||
|
||||
if (!best) return -1;
|
||||
@@ -1311,8 +1396,9 @@ void Node_GetClientList(net_node_t* node, tcp_connection_t** outClients, size_t*
|
||||
pthread_mutex_lock(&node->outboundLock);
|
||||
size_t count = 0;
|
||||
for (size_t i = 0; i < MAX_CONS; ++i) {
|
||||
if (node->outboundClients[i].connection) {
|
||||
outClients[count++] = node->outboundClients[i].connection;
|
||||
tcp_connection_t* c = node->outboundClients[i].connection;
|
||||
if (c && !TcpConnection_IsDisconnectNotified(c)) { // skip connections that are tearing down
|
||||
outClients[count++] = c;
|
||||
}
|
||||
}
|
||||
pthread_mutex_unlock(&node->outboundLock);
|
||||
|
||||
@@ -274,6 +274,24 @@ void NodeDiscovery_OnPeersReceived(node_discovery_t* disc, tcp_connection_t* fro
|
||||
pthread_mutex_unlock(&disc->lock);
|
||||
}
|
||||
|
||||
void NodeDiscovery_RemovePeer(node_discovery_t* disc, const struct sockaddr_storage* endpoint) {
|
||||
if (!disc || !endpoint) return;
|
||||
pthread_mutex_lock(&disc->lock);
|
||||
size_t n = DynArr_size(disc->peers);
|
||||
for (size_t i = 0; i < n; ++i) {
|
||||
discovered_peer_t* p = (discovered_peer_t*)DynArr_at(disc->peers, i);
|
||||
if (Discovery_AddrEqual(&p->addr, endpoint)) {
|
||||
char ip[INET6_ADDRSTRLEN] = {0};
|
||||
unsigned short port = 0;
|
||||
Discovery_AddrToIpPort(&p->addr, ip, sizeof(ip), &port);
|
||||
printf("NodeDiscovery: struck disconnected peer %s:%u from peer list\n", ip, port);
|
||||
DynArr_remove(disc->peers, i);
|
||||
break;
|
||||
}
|
||||
}
|
||||
pthread_mutex_unlock(&disc->lock);
|
||||
}
|
||||
|
||||
// ---- periodic tick -----------------------------------------------------------------------
|
||||
|
||||
void NodeDiscovery_Iterate(node_discovery_t* disc) {
|
||||
|
||||
@@ -31,6 +31,7 @@ int TcpConnection_Init(tcp_connection_t* conn, int sockFd, const struct sockaddr
|
||||
|
||||
conn->closing = false;
|
||||
conn->disconnectedNotified = false;
|
||||
atomic_init(&conn->pinCount, 0);
|
||||
conn->dataBuf = NULL;
|
||||
conn->dataBufLen = 0;
|
||||
conn->dataBufCap = 0;
|
||||
@@ -262,6 +263,20 @@ bool TcpConnection_IsDisconnectNotified(tcp_connection_t* conn) {
|
||||
return notified;
|
||||
}
|
||||
|
||||
void TcpConnection_Pin(tcp_connection_t* conn) {
|
||||
if (!conn) {
|
||||
return;
|
||||
}
|
||||
atomic_fetch_add(&conn->pinCount, 1);
|
||||
}
|
||||
|
||||
void TcpConnection_Unpin(tcp_connection_t* conn) {
|
||||
if (!conn) {
|
||||
return;
|
||||
}
|
||||
atomic_fetch_sub(&conn->pinCount, 1);
|
||||
}
|
||||
|
||||
static int extract_v4(const tcp_connection_t* conn, struct in_addr* v4out) {
|
||||
if (conn->addrFamily == AF_INET6) {
|
||||
const struct sockaddr_in6* a6 = (const struct sockaddr_in6*)&conn->peerAddr;
|
||||
|
||||
Reference in New Issue
Block a user