Ask for mempool after sync (neo-project#1544)

* Query mempool if I have less tx than the remote node

* Revert some changes

* Wait for verack

* Clean file

* Move to verack

* Wait for block sync

* Clean changes

* Only accept one mempool message per remote node

* Remove one variable

* Revert "Only accept one mempool message per remote node"

This reverts commit 359a3bb.

* Optimize

* Check it in TaskManager

* Update src/neo/Network/P2P/TaskSession.cs

Co-authored-by: Luchuan <luchuan@neo.org>

* Revert

* fix MemPoolTaskHash (neo-project#1660)

* Fix ms bug

* private

Co-authored-by: Vitor Nazário Coelho <vncoelho@gmail.com>
Co-authored-by: Luchuan <luchuan@neo.org>
Co-authored-by: Erik Zhang <erik@neo.org>

src/neo/Network/P2P/TaskManager.cs

@@ -24,11 +24,13 @@ private class Timer { }
 
         private static readonly TimeSpan TimerInterval = TimeSpan.FromSeconds(30);
         private static readonly TimeSpan TaskTimeout = TimeSpan.FromMinutes(1);
+        private static readonly UInt256 HeaderTaskHash = UInt256.Zero;
+        private static readonly UInt256 MemPoolTaskHash = UInt256.Parse("0x0000000000000000000000000000000000000000000000000000000000000001");
 
         private readonly NeoSystem system;
         private const int MaxConncurrentTasks = 3;
 
-        private const int PingCoolingOffPeriod = 60; // in secconds.
+        private const int PingCoolingOffPeriod = 60_000; // in ms.
         /// <summary>
         /// A set of known hashes, of inventories or payloads, already received.
         /// </summary>        
@@ -37,7 +39,6 @@ private class Timer { }
         private readonly Dictionary<IActorRef, TaskSession> sessions = new Dictionary<IActorRef, TaskSession>();
         private readonly ICancelable timer = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(TimerInterval, TimerInterval, Context.Self, new Timer(), ActorRefs.NoSender);
 
-        private readonly UInt256 HeaderTaskHash = UInt256.Zero;
         private bool HasHeaderTask => globalTasks.ContainsKey(HeaderTaskHash);
 
         public TaskManager(NeoSystem system)
@@ -126,6 +127,8 @@ private void OnRegister(VersionPayload version)
         {
             Context.Watch(Sender);
             TaskSession session = new TaskSession(Sender, version);
+            if (session.IsFullNode)
+                session.AvailableTasks.Add(TaskManager.MemPoolTaskHash);
             sessions.Add(Sender, session);
             RequestTasks(session);
         }
@@ -183,7 +186,6 @@ private bool IncrementGlobalTask(UInt256 hash)
                 return false;
 
             globalTasks[hash] = value + 1;
-
             return true;
         }
 
@@ -230,6 +232,7 @@ private void RequestTasks(TaskSession session)
                 // Search any similar hash that is on Singleton's knowledge, which means, on the way or already processed
                 session.AvailableTasks.RemoveWhere(p => Blockchain.Singleton.ContainsBlock(p));
                 HashSet<UInt256> hashes = new HashSet<UInt256>(session.AvailableTasks);
+                hashes.Remove(MemPoolTaskHash);
                 if (hashes.Count > 0)
                 {
                     foreach (UInt256 hash in hashes.ToArray())
@@ -269,8 +272,13 @@ private void RequestTasks(TaskSession session)
                 session.RemoteNode.Tell(Message.Create(MessageCommand.GetBlocks, GetBlocksPayload.Create(hash)));
             }
             else if (Blockchain.Singleton.HeaderHeight >= session.LastBlockIndex
-                    && TimeProvider.Current.UtcNow.ToTimestamp() - PingCoolingOffPeriod >= Blockchain.Singleton.GetBlock(Blockchain.Singleton.CurrentHeaderHash)?.Timestamp)
+                    && TimeProvider.Current.UtcNow.ToTimestampMS() - PingCoolingOffPeriod >= Blockchain.Singleton.GetBlock(Blockchain.Singleton.CurrentHeaderHash)?.Timestamp)
             {
+                if (session.AvailableTasks.Remove(MemPoolTaskHash))
+                {
+                    session.RemoteNode.Tell(Message.Create(MessageCommand.Mempool));
+                }
+
                 session.RemoteNode.Tell(Message.Create(MessageCommand.Ping, PingPayload.Create(Blockchain.Singleton.Height)));
             }
         }

src/neo/Network/P2P/TaskSession.cs

@@ -16,15 +16,17 @@ internal class TaskSession
 
         public bool HasTask => Tasks.Count > 0;
         public uint StartHeight { get; }
+        public bool IsFullNode { get; }
         public uint LastBlockIndex { get; set; }
 
         public TaskSession(IActorRef node, VersionPayload version)
         {
+            var fullNode = version.Capabilities.OfType<FullNodeCapability>().FirstOrDefault();
+
+            this.IsFullNode = fullNode != null;
             this.RemoteNode = node;
             this.Version = version;
-            this.StartHeight = version.Capabilities
-                .OfType<FullNodeCapability>()
-                .FirstOrDefault()?.StartHeight ?? 0;
+            this.StartHeight = fullNode?.StartHeight ?? 0;
             this.LastBlockIndex = this.StartHeight;
         }
     }