1 files changed, 117 insertions, 109 deletions

diff --git a/src/net.cpp b/src/net.cpp
index c24547e..bb39122 100644
--- a/src/net.cpp
+++ b/src/net.cpp
@@ -464,28 +464,9 @@ bool GetMyExternalIP(unsigned int& ipRet)
 
 
 
-void AbandonRequests(void (*fn)(void*, CDataStream&), void* param1)
-{
-    // If the dialog might get closed before the reply comes back,
-    // call this in the destructor so it doesn't get called after it's deleted.
-    CRITICAL_BLOCK(cs_vNodes)
-    {
-        foreach(CNode* pnode, vNodes)
-        {
-            CRITICAL_BLOCK(pnode->cs_mapRequests)
-            {
-                for (map<uint256, CRequestTracker>::iterator mi = pnode->mapRequests.begin(); mi != pnode->mapRequests.end();)
-                {
-                    CRequestTracker& tracker = (*mi).second;
-                    if (tracker.fn == fn && tracker.param1 == param1)
-                        pnode->mapRequests.erase(mi++);
-                    else
-                        mi++;
-                }
-            }
-        }
-    }
-}
+// AbandonRequests() lived here, sweeping CNode::mapRequests for a callback a
+// closing dialog had registered. Nothing ever registered one and nothing ever
+// called this, so it went with CRequestTracker.
 
 
 
@@ -493,25 +474,10 @@ void AbandonRequests(void (*fn)(void*, CDataStream&), void* param1)
 
 
 
-//
-// Subscription methods for the broadcast and subscription system.
-// Channel numbers are message numbers, i.e. MSG_TABLE and MSG_PRODUCT.
-//
-// The subscription system uses a meet-in-the-middle strategy.
-// With 100,000 nodes, if senders broadcast to 1000 random nodes and receivers
-// subscribe to 1000 random nodes, 99.995% (1 - 0.99^1000) of messages will get through.
-//
-
-bool AnySubscribed(unsigned int nChannel)
-{
-    if (pnodeLocalHost->IsSubscribed(nChannel))
-        return true;
-    CRITICAL_BLOCK(cs_vNodes)
-        foreach(CNode* pnode, vNodes)
-            if (pnode->IsSubscribed(nChannel))
-                return true;
-    return false;
-}
+// The broadcast/subscribe channel system (AnySubscribed, CNode::IsSubscribed,
+// Subscribe, CancelSubscribe) stood here. Its channels were MSG_TABLE and
+// MSG_PRODUCT, there was never an inbound handler for "subscribe",
+// "sub-cancel", "publish" or "pub-cancel", and the objects it carried are gone.
 
 void CNode::PushGetBlocks(CBlockIndex* pindexBegin, uint256 hashEnd)
 {
@@ -524,55 +490,6 @@ void CNode::PushGetBlocks(CBlockIndex* pindexBegin, uint256 hashEnd)
     PushMessage("getblocks", CBlockLocator(pindexBegin), hashEnd);
 }
 
-bool CNode::IsSubscribed(unsigned int nChannel)
-{
-    if (nChannel >= vfSubscribe.size())
-        return false;
-    return vfSubscribe[nChannel];
-}
-
-void CNode::Subscribe(unsigned int nChannel, unsigned int nHops)
-{
-    if (nChannel >= vfSubscribe.size())
-        return;
-
-    if (!AnySubscribed(nChannel))
-    {
-        // Relay subscribe
-        CRITICAL_BLOCK(cs_vNodes)
-            foreach(CNode* pnode, vNodes)
-                if (pnode != this)
-                    pnode->PushMessage("subscribe", nChannel, nHops);
-    }
-
-    vfSubscribe[nChannel] = true;
-}
-
-void CNode::CancelSubscribe(unsigned int nChannel)
-{
-    if (nChannel >= vfSubscribe.size())
-        return;
-
-    // Prevent from relaying cancel if wasn't subscribed
-    if (!vfSubscribe[nChannel])
-        return;
-    vfSubscribe[nChannel] = false;
-
-    if (!AnySubscribed(nChannel))
-    {
-        // Relay subscription cancel
-        CRITICAL_BLOCK(cs_vNodes)
-            foreach(CNode* pnode, vNodes)
-                if (pnode != this)
-                    pnode->PushMessage("sub-cancel", nChannel);
-
-        // Clear memory, no longer subscribed
-        if (nChannel == MSG_PRODUCT)
-            CRITICAL_BLOCK(cs_mapProducts)
-                mapProducts.clear();
-    }
-}
-
 
 
 
