GetBlocks by block index

src/neo/Ledger/Blockchain.cs

@@ -60,7 +60,6 @@ public class RelayResult { public IInventory Inventory; public VerifyResult Resu
         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>>();
         internal readonly RelayCache ConsensusRelayCache = new RelayCache(100);
         private SnapshotView currentSnapshot;
@@ -141,7 +140,6 @@ public Blockchain(NeoSystem system, IStore store)
 
         public bool ContainsBlock(UInt256 hash)
         {
-            if (block_cache.ContainsKey(hash)) return true;
             return View.ContainsBlock(hash);
         }
 
@@ -188,8 +186,6 @@ public Block GetBlock(uint index)
 
         public Block GetBlock(UInt256 hash)
         {
-            if (block_cache.TryGetValue(hash, out Block block))
-                return block;
             return View.GetBlock(hash);
         }
 
@@ -214,8 +210,6 @@ public Header GetHeader(uint index)
 
         public Header GetHeader(UInt256 hash)
         {
-            if (block_cache.TryGetValue(hash, out Block block))
-                return block.Header;
             return View.GetHeader(hash);
         }
 
@@ -255,12 +249,19 @@ private void OnImport(IEnumerable<Block> blocks, bool verify)
 
         private void AddUnverifiedBlockToCache(Block block)
         {
+            // Check if any block proposal for height `block.Index` exists
             if (!block_cache_unverified.TryGetValue(block.Index, out LinkedList<Block> blocks))
             {
+                // There are no blocks, a new LinkedList is created and, consequently, the current block is added to the list
                 blocks = new LinkedList<Block>();
                 block_cache_unverified.Add(block.Index, blocks);
             }
-
+            // Check if any block with the hash being added already exists on possible candidates to be processed
+            foreach (var unverifiedBlock in blocks)
+            {
+                if (block.Hash == unverifiedBlock.Hash)
+                    return;
+            }
             blocks.AddLast(block);
         }
 
@@ -311,78 +312,25 @@ private VerifyResult OnNewBlock(Block block)
         {
             if (block.Index <= Height)
                 return VerifyResult.AlreadyExists;
-            if (block_cache.ContainsKey(block.Hash))
-                return VerifyResult.AlreadyExists;
-            if (block.Index - 1 >= header_index.Count)
+            if (block.Index - 1 > Height)
             {
                 AddUnverifiedBlockToCache(block);
                 return VerifyResult.UnableToVerify;
             }
-            if (block.Index == header_index.Count)
+            if (block.Index == Height + 1)
             {
                 if (!block.Verify(currentSnapshot))
                     return VerifyResult.Invalid;
-            }
-            else
-            {
-                if (!block.Hash.Equals(header_index[(int)block.Index]))
-                    return VerifyResult.Invalid;
-            }
-            if (block.Index == Height + 1)
-            {
-                Block block_persist = block;
-                List<Block> blocksToPersistList = new List<Block>();
-                while (true)
-                {
-                    blocksToPersistList.Add(block_persist);
-                    if (block_persist.Index + 1 >= header_index.Count) break;
-                    UInt256 hash = header_index[(int)block_persist.Index + 1];
-                    if (!block_cache.TryGetValue(hash, out block_persist)) break;
-                }
-
-                int blocksPersisted = 0;
-                foreach (Block blockToPersist in blocksToPersistList)
-                {
-                    block_cache_unverified.Remove(blockToPersist.Index);
-                    Persist(blockToPersist);
-
-                    // 15000 is the default among of seconds per block, while MilliSecondsPerBlock is the current
-                    uint extraBlocks = (15000 - MillisecondsPerBlock) / 1000;
-
-                    if (blocksPersisted++ < blocksToPersistList.Count - (2 + Math.Max(0, extraBlocks))) continue;
-                    // Empirically calibrated for relaying the most recent 2 blocks persisted with 15s network
-                    // Increase in the rate of 1 block per second in configurations with faster blocks
-
-                    if (blockToPersist.Index + 100 >= header_index.Count)
-                        system.LocalNode.Tell(new LocalNode.RelayDirectly { Inventory = blockToPersist });
-                }
+                block_cache_unverified.Remove(block.Index);
+                Persist(block);
                 SaveHeaderHashList();
-
                 if (block_cache_unverified.TryGetValue(Height + 1, out LinkedList<Block> unverifiedBlocks))
                 {
                     foreach (var unverifiedBlock in unverifiedBlocks)
                         Self.Tell(unverifiedBlock, ActorRefs.NoSender);
                     block_cache_unverified.Remove(Height + 1);
                 }
             }
-            else
-            {
-                block_cache.Add(block.Hash, block);
-                if (block.Index + 100 >= header_index.Count)
-                    system.LocalNode.Tell(new LocalNode.RelayDirectly { Inventory = block });
-                if (block.Index == header_index.Count)
-                {
-                    header_index.Add(block.Hash);
-                    using (SnapshotView snapshot = GetSnapshot())
-                    {
-                        snapshot.Blocks.Add(block.Hash, block.Header.Trim());
-                        snapshot.HeaderHashIndex.GetAndChange().Set(block);
-                        SaveHeaderHashList(snapshot);
-                        snapshot.Commit();
-                    }
-                    UpdateCurrentSnapshot();
-                }
-            }
             return VerifyResult.Succeed;
         }
 
@@ -394,27 +342,6 @@ private VerifyResult OnNewConsensus(ConsensusPayload payload)
             return VerifyResult.Succeed;
         }
 
