TCP P2P optimization

neo.UnitTests/UT_P2PMessage.cs

@@ -0,0 +1,110 @@
+using Akka.IO;
+using FluentAssertions;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Neo.IO;
+using Neo.Network.P2P;
+using Neo.Network.P2P.Payloads;
+
+namespace Neo.UnitTests
+{
+    [TestClass]
+    public class UT_P2PMessage
+    {
+        [TestMethod]
+        public void Serialize_Deserialize()
+        {
+            var payload = PingPayload.Create(uint.MaxValue);
+            var msg = Message.Create(MessageCommand.Ping, payload);
+            var buffer = msg.ToArray();
+            var copy = buffer.AsSerializable<Message>();
+            var payloadCopy = (PingPayload)copy.Payload;
+
+            copy.Command.Should().Be(msg.Command);
+            copy.Flags.Should().Be(msg.Flags);
+
+            payloadCopy.LastBlockIndex.Should().Be(payload.LastBlockIndex);
+            payloadCopy.Nonce.Should().Be(payload.Nonce);
+            payloadCopy.Timestamp.Should().Be(payload.Timestamp);
+        }
+
+        [TestMethod]
+        public void Serialize_Deserialize_ByteString()
+        {
+            var payload = PingPayload.Create(uint.MaxValue);
+            var msg = Message.Create(MessageCommand.Ping, payload);
+            var buffer = ByteString.CopyFrom(msg.ToArray());
+            var length = Message.TryDeserialize(buffer, out var copy);
+
+            var payloadCopy = (PingPayload)copy.Payload;
+
+            copy.Command.Should().Be(msg.Command);
+            copy.Flags.Should().Be(msg.Flags);
+
+            payloadCopy.LastBlockIndex.Should().Be(payload.LastBlockIndex);
+            payloadCopy.Nonce.Should().Be(payload.Nonce);
+            payloadCopy.Timestamp.Should().Be(payload.Timestamp);
+
+            buffer.Count.Should().Be(length);
+        }
+
+        [TestMethod]
+        public void Serialize_Deserialize_WithoutPayload()
+        {
+            var msg = Message.Create(MessageCommand.GetAddr);
+            var buffer = msg.ToArray();
+            var copy = buffer.AsSerializable<Message>();
+
+            copy.Command.Should().Be(msg.Command);
+            copy.Flags.Should().Be(msg.Flags);
+            copy.Payload.Should().Be(null);
+        }
+
+        [TestMethod]
+        public void Serialize_Deserialize_WithoutPayload_ByteString()
+        {
+            var msg = Message.Create(MessageCommand.GetAddr);
+            var buffer = ByteString.CopyFrom(msg.ToArray());
+            var length = Message.TryDeserialize(buffer, out var copy);
+
+            copy.Command.Should().Be(msg.Command);
+            copy.Flags.Should().Be(msg.Flags);
+            copy.Payload.Should().Be(null);
+
+            buffer.Count.Should().Be(length);
+        }
+
+
+        [TestMethod]
+        public void Compression()
+        {
+            var payload = new VersionPayload()
+            {
+                UserAgent = "".PadLeft(1024, '0'),
+                Nonce = 1,
+                Port = 2,
+                Services = VersionServices.FullNode,
+                StartHeight = 4,
+                Timestamp = 5,
+                Version = 6
+            };
+            var msg = Message.Create(MessageCommand.Version, payload);
+            var buffer = msg.ToArray();
+
+            buffer.Length.Should().BeLessThan(80);
+
+            var copy = buffer.AsSerializable<Message>();
+            var payloadCopy = (VersionPayload)copy.Payload;
+
+            copy.Command.Should().Be(msg.Command);
+            copy.Flags.Should().HaveFlag(MessageFlags.Compressed);
+
+            payloadCopy.UserAgent.Should().Be(payload.UserAgent);
+            payloadCopy.Nonce.Should().Be(payload.Nonce);
+            payloadCopy.Port.Should().Be(payload.Port);
+            payloadCopy.Services.Should().Be(payload.Services);
+            payloadCopy.StartHeight.Should().Be(payload.StartHeight);
+            payloadCopy.Timestamp.Should().Be(payload.Timestamp);
+            payloadCopy.Version.Should().Be(payload.Version);
+        }
+    }
+}