@@ -1074,6 +991,18 @@ CNode* ConnectNodeBtfResolved(const string& strBtfAddr, const string& strMeeting
 void ThreadBtfAccept(void* parg)
 {
     printf("ThreadBtfAccept started\n");
+    // In operator mode the pool server registers this same node identity as a
+    // stratum service at the rendezvous relays, and a relay only holds ONE
+    // service registration per pubkey -- peer-accept and pool-accept would
+    // kick each other off forever, breaking both. The pool's registration is
+    // the one that must be stable (that's what miners resolve and dial), so
+    // the node skips inbound rendezvous here and relies on outbound .btf
+    // connections for block sync.
+    if (nMineMode == MINE_OPERATOR)
+    {
+        printf("ThreadBtfAccept: operator mode -- skipping peer rendezvous registration\n");
+        return;
+    }
     unsigned char pk[32];
     unsigned char sk[32];
     while (!BtfGetIdentity(pk, sk))
@@ -1223,17 +1152,6 @@ void CNode::Disconnect()
     BtfCloseSocket(hSocket);
     hSocket = INVALID_SOCKET;
 
-    // All of a nodes broadcasts and subscriptions are automatically torn down
-    // when it goes down, so a node has to stay up to keep its broadcast going.
-
-    CRITICAL_BLOCK(cs_mapProducts)
-        for (map<uint256, CProduct>::iterator mi = mapProducts.begin(); mi != mapProducts.end();)
-            AdvertRemoveSource(this, MSG_PRODUCT, 0, (*(mi++)).second);
-
-    // Cancel subscriptions
-    for (unsigned int nChannel = 0; nChannel < vfSubscribe.size(); nChannel++)
-        if (vfSubscribe[nChannel])
-            CancelSubscribe(nChannel);
 }
 
 
@@ -1358,7 +1276,6 @@ void ThreadSocketHandler2(void* parg)
                     bool fDelete = false;
                     TRY_CRITICAL_BLOCK(pnode->cs_vSend)
                      TRY_CRITICAL_BLOCK(pnode->cs_vRecv)
-                      TRY_CRITICAL_BLOCK(pnode->cs_mapRequests)
                        TRY_CRITICAL_BLOCK(pnode->cs_inventory)
                         fDelete = true;
                     if (fDelete)
@@ -1466,6 +1383,27 @@ void ThreadSocketHandler2(void* parg)
                         printf("dropping node %s: its socket is no longer valid\n",
                                pnode->addr.ToString().c_str());
                         pnode->fDisconnect = true;
+
+                        // Stop referring to it, or it goes into the next poll
+                        // set and fails it again.
+                        //
+                        // Marking fDisconnect is not enough: a node is only
+                        // removed from vNodes once it has drained or once
+                        // DISCONNECT_DRAIN_SECS has passed, so one dead
+                        // descriptor kept failing the wait for a full minute --
+                        // and a failed wait skips the service loop below
+                        // entirely, which means NO peer is read on that pass.
+                        // Measured on a wedged pool under the old select()
+                        // implementation: 1,027 consecutive passes over 61
+                        // seconds, one bad socket, the entire node deaf. poll()
+                        // reports a bad descriptor as POLLNVAL on that entry
+                        // rather than failing the whole call, so this path is
+                        // now the backstop rather than the common case.
+                        //
+                        // Not closed, only forgotten. getsockopt just told us
+                        // this descriptor is not ours any more; closing it now
+                        // is how an unrelated socket gets closed instead.
+                        pnode->hSocket = INVALID_SOCKET;
                     }
                 }
             }
@@ -1710,6 +1648,13 @@ void ThreadMessageHandler(void* parg)
     }
 }
 
+// When any peer last told us a height. A wedged node -- one that still has
+// peers but has stopped hearing them -- is exactly a node where this stops
+// advancing, and that is the one thing the height comparison itself cannot
+// see: there is no height to compare.
+static int64 nLastPeerHeightSeen = 0;
+static int64 nPeerHeightFirstAsked = 0;
+
 int GetPeerMedianHeight()
 {
     vector<int> vHeights;
@@ -1721,10 +1666,26 @@ int GetPeerMedianHeight()
     if (vHeights.empty())
         return -1;
 
+    nLastPeerHeightSeen = GetTime();
     sort(vHeights.begin(), vHeights.end());
     return vHeights[vHeights.size() / 2];
 }
 
+// Seconds since any peer last reported a height.
+//
+// Before ever hearing one, measure from the first time somebody asked rather
+// than from the epoch -- otherwise a node refuses jobs the instant it starts,
+// before it has had a chance to complete a single handshake.
+int64 GetPeerHeightAgeSecs()
+{
+    int64 nNow = GetTime();
+    if (nPeerHeightFirstAsked == 0)
+        nPeerHeightFirstAsked = nNow;
+    if (nLastPeerHeightSeen == 0)
+        return nNow - nPeerHeightFirstAsked;
+    return nNow - nLastPeerHeightSeen;
+}
+
 
 // Say out loud when we are behind the network.
 //
