1 files changed, 237 insertions, 53 deletions

diff --git a/src/rpc.cpp b/src/rpc.cpp
index be265db..403cb14 100644
--- a/src/rpc.cpp
+++ b/src/rpc.cpp
@@ -451,9 +451,20 @@ static std::string PoolRoundsFilePath()
 // number; publishing happens with the data mutex released and refuses any
 // blob whose sequence number is not newer than what is already on disk. So
 // writers can queue in any order and finish in any order -- the file only
-// ever moves forwards. Crucially, no thread ever waits for I/O while holding
-// gPayoutLedgerMutex / gPayoutExecMutex / gPoolRoundsMutex, which is the
-// contention the merge was right to remove from the event loop.
+// ever moves forwards.
+//
+// The other half of the rule is that no thread waits for I/O while holding
+// gPayoutLedgerMutex, gPayoutExecMutex or gPoolRoundsMutex, which is the
+// contention the merge was right to remove from the event loop. That was
+// claimed here before it was true: pool_status.json is written by the same
+// WriteFileAtomic and RefreshStats held gPayoutLedgerMutex at function scope
+// across the call, so once the writes gained fsync() the event loop was doing
+// two of them every ten seconds with the payout thread's mutex held -- worse
+// than the buffered write the merge replaced. The status file now goes through
+// the same serialise-under-the-lock / publish-outside-it split as the ledgers,
+// and the rule holds for all four files. It is checked, not asserted: see the
+// "no payout mutex is held across a publish" section of the payout-ledger
+// self-test.
 
 // A serialised file image plus the version of the in-memory state it was
 // taken from.
@@ -2100,7 +2111,22 @@ static std::string PoolStatusFilePath()
     return GetAppDir() + "/pool_status.json";
 }
 
-static void WritePoolStatusJsonLocked()
+// Serialise the operator status file. Caller must hold gStatsMutex (this reads
+// gStats / gWorkerStats / gPayoutViews); gPoolRoundsMutex is taken and
+// released inside.
+//
+// Split from the write for the same reason as the three ledgers, and for a
+// worse instance of the same problem. This used to be called from RefreshStats
+// with gPayoutLedgerMutex held at function scope, and once the ledger writes
+// gained fsync() that call became two fsyncs -- one on the file, one on its
+// directory -- performed by the event loop, the thread that owns every miner's
+// socket, with the mutex the payout thread needs in its hand. Measured on the
+// build host (ext4 on LVM on SSD): 3.78 ms per publish, every 10 seconds, and
+// unbounded on a shared virtual disk under load.
+//
+// Nothing here needs the ledger mutex at all: the pending-payout rows were
+// already copied into gPayoutViews under it, further up RefreshStats.
+static std::string SerializePoolStatusLocked()
 {
     nlohmann::json j;
     j["schema"] = "bitflash-pool-status-1";
@@ -2184,11 +2210,20 @@ static void WritePoolStatusJsonLocked()
         }
     }
 
-    // No version guard and no write mutex here, deliberately: RefreshStats is
-    // the only caller and it only ever runs on the event loop, so there is a
-    // single writer of pool_status.json by construction. It shares
-    // WriteFileAtomic so the durability rules stay in one place.
-    WriteFileAtomic(PoolStatusFilePath(), j.dump(2), "pool status");
+    return j.dump(2);
+}
+
+// Publish the operator status file. Call with every payout mutex released.
+//
+// Still no version guard and no write mutex, deliberately, and the split does
+// not change the argument: RefreshStats remains the only caller, it still runs
+// only on the event loop, and it now calls this after leaving the scope that
+// serialised -- so the serialise and the write stay strictly ordered on one
+// thread and pool_status.json still has a single writer by construction. It
+// shares WriteFileAtomic so the durability rules stay in one place.
+static void PublishPoolStatus(const std::string& blob)
+{
+    WriteFileAtomic(PoolStatusFilePath(), blob, "pool status");
 }
 
 // ---------------------------------------------------------------------------
@@ -2907,6 +2942,10 @@ static std::string PayoutTestAddWalletTx(const std::string& comment,
     return hash.GetHex();
 }
 
+// Defined below with the event loop; the status-file publication test drives
+// it directly.
+static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal);
+
 int RunPayoutLedgerSelfTest()
 {
     int nFail = 0;
@@ -3455,6 +3494,136 @@ int RunPayoutLedgerSelfTest()
         strPoolRoundsFile = savedRoundsFile;
     }
 