neo/Consensus/ConsensusService.cs

@@ -624,7 +624,7 @@ private void SendPrepareRequest()
             if (context.TransactionHashes.Length > 1)
             {
                 foreach (InvPayload payload in InvPayload.CreateGroup(InventoryType.TX, context.TransactionHashes.Skip(1).ToArray()))
-                    localNode.Tell(Message.Create("inv", payload));
+                    localNode.Tell(Message.Create(MessageCommand.Inv, payload));
             }
             ChangeTimer(TimeSpan.FromSeconds((Blockchain.SecondsPerBlock << (context.ViewNumber + 1)) - (context.ViewNumber == 0 ? Blockchain.SecondsPerBlock : 0)));
         }

neo/Network/P2P/Helper.cs

@@ -1,10 +1,42 @@
-using Neo.Network.P2P.Payloads;
+using K4os.Compression.LZ4;
+using Neo.Network.P2P.Payloads;
+using System;
+using System.Buffers;
 using System.IO;
 
 namespace Neo.Network.P2P
 {
     public static class Helper
     {
+        public static byte[] CompressLz4(this byte[] data)
+        {
+            int maxLength = LZ4Codec.MaximumOutputSize(data.Length);
+            byte[] buffer = ArrayPool<byte>.Shared.Rent(maxLength);
+            int length = LZ4Codec.Encode(data, 0, data.Length, buffer, 0, buffer.Length);
+            data = new byte[length];
+            Buffer.BlockCopy(buffer, 0, data, 0, length);
+            ArrayPool<byte>.Shared.Return(buffer);
+            return data;
+        }
+
+        public static byte[] DecompressLz4(this byte[] data, int maxOutput)
+        {
+            maxOutput = Math.Min(maxOutput, data.Length * 255);
+            byte[] buffer = ArrayPool<byte>.Shared.Rent(maxOutput);
+            try
+            {
+                int length = LZ4Codec.Decode(data, 0, data.Length, buffer, 0, buffer.Length);
+                if (length < 0 || length > maxOutput) throw new FormatException();
+                data = new byte[length];
+                Buffer.BlockCopy(buffer, 0, data, 0, length);
+                return data;
+            }
+            finally
+            {
+                ArrayPool<byte>.Shared.Return(buffer);
+            }
+        }
+
         public static byte[] GetHashData(this IVerifiable verifiable)
         {
             using (MemoryStream ms = new MemoryStream())
@@ -15,5 +47,20 @@ public static byte[] GetHashData(this IVerifiable verifiable)
                 return ms.ToArray();
             }
         }
+
+        internal static MessageCommand ToMessageCommand(this InventoryType inventoryType)
+        {
+            switch (inventoryType)
+            {
+                case InventoryType.TX:
+                    return MessageCommand.Transaction;
+                case InventoryType.Block:
+                    return MessageCommand.Block;
+                case InventoryType.Consensus:
+                    return MessageCommand.Consensus;
+                default:
+                    throw new ArgumentOutOfRangeException(nameof(inventoryType));
+            }
+        }
     }
 }

neo/Network/P2P/LocalNode.cs

@@ -58,7 +58,7 @@ public LocalNode(NeoSystem system)
             }
         }
 
-        private void BroadcastMessage(string command, ISerializable payload = null)
+        private void BroadcastMessage(MessageCommand command, ISerializable payload = null)
         {
             BroadcastMessage(Message.Create(command, payload));
         }
@@ -126,7 +126,7 @@ protected override void NeedMorePeers(int count)
             count = Math.Max(count, 5);
             if (ConnectedPeers.Count > 0)
             {
-                BroadcastMessage("getaddr");
+                BroadcastMessage(MessageCommand.GetAddr);
             }
             else
             {