@@ -1861,7 +1822,21 @@ void ThreadBitcoinMiner(void* parg)
 // Start the miners. Returns how many actually got off the ground.
 int StartMinerThreads()
 {
-    int nThreads = MinerThreadCount();
+    // MinerThreadCount() sizes the SOLO miner, where each thread hashes a
+    // template of its own and more threads is simply more hashrate.
+    //
+    // Participant mode is not that shape. Every thread that enters
+    // BitcoinMiner() takes the MINE_PARTICIPANT branch, and that branch is a
+    // whole miner: its own Stratum session to the pool, its own payout address
+    // from the rotation, and its own pool of /hashthreads worker threads. So
+    // the two counts multiply instead of dividing, and the machine ends up
+    // running MinerThreadCount() x nHashThreads hashing threads plus one pool
+    // connection each. On a 12-core host with /hashthreads=2 that is 11
+    // sessions and 22 hash threads -- roughly 2x the cores, which is not more
+    // work done, only more contention -- and with /hashthreads unset it is 11
+    // sessions of 11 workers. One session is the entirety of participant mode;
+    // /hashthreads is what sizes it, and it can then mean what it says.
+    int nThreads = (nMineMode == MINE_PARTICIPANT) ? 1 : MinerThreadCount();
     int nStarted = 0;
     for (int i = 0; i < nThreads; i++)
     {
@@ -1873,9 +1848,24 @@ int StartMinerThreads()
     // The dataset is one allocation shared by all of them; only the 2 MB
     // scratchpads multiply. Printed because "why is it using 2 GB" is the first
     // question anyone asks.
-    printf("Mining with %d thread(s) of %u core(s) -- about %d MB "
-           "(~2080 MB shared RandomX dataset + ~2 MB per thread)\n",
-           nStarted, std::thread::hardware_concurrency(), 2080 + 2 * nStarted);
+    //
+    // Participant mode reports the workers that session will run, not the one
+    // thread that runs it: the hash thread count is the number an operator
+    // watches, and it is the number that was wrong here. It is otherwise only
+    // logged under -debug, which is how 11 sessions ran unnoticed.
+    if (nMineMode == MINE_PARTICIPANT)
+    {
+        int nWorkers = ParticipantHashThreadCount();
+        printf("Mining for the pool with %d hash thread(s) of %u core(s) -- about %d MB "
+               "(~2080 MB shared RandomX dataset + ~2 MB per thread)\n",
+               nWorkers, AvailableCoreCount(), 2080 + 2 * nWorkers);
+    }
+    else
+    {
+        printf("Mining with %d thread(s) of %u core(s) -- about %d MB "
+               "(~2080 MB shared RandomX dataset + ~2 MB per thread)\n",
+               nStarted, AvailableCoreCount(), 2080 + 2 * nStarted);
+    }
     return nStarted;
 }
 
@@ -1950,11 +1940,29 @@ bool StartNode(string& strError)
     {
         int nErr = WSAGetLastError();
         if (nErr == WSAEADDRINUSE)
-            strError = strprintf("Error: Unable to bind to port %s on this computer. The program is probably already running.", addrLocalHost.ToString().c_str());
+        {
+            // Another local instance holds the default port. Fall back to an
+            // ephemeral port: inbound direct-IP peering is optional, and .btf
+            // rendezvous connectivity is unaffected by the local listen port.
+            sockaddr.sin_port = htons(0);
+            if (bind(hListenSocket, (struct sockaddr*)&sockaddr, sizeof(sockaddr)) == SOCKET_ERROR)
+            {
+                strError = strprintf("Error: Unable to bind any port on this computer (bind returned error %d)", WSAGetLastError());
+                printf("%s\n", strError.c_str());
+                return false;
+            }
+            struct sockaddr_in bound;
+            socklen_t nBoundLen = sizeof(bound);
+            if (getsockname(hListenSocket, (struct sockaddr*)&bound, &nBoundLen) == 0)
+                addrLocalHost.port = bound.sin_port;
+            printf("default port in use by another instance; bound to ephemeral %s instead\n", addrLocalHost.ToString().c_str());
+        }
         else
+        {
             strError = strprintf("Error: Unable to bind to port %s on this computer (bind returned error %d)", addrLocalHost.ToString().c_str(), nErr);
-        printf("%s\n", strError.c_str());
-        return false;
+            printf("%s\n", strError.c_str());
+            return false;
+        }
     }
     printf("bound to addrLocalHost = %s\n\n", addrLocalHost.ToString().c_str());