|
| 1 | +#include "node.hpp" |
| 2 | +#include "share.hpp" |
| 3 | + |
| 4 | +#include <core/uint256.hpp> |
| 5 | +#include <core/common.hpp> |
| 6 | + |
| 7 | +#include <boost/asio/post.hpp> |
| 8 | + |
| 9 | +#include <atomic> |
| 10 | +#include <memory> |
| 11 | +#include <vector> |
| 12 | + |
| 13 | +// Dash p2pool sharechain pool-node — S8 pool-node reception, slice .4 (bodies). |
| 14 | +// |
| 15 | +// SLICE SCOPE (integrator-confirmed 2026-07-09, #656 a1444b4 branch-tip (awaiting merge tap)): the two |
| 16 | +// link-deferred declarations from node.hpp — processing_shares() and |
| 17 | +// handle_get_share() — get their REDUCED dash-native bodies here. "Reduced" |
| 18 | +// means: the parallel X11 share_init_verify -> try_lock + m_tracker.add core, |
| 19 | +// with think()/download/best-share/persist DEFERRED to their own later slices. |
| 20 | +// |
| 21 | +// This is a deliberate, non-1:1 port of src/impl/ltc/node.cpp. The DASH node is |
| 22 | +// header-only and does NOT carry the ~8 LTC-only symbols the full ltc bodies |
| 23 | +// touch (run_think/think, chain::PreparedList, m_raw_share_cache, |
| 24 | +// m_best_share_hash, chain::get_reverse fork-logging, m_rejected_share_hashes, |
| 25 | +// processing_shares_phase2 as a distinct TU method). Those are intentionally |
| 26 | +// absent — force-fitting them would be scope creep. What remains is the minimal |
| 27 | +// admit path that makes the Legacy/Actual dispatch (protocol_{legacy,actual}.cpp) |
| 28 | +// actually reach the tracker: verify in parallel off the io_context, then insert |
| 29 | +// under the non-blocking tracker lock on the io_context thread. |
| 30 | + |
| 31 | +namespace dash |
| 32 | +{ |
| 33 | + |
| 34 | +void NodeImpl::processing_shares(HandleSharesData& data_ref, NetService addr) |
| 35 | +{ |
| 36 | + // Take ownership immediately so the caller (the dispatch handler) can return |
| 37 | + // and free its local HandleSharesData. |
| 38 | + auto data = std::make_shared<HandleSharesData>(std::move(data_ref)); |
| 39 | + size_t n = data->m_items.size(); |
| 40 | + if (n == 0) |
| 41 | + return; |
| 42 | + |
| 43 | + // ── Phase 1 (m_verify_pool, parallel) ──────────────────────────────── |
| 44 | + // Run share_init_verify() for each received share off the io_context thread. |
| 45 | + // DASH is X11 (~ms of CPU per share); the structural + PoW verify must NOT |
| 46 | + // block network I/O. Each share's hash computation is independent, so the |
| 47 | + // batch fully parallelises across the pool. A share that throws is left with |
| 48 | + // a null hash and skipped in phase 2. |
| 49 | + auto remaining = std::make_shared<std::atomic<int>>(static_cast<int>(n)); |
| 50 | + for (size_t i = 0; i < n; i++) |
| 51 | + { |
| 52 | + boost::asio::post(m_verify_pool, |
| 53 | + [i, data, remaining, this, addr]() |
| 54 | + { |
| 55 | + auto& share = data->m_items[i]; |
| 56 | + if (share.hash().IsNull()) |
| 57 | + { |
| 58 | + try |
| 59 | + { |
| 60 | + share.ACTION({ |
| 61 | + obj->m_hash = share_init_verify(*obj, m_tracker.m_coin_params, true); |
| 62 | + }); |
| 63 | + } |
| 64 | + catch (const std::exception&) |
| 65 | + { |
| 66 | + // leave hash null — phase 2 will skip this share |
| 67 | + } |
| 68 | + } |
| 69 | + |
| 70 | + // Last verify done → hop back to the io_context thread for the |
| 71 | + // tracker insertion (writes stay single-threaded on io_context). |
| 72 | + if (--(*remaining) == 0) |
| 73 | + { |
| 74 | + boost::asio::post(*m_context, |
| 75 | + [data, this, addr]() |
| 76 | + { |
| 77 | + add_verified_shares(*data, addr); |
| 78 | + }); |
| 79 | + } |
| 80 | + }); |
| 81 | + } |
| 82 | +} |
| 83 | + |
| 84 | +void NodeImpl::add_verified_shares(HandleSharesData& data, NetService addr) |
| 85 | +{ |
| 86 | + // io_context thread. Non-blocking tracker lock (architectural rule, |
| 87 | + // node.hpp): the compute thread (a later slice's think()) takes the |
| 88 | + // exclusive lock; the IO thread must NEVER block on it. If think() holds the |
| 89 | + // lock, defer this batch onto m_pending_adds — the compute thread drains it |
| 90 | + // after releasing. Until the think() slice lands nothing takes the exclusive |
| 91 | + // lock, so try_to_lock always succeeds here. |
| 92 | + { |
| 93 | + std::unique_lock lock(m_tracker_mutex, std::try_to_lock); |
| 94 | + if (!lock.owns_lock()) |
| 95 | + { |
| 96 | + if (m_pending_adds.size() < MAX_PENDING_ADDS) |
| 97 | + { |
| 98 | + LOG_INFO << "[ASYNC-DEFER] add_verified_shares: tracker busy, queuing " |
| 99 | + << data.m_items.size() << " shares from " << addr.to_string() |
| 100 | + << " (pending=" << m_pending_adds.size() + 1 << ")"; |
| 101 | + m_pending_adds.push_back(PendingShareBatch{ |
| 102 | + std::make_unique<HandleSharesData>(std::move(data)), addr}); |
| 103 | + } |
| 104 | + else |
| 105 | + { |
| 106 | + LOG_WARNING << "[ASYNC-DEFER] add_verified_shares: pending queue full (" |
| 107 | + << MAX_PENDING_ADDS << ") — dropping batch from " |
| 108 | + << addr.to_string(); |
| 109 | + } |
| 110 | + return; |
| 111 | + } |
| 112 | + // Lock held for the whole insertion below. |
| 113 | + |
| 114 | + int32_t new_count = 0; |
| 115 | + int32_t dup_count = 0; |
| 116 | + for (auto& share : data.m_items) |
| 117 | + { |
| 118 | + // Skip shares that failed phase-1 verification (hash still null). |
| 119 | + if (share.hash().IsNull()) |
| 120 | + continue; |
| 121 | + |
| 122 | + if (m_chain->contains(share.hash())) |
| 123 | + { |
| 124 | + ++dup_count; |
| 125 | + continue; |
| 126 | + } |
| 127 | + |
| 128 | + m_tracker.add(share); |
| 129 | + ++new_count; |
| 130 | + } |
| 131 | + |
| 132 | + if (new_count > 0) |
| 133 | + { |
| 134 | + auto as = addr.to_string(); |
| 135 | + std::string source = (addr.port() == 0) ? as.substr(0, as.rfind(':')) : as; |
| 136 | + LOG_INFO << "Processing " << new_count << " shares from " << source |
| 137 | + << "... (dup=" << dup_count |
| 138 | + << " chain=" << m_tracker.chain.size() << ")"; |
| 139 | + } |
| 140 | + } |
| 141 | + // Scoring / best-share selection / persist / relay ride the think() and |
| 142 | + // broadcast slices; slice .4 stops at tracker insertion. |
| 143 | +} |
| 144 | + |
| 145 | +std::vector<dash::ShareType> |
| 146 | +NodeImpl::handle_get_share(std::vector<uint256> hashes, uint64_t parents, |
| 147 | + std::vector<uint256> stops, NetService peer_addr) |
| 148 | +{ |
| 149 | + // try_to_lock per the architectural rule (node.hpp): the IO thread MUST |
| 150 | + // never block on m_tracker_mutex. An empty reply does not disconnect the |
| 151 | + // requesting peer — p2pool's downloader retries against another random peer, |
| 152 | + // so a busy-skip just shifts the request one iteration. |
| 153 | + std::shared_lock<std::shared_mutex> lock(m_tracker_mutex, std::try_to_lock); |
| 154 | + if (!lock.owns_lock()) |
| 155 | + { |
| 156 | + static int defer_log = 0; |
| 157 | + if (defer_log++ % 50 == 0) |
| 158 | + LOG_INFO << "[handle_get_share] tracker busy — returning empty to " |
| 159 | + << peer_addr.to_string() |
| 160 | + << " (peer will retry against another peer)"; |
| 161 | + return {}; |
| 162 | + } |
| 163 | + |
| 164 | + if (hashes.empty()) |
| 165 | + return {}; |
| 166 | + |
| 167 | + parents = std::min(parents, (uint64_t)1000 / hashes.size()); |
| 168 | + std::vector<dash::ShareType> shares; |
| 169 | + for (const auto& handle_hash : hashes) |
| 170 | + { |
| 171 | + if (!m_chain->contains(handle_hash)) |
| 172 | + { |
| 173 | + static int miss_log = 0; |
| 174 | + if (miss_log++ < 5) |
| 175 | + LOG_WARNING << "[handle_get_share] hash NOT in chain: " |
| 176 | + << handle_hash.ToString().substr(0, 16) |
| 177 | + << " chain_size=" << m_chain->size(); |
| 178 | + continue; |
| 179 | + } |
| 180 | + uint64_t n = std::min(parents + 1, (uint64_t)m_chain->get_height(handle_hash)); |
| 181 | + for (auto [hash, data] : m_chain->get_chain(handle_hash, n)) |
| 182 | + { |
| 183 | + if (std::find(stops.begin(), stops.end(), hash) != stops.end()) |
| 184 | + break; |
| 185 | + shares.push_back(data.share); |
| 186 | + } |
| 187 | + } |
| 188 | + |
| 189 | + if (!shares.empty()) |
| 190 | + LOG_INFO << "[Pool] Sending " << shares.size() << " shares to " << peer_addr.to_string(); |
| 191 | + |
| 192 | + return shares; |
| 193 | +} |
| 194 | + |
| 195 | +} // namespace dash |
0 commit comments