From 935f3761c2c0550dfa98a00edac416d58f9cb98f Mon Sep 17 00:00:00 2001 From: Brandon Koerner Date: Thu, 29 Feb 2024 13:28:19 -0500 Subject: [PATCH] add lock for orphans --- Discreet/Daemon/Daemon.cs | 18 +++++- Discreet/Network/Handler.cs | 103 ++++++++++++++++++++----------- Discreet/Network/MessageCache.cs | 2 + 3 files changed, 86 insertions(+), 37 deletions(-) diff --git a/Discreet/Daemon/Daemon.cs b/Discreet/Daemon/Daemon.cs index d7096e9..a4fa51a 100644 --- a/Discreet/Daemon/Daemon.cs +++ b/Discreet/Daemon/Daemon.cs @@ -763,6 +763,8 @@ public async Task TestnetMinter(TimeSpan interval, int n, int pid, ChannelReader bool paused = true; bool reloop = false; + HashSet produced = new HashSet(); + // we start off paused, waiting for the first call to us while (paused) { @@ -826,7 +828,7 @@ public async Task TestnetMinter(TimeSpan interval, int n, int pid, ChannelReader if (numProduced % n == pid) { await MintLocker.WaitAsync(); - await MintTestnetBlock(); + await MintTestnetBlock(produced); MintLocker.Release(); } @@ -930,7 +932,7 @@ public async Task MintTestnet() } } - public async Task MintTestnetBlock() + public async Task MintTestnetBlock(HashSet debugProduced = null) { try { @@ -940,6 +942,18 @@ public async Task MintTestnetBlock() var txs = txpool.GetTransactionsForBlock(); var blk = Block.Build(txs, new StealthAddress(SQLiteWallet.Wallets["TESTNET_EMISSIONS"].Accounts[0].Address), sigKey); + if (debugProduced != null) + { + if (debugProduced.Contains(blk.Header.Height)) + { + Logger.Critical($"Discreet.Daemon: produced a block at a previously seen height!"); + } + else + { + debugProduced.Add(blk.Header.Height); + } + } + try { DB.ValidationCache vCache = new(blk); diff --git a/Discreet/Network/Handler.cs b/Discreet/Network/Handler.cs index f3b2442..a8dff4b 100644 --- a/Discreet/Network/Handler.cs +++ b/Discreet/Network/Handler.cs @@ -486,14 +486,14 @@ public bool IsParallel(Packet p) case PacketType.SENDTX: case PacketType.GETTXS: case PacketType.TXS: - case PacketType.BLOCKS: case PacketType.GETHEADERS: - case PacketType.HEADERS: case PacketType.GETPOOL: case PacketType.POOL: return true; case PacketType.SENDBLOCK: case PacketType.SENDBLOCKS: + case PacketType.BLOCKS: + case PacketType.HEADERS: case PacketType.ALERT: case PacketType.NONE: case PacketType.INVENTORY: @@ -1185,18 +1185,26 @@ public async Task HandleSendBlock(SendBlockPacket p, Peerbloom.Connection conn, { if (!MessageCache.GetMessageCache().OrphanBlockParents.ContainsKey(p.Block.Header.PreviousBlock)) { + await MessageCache.GetMessageCache().OrphanLock.WaitAsync(); + Daemon.Logger.Warn($"HandleSendBlock: orphan block ({p.Block.Header.BlockHash.ToHexShort()}, height {p.Block.Header.Height}) added; querying {conn.Receiver} for previous block", verbose: 3); MessageCache.GetMessageCache().OrphanBlocks[p.Block.Header.PreviousBlock] = p.Block; MessageCache.GetMessageCache().OrphanBlockParents[p.Block.Header.BlockHash] = p.Block.Header.PreviousBlock; Peerbloom.Network.GetNetwork().SendRequest(conn, new Packet(PacketType.GETBLOCKS, new GetBlocksPacket { Blocks = new Cipher.SHA256[] { p.Block.Header.PreviousBlock } }), durationMilliseconds: 60000); + + MessageCache.GetMessageCache().OrphanLock.Release(); return; } else { + await MessageCache.GetMessageCache().OrphanLock.WaitAsync(); + Daemon.Logger.Warn($"HandleSendBlock: orphan block ({p.Block.Header.BlockHash.ToHexShort()}, height {p.Block.Header.Height}) added", verbose: 3); MessageCache.GetMessageCache().OrphanBlocks[p.Block.Header.PreviousBlock] = p.Block; MessageCache.GetMessageCache().OrphanBlockParents[p.Block.Header.BlockHash] = p.Block.Header.PreviousBlock; CheckRoot(p.Block, conn); + + MessageCache.GetMessageCache().OrphanLock.Release(); return; } } @@ -1349,17 +1357,28 @@ public async Task HandleBlocks(BlocksPacket p, Peerbloom.Connection conn) { Daemon.Logger.Warn($"HandleBlocks: orphan block ({p.Blocks[0].Header.BlockHash.ToHexShort()}, height {p.Blocks[0].Header.Height}) added; querying {conn.Receiver} for previous block", verbose: 3); + await MessageCache.GetMessageCache().OrphanLock.WaitAsync(); + MessageCache.GetMessageCache().OrphanBlocks[p.Blocks[0].Header.PreviousBlock] = p.Blocks[0]; MessageCache.GetMessageCache().OrphanBlockParents[p.Blocks[0].Header.BlockHash] = p.Blocks[0].Header.PreviousBlock; Peerbloom.Network.GetNetwork().SendRequest(conn, new Packet(PacketType.GETBLOCKS, new GetBlocksPacket { Blocks = new Cipher.SHA256[] { p.Blocks[0].Header.PreviousBlock } }), durationMilliseconds: 60000); + + MessageCache.GetMessageCache().OrphanLock.Release(); + return; } else { Daemon.Logger.Warn($"HandleBlocks: orphan block ({p.Blocks[0].Header.BlockHash.ToHexShort()}, height {p.Blocks[0].Header.Height}) added", verbose: 1); + + await MessageCache.GetMessageCache().OrphanLock.WaitAsync(); + MessageCache.GetMessageCache().OrphanBlocks[p.Blocks[0].Header.PreviousBlock] = p.Blocks[0]; MessageCache.GetMessageCache().OrphanBlockParents[p.Blocks[0].Header.BlockHash] = p.Blocks[0].Header.PreviousBlock; CheckRoot(p.Blocks[0], conn); + + MessageCache.GetMessageCache().OrphanLock.Release(); + return; } } @@ -1368,7 +1387,11 @@ public async Task HandleBlocks(BlocksPacket p, Peerbloom.Connection conn) Daemon.Logger.Error($"HandleBlocks: Malformed or invalid block received from peer {conn.Receiver}: {err.Message} (bogus block for orphan requirement)", err); /* for now assume invalid root always has invalid leaves */ + + await MessageCache.GetMessageCache().OrphanLock.WaitAsync(); TossOrphans(p.Blocks[0].Header.BlockHash); + MessageCache.GetMessageCache().OrphanLock.Release(); + return; } @@ -1575,44 +1598,54 @@ public void CheckRoot(Block block, Peerbloom.Connection conn) public async Task AcceptOrphans(Cipher.SHA256 bHash) { MessageCache mCache = MessageCache.GetMessageCache(); - while (mCache.OrphanBlocks.ContainsKey(bHash)) + + await mCache.OrphanLock.WaitAsync(); + + try { - mCache.OrphanBlocks.Remove(bHash, out var block); - DB.ValidationCache vCache = new DB.ValidationCache(block); - var err = vCache.Validate(); - if (err is OrphanBlockException) + while (mCache.OrphanBlocks.ContainsKey(bHash)) { - // simply return - return; - } - if (err != null) - { - Daemon.Logger.Error($"AcceptOrphans: Malformed or invalid block in orphan branch {bHash.ToHexShort()} (height {block.Header.Height}): {err.Message}; tossing branch", err); - TossOrphans(bHash); - return; - } + mCache.OrphanBlocks.Remove(bHash, out var block); + DB.ValidationCache vCache = new DB.ValidationCache(block); + var err = vCache.Validate(); + if (err is OrphanBlockException) + { + // simply return + return; + } + if (err != null) + { + Daemon.Logger.Error($"AcceptOrphans: Malformed or invalid block in orphan branch {bHash.ToHexShort()} (height {block.Header.Height}): {err.Message}; tossing branch", err); + TossOrphans(bHash); + return; + } - try - { - await vCache.Flush(); - } - catch (Exception e) - { - Daemon.Logger.Error($"AcceptOrphans: an error was encountered while flushing validation cache for block at height {block.Header.Height}: {e.Message}", e); - } + try + { + await vCache.Flush(); + } + catch (Exception e) + { + Daemon.Logger.Error($"AcceptOrphans: an error was encountered while flushing validation cache for block at height {block.Header.Height}: {e.Message}", e); + } - try - { - daemon.ProcessBlock(block); - } - catch (Exception e) - { - Daemon.Logger.Error($"AcceptOrphans: an error was encountered while processing block at height {block.Header.Height}: {e.Message}", e); - } + try + { + daemon.ProcessBlock(block); + } + catch (Exception e) + { + Daemon.Logger.Error($"AcceptOrphans: an error was encountered while processing block at height {block.Header.Height}: {e.Message}", e); + } - OnBlockSuccess?.Invoke(new BlockSuccessEventArgs { Block = block }); - bHash = block.Header.BlockHash; - mCache.OrphanBlockParents.Remove(bHash, out _); + OnBlockSuccess?.Invoke(new BlockSuccessEventArgs { Block = block }); + bHash = block.Header.BlockHash; + mCache.OrphanBlockParents.Remove(bHash, out _); + } + } + finally + { + mCache.OrphanLock.Release(); } } } diff --git a/Discreet/Network/MessageCache.cs b/Discreet/Network/MessageCache.cs index e2d3520..c121c3f 100644 --- a/Discreet/Network/MessageCache.cs +++ b/Discreet/Network/MessageCache.cs @@ -8,6 +8,7 @@ using System.Collections.Concurrent; using Discreet.Coin.Models; using Discreet.DB; +using System.Threading; namespace Discreet.Network { @@ -39,6 +40,7 @@ public static MessageCache GetMessageCache() public ConcurrentDictionary OrphanBlocks; public ConcurrentDictionary OrphanBlockParents = new(new Cipher.SHA256EqualityComparer()); + public readonly SemaphoreSlim OrphanLock = new SemaphoreSlim(1, 1); public MessageCache() {