+    // -- 11. No payout mutex is held across the status-file write ------------
+    //
+    // RefreshStats runs on the event loop -- the thread that owns every
+    // miner's socket -- and used to hold gStatsMutex and gPayoutLedgerMutex at
+    // function scope across WriteFileAtomic. Once that write gained fsync() on
+    // the file and on its directory, publishing pool_status.json parked the
+    // payout thread behind two disk flushes every ten seconds.
+    //
+    // Timing assertions would be flaky, so this does not measure anything. It
+    // makes the write block for as long as the test likes: strPoolStatusFile
+    // points at a path whose ".tmp" is a FIFO, and open(2) for writing on a
+    // FIFO does not return until a reader arrives. While RefreshStats is
+    // parked in there, every mutex it must NOT be holding has to be
+    // acquirable. Against the previous code this test does not merely fail, it
+    // cannot pass: the writer is inside fopen() with both mutexes held and
+    // nothing will release them until the test lets the write finish.
+#ifndef _WIN32
+    {
+        PayoutTestResetState();
+        const std::string statusPath = dir + "/pool_status_selftest.json";
+        const std::string statusTmp  = statusPath + ".tmp";
+        const std::string savedStatusFile = strPoolStatusFile;
+        const int64       savedLastWrite  = gLastPoolStatusWrite;
+        remove(statusPath.c_str());
+        remove(statusTmp.c_str());
+        strPoolStatusFile    = statusPath;
+        gLastPoolStatusWrite = 0;   // due a publish on the next refresh
+
+        // Give the status file something to report, so the ledger mutex is
+        // genuinely taken on the way through rather than skipped over.
+        {
+            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
+            PendingPayout pp;
+            pp.blockHeight = 909; pp.blockHash = 0; pp.matureAtHeight = 909;
+            pp.amounts[addrA] = 3 * COIN;
+            gPendingPayouts.push_back(pp);
+        }
+
+        bool fifoOk = (mkfifo(statusTmp.c_str(), 0600) == 0);
+        check(fifoOk, "a FIFO stands in for the status file's temp file, so the"
+                      " write can be held open");
+        if (fifoOk) {
+            std::atomic<bool> done(false);
+            std::thread writer([&done]() {
+                RefreshStats(0, 0);
+                done = true;
+            });
+
+            // Settle: the work before the write is a handful of vector
+            // operations, so anything past this point is inside fopen(). The
+            // !done check below is what actually proves it -- with no reader
+            // there is no other way for this thread to still be running.
+            Sleep(500);
+            bool blocked = !done;
+            bool gotLedger = false, gotStats = false, gotExec = false,
+                 gotRounds = false;
+            for (int i = 0; i < 1000 && blocked; i++) {
+                if (!gotLedger && gPayoutLedgerMutex.try_lock()) {
+                    gotLedger = true; gPayoutLedgerMutex.unlock();
+                }
+                if (!gotStats && gStatsMutex.try_lock()) {
+                    gotStats = true; gStatsMutex.unlock();
+                }
+                if (!gotExec && gPayoutExecMutex.try_lock()) {
+                    gotExec = true; gPayoutExecMutex.unlock();
+                }
+                if (!gotRounds && gPoolRoundsMutex.try_lock()) {
+                    gotRounds = true; gPoolRoundsMutex.unlock();
+                }
+                if (gotLedger && gotStats && gotExec && gotRounds)
+                    break;
+                Sleep(2);
+            }
+
+            check(blocked,
+                  "the status-file write is held open for the duration of this"
+                  " test");
+            check(!blocked || gotLedger,
+                  "gPayoutLedgerMutex is free while pool_status.json is being"
+                  " written");
+            check(!blocked || (gotExec && gotRounds),
+                  "gPayoutExecMutex and gPoolRoundsMutex are free while"
+                  " pool_status.json is being written");
+            check(!blocked || gotStats,
+                  "gStatsMutex is free while pool_status.json is being written");
+
+            // Let the write finish. Drain to EOF rather than closing early:
+            // the writer is still filling the pipe and a closed read end would
+            // deliver it SIGPIPE.
+            std::thread reader([&statusTmp]() {
+                FILE* rf = fopen(statusTmp.c_str(), "r");
+                if (!rf) return;
+                char buf[4096];
+                while (fread(buf, 1, sizeof(buf), rf) > 0) { }
+                fclose(rf);
+            });
+            writer.join();
+            reader.join();
+        }
+        remove(statusTmp.c_str());
+
+        // And the ordinary path still produces the file it always did.
+        gLastPoolStatusWrite = 0;
+        RefreshStats(0, 0);
+        std::string onDisk;
+        bool statusOk = false;
+        if (PayoutTestReadFile(statusPath, onDisk)) {
+            try {
+                nlohmann::json j = nlohmann::json::parse(onDisk);
+                statusOk = j.value("schema", "") == "bitflash-pool-status-1" &&
+                           j["pendingPayouts"].is_array() &&
+                           j["pendingPayouts"].size() == 1 &&
+                           j["pendingPayouts"][0].value("matureAtHeight", 0) == 909 &&
+                           j["pendingPayouts"][0].value("totalAmountSatoshis",
+                                                        (int64)0) == 3 * COIN;
+            } catch (...) {
+                statusOk = false;
+            }
+        }
+        check(statusOk,
+              "splitting the serialise from the write leaves pool_status.json"
+              " saying the same thing");
+
+        remove(statusPath.c_str());
+        remove((statusPath + ".tmp").c_str());
+        strPoolStatusFile    = savedStatusFile;
+        gLastPoolStatusWrite = savedLastWrite;
+    }
+#endif
+
     // -- clean up ------------------------------------------------------------
     PayoutTestResetState();
     CRITICAL_BLOCK(cs_mapWallet)