-        private void OnNewHeaders(Header[] headers)
-        {
-            using (SnapshotView snapshot = GetSnapshot())
-            {
-                foreach (Header header in headers)
-                {
-                    if (header.Index - 1 >= header_index.Count) break;
-                    if (header.Index < header_index.Count) continue;
-                    if (!header.Verify(snapshot)) break;
-                    header_index.Add(header.Hash);
-                    snapshot.Blocks.Add(header.Hash, header.Trim());
-                    snapshot.HeaderHashIndex.GetAndChange().Hash = header.Hash;
-                    snapshot.HeaderHashIndex.GetAndChange().Index = header.Index;
-                }
-                SaveHeaderHashList(snapshot);
-                snapshot.Commit();
-            }
-            UpdateCurrentSnapshot();
-            system.TaskManager.Tell(new TaskManager.HeaderTaskCompleted(), Sender);
-        }
-
         private VerifyResult OnNewTransaction(Transaction transaction)
         {
             if (ContainsTransaction(transaction.Hash)) return VerifyResult.AlreadyExists;
@@ -427,9 +354,9 @@ private VerifyResult OnNewTransaction(Transaction transaction)
 
         private void OnPersistCompleted(Block block)
         {
-            block_cache.Remove(block.Hash);
             MemPool.UpdatePoolForBlockPersisted(block, currentSnapshot);
             Context.System.EventStream.Publish(new PersistCompleted { Block = block });
+            system.LocalNode.Tell(Message.Create(MessageCommand.Ping, PingPayload.Create(Singleton.Height)));
         }
 
         protected override void OnReceive(object message)
@@ -442,9 +369,6 @@ protected override void OnReceive(object message)
                 case FillMemoryPool fill:
                     OnFillMemoryPool(fill.Transactions);
                     break;
-                case Header[] headers:
-                    OnNewHeaders(headers);
-                    break;
                 case Block block:
                     OnInventory(block, false);
                     break;
@@ -593,7 +517,6 @@ internal protected override bool IsHighPriority(object message)
         {
             switch (message)
             {
-                case Header[] _:
                 case Block _:
                 case ConsensusPayload _:
                 case Terminated _:

src/neo/NeoSystem.cs

@@ -21,6 +21,7 @@ public class NeoSystem : IDisposable
         public IActorRef Blockchain { get; }
         public IActorRef LocalNode { get; }
         internal IActorRef TaskManager { get; }
+        internal IActorRef SyncManager { get; }
         public IActorRef Consensus { get; private set; }
 
         private readonly IStore store;
@@ -36,6 +37,7 @@ public NeoSystem(string storageEngine = null)
             this.Blockchain = ActorSystem.ActorOf(Ledger.Blockchain.Props(this, store));
             this.LocalNode = ActorSystem.ActorOf(Network.P2P.LocalNode.Props(this));
             this.TaskManager = ActorSystem.ActorOf(Network.P2P.TaskManager.Props(this));
+            this.SyncManager = ActorSystem.ActorOf(Network.P2P.SyncManager.Props(this));
             foreach (var plugin in Plugin.Plugins)
                 plugin.OnPluginsLoaded();
         }

src/neo/Network/P2P/MessageCommand.cs

@@ -20,7 +20,7 @@ public enum MessageCommand : byte
         Pong = 0x19,
 
         //synchronization
-        [ReflectionCache(typeof(GetBlocksPayload))]
+        [ReflectionCache(typeof(GetHeadersPayload))]
         GetHeaders = 0x20,
         [ReflectionCache(typeof(HeadersPayload))]
         Headers = 0x21,

src/neo/Network/P2P/NodeSession.cs

@@ -0,0 +1,17 @@
+using System;
+using System.Collections.Generic;
+
+namespace Neo.Network.P2P
+{
+    internal class NodeSession
+    {
+        public readonly Dictionary<UInt256, DateTime> InvTasks = new Dictionary<UInt256, DateTime>();
+        public Dictionary<uint, DateTime> IndexTasks = new Dictionary<uint, DateTime>();
+
+        public uint TimeoutTimes = 0;
+        public uint InvalidBlockCount = 0;
+
+        public bool HasInvTask => InvTasks.Count > 0;
+        public bool HasIndexTask => IndexTasks.Count > 0;
+    }
+}

src/neo/Network/P2P/Payloads/GetHeadersPayload.cs

@@ -0,0 +1,37 @@
+using Neo.IO;
+using System;
+using System.IO;
+
+namespace Neo.Network.P2P.Payloads
+{
+    public class GetHeadersPayload : ISerializable
+    {
+        public uint IndexStart;
+        public short Count;
+
+        public int Size => sizeof(uint) + sizeof(short);
+
+        public static GetHeadersPayload Create(uint index_start, short count = -1)
+        {
+            return new GetHeadersPayload
+            {
+                IndexStart = index_start,
+                Count = count
+            };
+        }
+
+        void ISerializable.Deserialize(BinaryReader reader)
+        {
+            IndexStart = reader.ReadUInt32();
+            Count = reader.ReadInt16();
+            if (Count < -1 || Count == 0 || Count > HeadersPayload.MaxHeadersCount)
+                throw new FormatException();
+        }
+
+        void ISerializable.Serialize(BinaryWriter writer)
+        {
+            writer.Write(IndexStart);
+            writer.Write(Count);
+        }
+    }
+}

src/neo/Network/P2P/ProtocolHandler.cs

@@ -89,7 +89,7 @@ private void OnMessage(Message msg)
                     OnAddrMessageReceived((AddrPayload)msg.Payload);
                     break;
                 case MessageCommand.Block:
-                    OnInventoryReceived((Block)msg.Payload);
+                    OnBlockReceived((Block)msg.Payload);
                     break;
                 case MessageCommand.Consensus:
                     OnInventoryReceived((ConsensusPayload)msg.Payload);
@@ -116,7 +116,7 @@ private void OnMessage(Message msg)
                     OnGetDataMessageReceived((InvPayload)msg.Payload);
                     break;
                 case MessageCommand.GetHeaders:
-                    OnGetHeadersMessageReceived((GetBlocksPayload)msg.Payload);
+                    OnGetHeadersMessageReceived((GetHeadersPayload)msg.Payload);
                     break;
                 case MessageCommand.Headers:
                     OnHeadersMessageReceived((HeadersPayload)msg.Payload);
@@ -148,6 +148,12 @@ private void OnMessage(Message msg)
             }
         }
 
