TCP P2P optimization

neo.UnitTests/UT_P2PMessage.cs

@@ -0,0 +1,85 @@
+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 = copy.GetPayload<PingPayload>();
+
+            copy.Command.Should().Be(MessageCommand.ping);
+            copy.Flags.Should().Be(MessageFlags.None);
+
+            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);
+
+            length.Should().Be(buffer.Count);
+
+            var payloadCopy = copy.GetPayload<PingPayload>();
+
+            copy.Command.Should().Be(MessageCommand.ping);
+            copy.Flags.Should().Be(MessageFlags.None);
+
+            payloadCopy.LastBlockIndex.Should().Be(payload.LastBlockIndex);
+            payloadCopy.Nonce.Should().Be(payload.Nonce);
+            payloadCopy.Timestamp.Should().Be(payload.Timestamp);
+        }
+
+        [TestMethod]
+        public void Compression()
+        {
+            var payload = new VersionPayload()
+            {
+                Relay = true,
+                UserAgent = "".PadLeft(1024, '0'),
+                Nonce = 1,
+                Port = 2,
+                Services = VersionServices.NodeNetwork,
+                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 = copy.GetPayload<VersionPayload>();
+
+            copy.Command.Should().Be(MessageCommand.version);
+            copy.Flags.Should().Be(MessageFlags.CompressedGzip);
+
+            payloadCopy.Relay.Should().Be(payload.Relay);
+            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,43 @@
 using Neo.Network.P2P.Payloads;
 using System.IO;
+using System.IO.Compression;
 
 namespace Neo.Network.P2P
 {
     public static class Helper
     {
+        public static byte[] UncompressGzip(this byte[] data)
+        {
+            using (var output = new MemoryStream())
+            using (var input = new MemoryStream(data))
+            using (var gzip = new GZipStream(input, CompressionMode.Decompress))
+            {
+                int nRead;
+                byte[] buffer = new byte[1024];
+
+                while ((nRead = gzip.Read(buffer, 0, buffer.Length)) > 0)
+                {
+                    output.Write(buffer, 0, nRead);
+                }
+
+                return output.ToArray();
+            }
+        }
+
+        public static byte[] CompressGzip(this byte[] data)
+        {
+            using (var stream = new MemoryStream())
+            {
+                using (var gzip = new GZipStream(stream, CompressionLevel.Optimal, true))
+                {
+                    gzip.Write(data, 0, data.Length);
+                    gzip.Flush();
+                }
+
+                return stream.ToArray();
+            }
+        }
+
         public static byte[] GetHashData(this IVerifiable verifiable)
         {
             using (MemoryStream ms = new MemoryStream())

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
             {

neo/Network/P2P/Message.cs

@@ -1,79 +1,139 @@
-using Neo.Cryptography;
+using System;
+using System.IO;
+using Akka.IO;
+using Neo.Cryptography;
 using Neo.IO;
 using Neo.Network.P2P.Payloads;
-using System;
-using System.IO;
 
 namespace Neo.Network.P2P
 {
     public class Message : ISerializable
     {
-        public const int HeaderSize = sizeof(uint) + 12 + sizeof(int) + sizeof(uint);
         public const int PayloadMaxSize = 0x02000000;
+        public const int CompressionMinSize = 180;
+        public const int CompressionThreshold = 100;
 
-        public static readonly uint Magic = ProtocolSettings.Default.Magic;
-        public string Command;
-        public uint Checksum;
+        public MessageFlags Flags;
+        public MessageCommand Command;
         public byte[] Payload;
 
-        public int Size => HeaderSize + Payload.Length;
+        private ISerializable _payload_deserialized = null;
 
-        public static Message Create(string command, ISerializable payload = null)
+        public int Size => 2 + IO.Helper.GetVarSize(Payload.Length) + Payload.Length;
+
+        public static Message Create(MessageCommand command, ISerializable payload = null)
         {
-            return Create(command, payload == null ? new byte[0] : payload.ToArray());
+            var ret = Create(command, payload == null ? new byte[0] : payload.ToArray());
+            ret._payload_deserialized = payload;
+
+            return ret;
         }
 
-        public static Message Create(string command, byte[] payload)
+        public static Message Create(MessageCommand command, byte[] payload)
         {
+            var flags = MessageFlags.None;
+
+            // Try compression
+
+            if (payload.Length > CompressionMinSize)
+            {
+                var compressed = payload.CompressGzip();
+
+                if (compressed.Length < payload.Length - CompressionThreshold)
+                {
+                    payload = compressed;
+                    flags |= MessageFlags.CompressedGzip;
+                }
+            }
+
             return new Message
             {
+                Flags = flags,
                 Command = command,
-                Checksum = GetChecksum(payload),
                 Payload = payload
             };
         }
 
+        void ISerializable.Serialize(BinaryWriter writer)
+        {
+            writer.Write((byte)Flags);
+            writer.Write((byte)Command);
+            writer.WriteVarBytes(Payload);
+        }
+
         void ISerializable.Deserialize(BinaryReader reader)
         {
-            if (reader.ReadUInt32() != Magic)
-                throw new FormatException();
-            this.Command = reader.ReadFixedString(12);
-            uint length = reader.ReadUInt32();
-            if (length > PayloadMaxSize)
-                throw new FormatException();
-            this.Checksum = reader.ReadUInt32();
-            this.Payload = reader.ReadBytes((int)length);
-            if (GetChecksum(Payload) != Checksum)
-                throw new FormatException();
+            this.Flags = (MessageFlags)reader.ReadByte();
+            this.Command = (MessageCommand)reader.ReadByte();
+            var length = (int)reader.ReadVarInt(int.MaxValue);
+
+            if (length > PayloadMaxSize) throw new FormatException();
+            this.Payload = reader.ReadBytes(length);
         }
 
-        private static uint GetChecksum(byte[] value)
+        public static int TryDeserialize(ByteString data, out Message msg)
         {
-            return Crypto.Default.Hash256(value).ToUInt32(0);
+            msg = null;
+            if (data.Count < 3) return 0;
+
+            var header = data.Slice(0, 3).ToArray();
+            ulong length = header[2];
+            int payloadIndex = 3;
+
+            if (length == 0xFD)
+            {
+                if (data.Count < 5) return 0;
+                length = data.Slice(payloadIndex, 2).ToArray().ToUInt16(0);
+                payloadIndex += 2;
+            }
+            else if (length == 0xFE)
+            {
+                if (data.Count < 7) return 0;
+                length = data.Slice(payloadIndex, 4).ToArray().ToUInt32(0);
+                payloadIndex += 4;
+            }
+            else if (length == 0xFF)
+            {
+                if (data.Count < 11) return 0;
+                length = data.Slice(payloadIndex, 8).ToArray().ToUInt64(0);
+                payloadIndex += 8;
+            }
+
+            if (length > PayloadMaxSize) throw new FormatException();
+            if (data.Count < (int)length) return 0;
+
+            msg = new Message()
+            {
+                Flags = (MessageFlags)header[0],
+                Command = (MessageCommand)header[1],
+                Payload = data.Slice(payloadIndex, (int)length).ToArray()
+            };
+
+            return payloadIndex + (int)length;
+        }
+
+        public byte[] GetPayload()
+        {
+            if (this.Flags.HasFlag(MessageFlags.CompressedGzip))
+            {
+                return this.Payload.UncompressGzip();
+            }
+
+            return this.Payload;
         }
 
-        private ISerializable _payload_deserialized = null;
         public T GetPayload<T>() where T : ISerializable, new()
         {
             if (_payload_deserialized is null)
-                _payload_deserialized = Payload.AsSerializable<T>();
+                _payload_deserialized = GetPayload().AsSerializable<T>();
             return (T)_payload_deserialized;
         }
 
         public Transaction GetTransaction()
         {
             if (_payload_deserialized is null)
-                _payload_deserialized = Transaction.DeserializeFrom(Payload);
+                _payload_deserialized = Transaction.DeserializeFrom(GetPayload());
             return (Transaction)_payload_deserialized;
         }
-
-        void ISerializable.Serialize(BinaryWriter writer)
-        {
-            writer.Write(Magic);
-            writer.WriteFixedString(Command, 12);
-            writer.Write(Payload.Length);
-            writer.Write(Checksum);
-            writer.Write(Payload);
-        }
     }
-}
+}

neo/Network/P2P/MessageCommand.cs

@@ -0,0 +1,36 @@
+namespace Neo.Network.P2P
+{
+    public enum MessageCommand : byte
+    {
+        // Same value as InventoryType
+        tx = 0x01,
+        // Same value as InventoryType
+        block = 0x02,
+        mempool = 0x03,
+        addr = 0x04,
+        inv = 0x05,
+        headers = 0x06,
+        merkleblock = 0x07,
+        version = 0x08,
+        verack = 0x09,
+        alert = 0x0A,
+        reject = 0x0B,
+
+        ping = 0x10,
+        pong = 0x11,
+
+        getaddr = 0x20,
+        getblocks = 0x21,
+        getdata = 0x22,
+        getheaders = 0x23,
+
+        filteradd = 0x30,
+        filterclear = 0x31,
+        filterload = 0x32,
+
+        // Same value as InventoryType
+        consensus = 0xE0,
+
+        notfound = 0xFF,
+    }
+}

neo/Network/P2P/MessageFlags.cs

@@ -0,0 +1,11 @@
+using System;
+
+namespace Neo.Network.P2P
+{
+    [Flags]
+    public enum MessageFlags : byte
+    {
+        None = 0,
+        CompressedGzip = 1 << 0
+    }
+}