diff --git a/src/impl/dash/CMakeLists.txt b/src/impl/dash/CMakeLists.txt index 3d185eaeb..803d099e1 100644 --- a/src/impl/dash/CMakeLists.txt +++ b/src/impl/dash/CMakeLists.txt @@ -50,7 +50,7 @@ add_library(dash OBJECT peer.hpp messages.hpp share.hpp - protocol_actual.cpp protocol_legacy.cpp + protocol_actual.cpp protocol_legacy.cpp node.cpp ) target_include_directories(dash PRIVATE ${CMAKE_SOURCE_DIR}/src diff --git a/src/impl/dash/coin/transaction.hpp b/src/impl/dash/coin/transaction.hpp index b9e62ee1b..3cb03805e 100644 --- a/src/impl/dash/coin/transaction.hpp +++ b/src/impl/dash/coin/transaction.hpp @@ -27,6 +27,8 @@ using bitcoin_family::coin::TxOut; // Dash transaction type field (nType in serialization) // type=0: standard, type=5: CBTX (coinbase with DIP3/DIP4 payload) +class Transaction; // fwd — for the symmetric MutableTransaction(const Transaction&) below + struct MutableTransaction { std::vector vin; @@ -37,6 +39,12 @@ struct MutableTransaction std::vector extra_payload; // DIP3/DIP4 payload (for type=5 CBTX) MutableTransaction() = default; + // Rebuild the mutable form from an (immutable) Transaction. Symmetric to + // the explicit Transaction(const MutableTransaction&) below; used by the + // p2p dispatch path (protocol_{legacy,actual}.cpp) when staging peer- + // referenced txs for share admission. Defined out-of-line once Transaction + // is complete. + explicit MutableTransaction(const Transaction& tx); bool HasWitness() const { return false; } // Dash has no segwit @@ -103,5 +111,9 @@ class Transaction bool HasWitness() const { return false; } }; +inline MutableTransaction::MutableTransaction(const Transaction& tx) + : vin(tx.vin), vout(tx.vout), version(tx.version), + type(tx.type), locktime(tx.locktime), extra_payload(tx.extra_payload) {} + } // namespace coin } // namespace dash diff --git a/src/impl/dash/node.cpp b/src/impl/dash/node.cpp new file mode 100644 index 000000000..6f61b9b8c --- /dev/null +++ b/src/impl/dash/node.cpp @@ -0,0 +1,195 @@ +#include "node.hpp" +#include "share.hpp" + +#include +#include + +#include + +#include +#include +#include + +// Dash p2pool sharechain pool-node — S8 pool-node reception, slice .4 (bodies). +// +// SLICE SCOPE (integrator-confirmed 2026-07-09, #656 a1444b4 branch-tip (awaiting merge tap)): the two +// link-deferred declarations from node.hpp — processing_shares() and +// handle_get_share() — get their REDUCED dash-native bodies here. "Reduced" +// means: the parallel X11 share_init_verify -> try_lock + m_tracker.add core, +// with think()/download/best-share/persist DEFERRED to their own later slices. +// +// This is a deliberate, non-1:1 port of src/impl/ltc/node.cpp. The DASH node is +// header-only and does NOT carry the ~8 LTC-only symbols the full ltc bodies +// touch (run_think/think, chain::PreparedList, m_raw_share_cache, +// m_best_share_hash, chain::get_reverse fork-logging, m_rejected_share_hashes, +// processing_shares_phase2 as a distinct TU method). Those are intentionally +// absent — force-fitting them would be scope creep. What remains is the minimal +// admit path that makes the Legacy/Actual dispatch (protocol_{legacy,actual}.cpp) +// actually reach the tracker: verify in parallel off the io_context, then insert +// under the non-blocking tracker lock on the io_context thread. + +namespace dash +{ + +void NodeImpl::processing_shares(HandleSharesData& data_ref, NetService addr) +{ + // Take ownership immediately so the caller (the dispatch handler) can return + // and free its local HandleSharesData. + auto data = std::make_shared(std::move(data_ref)); + size_t n = data->m_items.size(); + if (n == 0) + return; + + // ── Phase 1 (m_verify_pool, parallel) ──────────────────────────────── + // Run share_init_verify() for each received share off the io_context thread. + // DASH is X11 (~ms of CPU per share); the structural + PoW verify must NOT + // block network I/O. Each share's hash computation is independent, so the + // batch fully parallelises across the pool. A share that throws is left with + // a null hash and skipped in phase 2. + auto remaining = std::make_shared>(static_cast(n)); + for (size_t i = 0; i < n; i++) + { + boost::asio::post(m_verify_pool, + [i, data, remaining, this, addr]() + { + auto& share = data->m_items[i]; + if (share.hash().IsNull()) + { + try + { + share.ACTION({ + obj->m_hash = share_init_verify(*obj, m_tracker.m_coin_params, true); + }); + } + catch (const std::exception&) + { + // leave hash null — phase 2 will skip this share + } + } + + // Last verify done → hop back to the io_context thread for the + // tracker insertion (writes stay single-threaded on io_context). + if (--(*remaining) == 0) + { + boost::asio::post(*m_context, + [data, this, addr]() + { + add_verified_shares(*data, addr); + }); + } + }); + } +} + +void NodeImpl::add_verified_shares(HandleSharesData& data, NetService addr) +{ + // io_context thread. Non-blocking tracker lock (architectural rule, + // node.hpp): the compute thread (a later slice's think()) takes the + // exclusive lock; the IO thread must NEVER block on it. If think() holds the + // lock, defer this batch onto m_pending_adds — the compute thread drains it + // after releasing. Until the think() slice lands nothing takes the exclusive + // lock, so try_to_lock always succeeds here. + { + std::unique_lock lock(m_tracker_mutex, std::try_to_lock); + if (!lock.owns_lock()) + { + if (m_pending_adds.size() < MAX_PENDING_ADDS) + { + LOG_INFO << "[ASYNC-DEFER] add_verified_shares: tracker busy, queuing " + << data.m_items.size() << " shares from " << addr.to_string() + << " (pending=" << m_pending_adds.size() + 1 << ")"; + m_pending_adds.push_back(PendingShareBatch{ + std::make_unique(std::move(data)), addr}); + } + else + { + LOG_WARNING << "[ASYNC-DEFER] add_verified_shares: pending queue full (" + << MAX_PENDING_ADDS << ") — dropping batch from " + << addr.to_string(); + } + return; + } + // Lock held for the whole insertion below. + + int32_t new_count = 0; + int32_t dup_count = 0; + for (auto& share : data.m_items) + { + // Skip shares that failed phase-1 verification (hash still null). + if (share.hash().IsNull()) + continue; + + if (m_chain->contains(share.hash())) + { + ++dup_count; + continue; + } + + m_tracker.add(share); + ++new_count; + } + + if (new_count > 0) + { + auto as = addr.to_string(); + std::string source = (addr.port() == 0) ? as.substr(0, as.rfind(':')) : as; + LOG_INFO << "Processing " << new_count << " shares from " << source + << "... (dup=" << dup_count + << " chain=" << m_tracker.chain.size() << ")"; + } + } + // Scoring / best-share selection / persist / relay ride the think() and + // broadcast slices; slice .4 stops at tracker insertion. +} + +std::vector +NodeImpl::handle_get_share(std::vector hashes, uint64_t parents, + std::vector stops, NetService peer_addr) +{ + // try_to_lock per the architectural rule (node.hpp): the IO thread MUST + // never block on m_tracker_mutex. An empty reply does not disconnect the + // requesting peer — p2pool's downloader retries against another random peer, + // so a busy-skip just shifts the request one iteration. + std::shared_lock lock(m_tracker_mutex, std::try_to_lock); + if (!lock.owns_lock()) + { + static int defer_log = 0; + if (defer_log++ % 50 == 0) + LOG_INFO << "[handle_get_share] tracker busy — returning empty to " + << peer_addr.to_string() + << " (peer will retry against another peer)"; + return {}; + } + + if (hashes.empty()) + return {}; + + parents = std::min(parents, (uint64_t)1000 / hashes.size()); + std::vector shares; + for (const auto& handle_hash : hashes) + { + if (!m_chain->contains(handle_hash)) + { + static int miss_log = 0; + if (miss_log++ < 5) + LOG_WARNING << "[handle_get_share] hash NOT in chain: " + << handle_hash.ToString().substr(0, 16) + << " chain_size=" << m_chain->size(); + continue; + } + uint64_t n = std::min(parents + 1, (uint64_t)m_chain->get_height(handle_hash)); + for (auto [hash, data] : m_chain->get_chain(handle_hash, n)) + { + if (std::find(stops.begin(), stops.end(), hash) != stops.end()) + break; + shares.push_back(data.share); + } + } + + if (!shares.empty()) + LOG_INFO << "[Pool] Sending " << shares.size() << " shares to " << peer_addr.to_string(); + + return shares; +} + +} // namespace dash diff --git a/src/impl/dash/node.hpp b/src/impl/dash/node.hpp index cf0864598..1d802321b 100644 --- a/src/impl/dash/node.hpp +++ b/src/impl/dash/node.hpp @@ -329,6 +329,11 @@ class NodeImpl : public pool::BaseNode handle_get_share(std::vector hashes, uint64_t parents, std::vector stops, NetService peer_addr); + // Slice .4 helper: io_context-thread tracker insertion for a phase-1- + // verified batch (non-blocking lock; defers to m_pending_adds if the + // compute thread holds the exclusive lock). Body in node.cpp. + void add_verified_shares(HandleSharesData& data, NetService addr); + // Completes a pending async share request when a sharereply arrives. // Inline (mirrors dgb) so the reply-matcher plumbing needs no node.cpp. void got_share_reply(uint256 id, dash::ShareReplyData shares) diff --git a/src/impl/dash/protocol_actual.cpp b/src/impl/dash/protocol_actual.cpp index d1af50556..29fbcb4d1 100644 --- a/src/impl/dash/protocol_actual.cpp +++ b/src/impl/dash/protocol_actual.cpp @@ -119,7 +119,7 @@ void Actual::HANDLER(shares) share.ACTION ({ if constexpr (share_t::version >= 13 && share_t::version < 34) - for (auto tx_hash : obj->m_tx_info.m_new_transaction_hashes) + for (auto tx_hash : obj->m_new_transaction_hashes) { auto it = m_known_txs.find(tx_hash); if (it != m_known_txs.end()) diff --git a/src/impl/dash/protocol_legacy.cpp b/src/impl/dash/protocol_legacy.cpp index 522c5faf4..e6ce04abc 100644 --- a/src/impl/dash/protocol_legacy.cpp +++ b/src/impl/dash/protocol_legacy.cpp @@ -114,7 +114,7 @@ void Legacy::HANDLER(shares) share.ACTION ({ if constexpr (share_t::version >= 13 && share_t::version < 34) - for (auto tx_hash : obj->m_tx_info.m_new_transaction_hashes) + for (auto tx_hash : obj->m_new_transaction_hashes) { auto it = m_known_txs.find(tx_hash); if (it != m_known_txs.end())