+        private void OnBlockReceived(Block payload)
+        {
+            if (payload != null)
+                system.SyncManager.Tell(payload, Context.Parent);
+        }
+
         private void OnAddrMessageReceived(AddrPayload payload)
         {
             system.LocalNode.Tell(new Peer.Peers
@@ -282,24 +288,20 @@ private void OnGetDataMessageReceived(InvPayload payload)
 
         /// <summary>
         /// Will be triggered when a MessageCommand.GetHeaders message is received.
-        /// Tell the specified number of blocks' headers starting with the requested HashStart to RemoteNode actor.
+        /// Tell the specified number of blocks' headers starting with the requested IndexStart to RemoteNode actor.
         /// A limit set by HeadersPayload.MaxHeadersCount is also applied to the number of requested Headers, namely payload.Count.
         /// </summary>
-        /// <param name="payload">A GetBlocksPayload including start block Hash and number of blocks' headers requested.</param>
-        private void OnGetHeadersMessageReceived(GetBlocksPayload payload)
+        /// <param name="payload">A GetBlocksPayload including start block index and number of blocks' headers requested.</param>
+        private void OnGetHeadersMessageReceived(GetHeadersPayload payload)
         {
-            UInt256 hash = payload.HashStart;
-            int count = payload.Count < 0 || payload.Count > HeadersPayload.MaxHeadersCount ? HeadersPayload.MaxHeadersCount : payload.Count;
-            DataCache<UInt256, TrimmedBlock> cache = Blockchain.Singleton.View.Blocks;
-            TrimmedBlock state = cache.TryGet(hash);
-            if (state == null) return;
+            uint index = payload.IndexStart;
+            int count = payload.Count < 0 ? HeadersPayload.MaxHeadersCount : payload.Count;
+            if (index > Blockchain.Singleton.HeaderHeight)
+                return;
             List<Header> headers = new List<Header>();
-            for (uint i = 1; i <= count; i++)
+            for (uint i = 0; i < count; i++)
             {
-                uint index = state.Index + i;
-                hash = Blockchain.Singleton.GetBlockHash(index);
-                if (hash == null) break;
-                Header header = cache.TryGet(hash)?.Header;
+                var header = Blockchain.Singleton.GetHeader(index + i);
                 if (header == null) break;
                 headers.Add(header);
             }

src/neo/Network/P2P/RemoteNode.cs

@@ -16,6 +16,7 @@ namespace Neo.Network.P2P
     public class RemoteNode : Connection
     {
         internal class Relay { public IInventory Inventory; }
+        internal NodeSession session = new NodeSession();
 
         private readonly NeoSystem system;
         private readonly IActorRef protocol;
@@ -161,7 +162,7 @@ private void OnPingPayload(PingPayload payload)
             if (payload.LastBlockIndex > LastBlockIndex)
             {
                 LastBlockIndex = payload.LastBlockIndex;
-                system.TaskManager.Tell(new TaskManager.Update { LastBlockIndex = LastBlockIndex });
+                system.SyncManager.Tell(new SyncManager.StartSync { });
             }
         }
 
@@ -195,7 +196,8 @@ private void OnSetFilter(BloomFilter filter)
         private void OnVerack()
         {
             verack = true;
-            system.TaskManager.Tell(new TaskManager.Register { Version = Version });
+            system.TaskManager.Tell(new TaskManager.Register { Node = this });
+            system.SyncManager.Tell(new SyncManager.Register { Node = this });
             CheckMessageQueue();
         }