@@ -3615,56 +3784,71 @@ int RunPoolStratumSelfTest()
 // ---------------------------------------------------------------------------
 static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal)
 {
-    std::lock_guard<std::mutex> lk(gStatsMutex);
+    std::string statusBlob;
+    bool        fPublish = false;
+    {
+        std::lock_guard<std::mutex> lk(gStatsMutex);
 
-    gStats.authorizedMiners = 0;
-    gStats.blocksFound      = blocksFoundThisSession;
-    gStats.roundShares      = roundShareTotal;
-    gStats.totalHashRate    = 0.0;
-    for (Miner* m : gMiners)
-        if (m->authorised) gStats.authorizedMiners++;
+        gStats.authorizedMiners = 0;
+        gStats.blocksFound      = blocksFoundThisSession;
+        gStats.roundShares      = roundShareTotal;
+        gStats.totalHashRate    = 0.0;
+        for (Miner* m : gMiners)
+            if (m->authorised) gStats.authorizedMiners++;
 
-    gWorkerStats.clear();
-    for (Miner* m : gMiners) {
-        if (!m->authorised || m->lastSeen <= 0) continue;
-        PoolWorkerStatView v;
-        v.address     = m->address;
-        v.worker      = m->worker;
-        v.roundShares = m->roundShares;
-        v.totalShares = m->totalShares;
-        v.lastSeen    = m->lastSeen;
-        v.hashRate    = m->hashRate;
-        gWorkerStats.push_back(v);
-        gStats.totalHashRate += m->hashRate;
-    }
-    std::sort(gWorkerStats.begin(), gWorkerStats.end(),
-        [](const PoolWorkerStatView& a, const PoolWorkerStatView& b) {
-            return a.roundShares != b.roundShares ? a.roundShares > b.roundShares
-                                                  : a.lastSeen > b.lastSeen;
-        });
+        gWorkerStats.clear();
+        for (Miner* m : gMiners) {
+            if (!m->authorised || m->lastSeen <= 0) continue;
+            PoolWorkerStatView v;
+            v.address     = m->address;
+            v.worker      = m->worker;
+            v.roundShares = m->roundShares;
+            v.totalShares = m->totalShares;
+            v.lastSeen    = m->lastSeen;
+            v.hashRate    = m->hashRate;
+            gWorkerStats.push_back(v);
+            gStats.totalHashRate += m->hashRate;
+        }
+        std::sort(gWorkerStats.begin(), gWorkerStats.end(),
+            [](const PoolWorkerStatView& a, const PoolWorkerStatView& b) {
+                return a.roundShares != b.roundShares ? a.roundShares > b.roundShares
+                                                      : a.lastSeen > b.lastSeen;
+            });
 
-    gPayoutViews.clear();
-    std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
-    for (const PendingPayout& pp : gPendingPayouts) {
-        PendingPayoutView pv;
-        pv.matureAtHeight = pp.matureAtHeight;
-        pv.recipients     = 0;
-        pv.totalAmount    = 0;
-        for (auto& kv : pp.amounts) {
-            if (pp.paidTxids.count(kv.first))
-                continue;
-            pv.recipients++;
-            pv.totalAmount += kv.second;
+        // Scoped, not held to the end of the function. At function scope this
+        // lock covered the status-file write below, so the event loop did two
+        // fsyncs holding the mutex the payout thread takes to record that a
+        // miner has been paid.
+        gPayoutViews.clear();
+        {
+            std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
+            for (const PendingPayout& pp : gPendingPayouts) {
+                PendingPayoutView pv;
+                pv.matureAtHeight = pp.matureAtHeight;
+                pv.recipients     = 0;
+                pv.totalAmount    = 0;
+                for (auto& kv : pp.amounts) {
+                    if (pp.paidTxids.count(kv.first))
+                        continue;
+                    pv.recipients++;
+                    pv.totalAmount += kv.second;
+                }
+                if (pv.recipients > 0)
+                    gPayoutViews.push_back(pv);
+            }
         }
-        if (pv.recipients > 0)
-            gPayoutViews.push_back(pv);
-    }
 
-    int64 now = GetTime();
-    if (now - gLastPoolStatusWrite >= 10) {
-        WritePoolStatusJsonLocked();
-        gLastPoolStatusWrite = now;
+        int64 now = GetTime();
+        if (now - gLastPoolStatusWrite >= 10) {
+            statusBlob = SerializePoolStatusLocked();
+            gLastPoolStatusWrite = now;
+            fPublish = true;
+        }
     }
+    // Every payout mutex is released here, and gStatsMutex with them. This is
+    // where the two fsyncs happen, and no other thread is waiting behind them.
+    if (fPublish)
+        PublishPoolStatus(statusBlob);
 }
 
 // ---------------------------------------------------------------------------