Improve Large MemoryPool Performance - Sort + intelligent TX reverification

neo/Consensus/ConsensusContext.cs

@@ -202,7 +202,7 @@ public void Reset()
 
         public void Fill()
         {
-            IEnumerable<Transaction> mem_pool = Blockchain.Singleton.GetMemoryPool();
+            IEnumerable<Transaction> mem_pool = Blockchain.Singleton.GetMemoryPoolVerified();
             foreach (IPolicyPlugin plugin in Plugin.Policies)
                 mem_pool = plugin.FilterForBlock(mem_pool);
             List<Transaction> transactions = mem_pool.ToList();

neo/Ledger/Blockchain.cs

@@ -1,6 +1,5 @@
 using Akka.Actor;
 using Akka.Configuration;
-using Neo.Cryptography;
 using Neo.Cryptography.ECC;
 using Neo.IO;
 using Neo.IO.Actors;
@@ -12,10 +11,8 @@
 using Neo.SmartContract;
 using Neo.VM;
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
-using System.Numerics;
 using System.Threading;
 
 namespace Neo.Ledger
@@ -114,14 +111,14 @@ public class ImportCompleted { }
             }
         };
 
+        private const int MaxTxToReverifyPerIdle = 10;
         private static readonly object lockObj = new object();
         private readonly NeoSystem system;
         private readonly List<UInt256> header_index = new List<UInt256>();
         private uint stored_header_count = 0;
         private readonly Dictionary<UInt256, Block> block_cache = new Dictionary<UInt256, Block>();
         private readonly Dictionary<uint, LinkedList<Block>> block_cache_unverified = new Dictionary<uint, LinkedList<Block>>();
         private readonly MemoryPool mem_pool = new MemoryPool(50_000);
-        private readonly ConcurrentDictionary<UInt256, Transaction> mem_pool_unverified = new ConcurrentDictionary<UInt256, Transaction>();
         internal readonly RelayCache RelayCache = new RelayCache(100);
         private readonly HashSet<IActorRef> subscribers = new HashSet<IActorRef>();
         private Snapshot currentSnapshot;
@@ -223,6 +220,11 @@ public IEnumerable<Transaction> GetMemoryPool()
             return mem_pool;
         }
 
+        public IEnumerable<Transaction> GetMemoryPoolVerified()
+        {
+            return mem_pool.GetVerifiedTransactions();
+        }
+
         public Snapshot GetSnapshot()
         {
             return Store.GetSnapshot();
@@ -237,7 +239,7 @@ public Transaction GetTransaction(UInt256 hash)
 
         internal Transaction GetUnverifiedTransaction(UInt256 hash)
         {
-            mem_pool_unverified.TryGetValue(hash, out Transaction transaction);
+            mem_pool.TryGetUnverified(hash, out Transaction transaction);
             return transaction;
         }
 
@@ -289,7 +291,7 @@ private RelayResultReason OnNewBlock(Block block)
             if (block.Index == Height + 1)
             {
                 Block block_persist = block;
-                List<Block> blocksToPersistList = new List<Block>();                                
+                List<Block> blocksToPersistList = new List<Block>();
                 while (true)
                 {
                     blocksToPersistList.Add(block_persist);
@@ -315,7 +317,7 @@ private RelayResultReason OnNewBlock(Block block)
                 if (block_cache_unverified.TryGetValue(Height + 1, out LinkedList<Block> unverifiedBlocks))
                 {
                     foreach (var unverifiedBlock in unverifiedBlocks)
-                        Self.Tell(unverifiedBlock, ActorRefs.NoSender);   
+                        Self.Tell(unverifiedBlock, ActorRefs.NoSender);
                     block_cache_unverified.Remove(Height + 1);
                 }
             }
@@ -385,7 +387,9 @@ private RelayResultReason OnNewTransaction(Transaction transaction)
                 return RelayResultReason.Invalid;
             if (ContainsTransaction(transaction.Hash))
                 return RelayResultReason.AlreadyExists;
-            if (!transaction.Verify(currentSnapshot, GetMemoryPool()))
+            if (!mem_pool.CanTransactionFitInPool(transaction))
+                return RelayResultReason.OutOfMemory;
+            if (!transaction.Verify(currentSnapshot, GetMemoryPoolVerified()))
                 return RelayResultReason.Invalid;
             if (!Plugin.CheckPolicy(transaction))
                 return RelayResultReason.Unknown;
@@ -400,18 +404,7 @@ private RelayResultReason OnNewTransaction(Transaction transaction)
         private void OnPersistCompleted(Block block)
         {
             block_cache.Remove(block.Hash);
-            foreach (Transaction tx in block.Transactions)
-                mem_pool.TryRemove(tx.Hash, out _);
-            mem_pool_unverified.Clear();
-            foreach (Transaction tx in mem_pool
-                .OrderByDescending(p => p.NetworkFee / p.Size)
-                .ThenByDescending(p => p.NetworkFee)
-                .ThenByDescending(p => new BigInteger(p.Hash.ToArray())))
-            {
-                mem_pool_unverified.TryAdd(tx.Hash, tx);
-                Self.Tell(tx, ActorRefs.NoSender);
-            }
-            mem_pool.Clear();
+            mem_pool.UpdatePoolForBlockPersisted(block, currentSnapshot);
             PersistCompleted completed = new PersistCompleted { Block = block };
             system.Consensus?.Tell(completed);
             Distribute(completed);
@@ -439,6 +432,10 @@ protected override void OnReceive(object message)
                 case ConsensusPayload payload:
                     Sender.Tell(OnNewConsensus(payload));
                     break;
+                case Idle _:
+                    if (mem_pool.ReVerifyTopUnverifiedTransactionsIfNeeded(MaxTxToReverifyPerIdle, currentSnapshot))
+                        Self.Tell(Idle.Instance, ActorRefs.NoSender);
+                    break;
                 case Terminated terminated:
                     subscribers.Remove(terminated.ActorRef);
                     break;