Guarantee at least 10 peers to broadcast

CHANGES.md

@@ -123,6 +123,11 @@ To be released.
     serialization.  [[#762]]
  -  `Swarm<T>` became to ignore broadcasted block that has lower index than
     the current tip.  [[#764]]
+ -  The way `Swarm<T>` chose peers to spread messages has changed.  [[#765], [#767]]
+     -  If there are less than 10 peers in the routing table, select all peers.
+     -  If there are more than 10 peers in the routing table,
+        choose one from each bucket, and if the number is less than 10,
+        then select an additional peers so that the total is 10.
 
 ### Bug fixes
 
@@ -207,6 +212,8 @@ To be released.
 [#762]: https://github.com/planetarium/libplanet/pull/762
 [#763]: https://github.com/planetarium/libplanet/pull/763
 [#764]: https://github.com/planetarium/libplanet/pull/764
+[#765]: https://github.com/planetarium/libplanet/issues/765
+[#767]: https://github.com/planetarium/libplanet/pull/767
 
 
 Version 0.7.0

Libplanet.Tests/Net/Protocols/ProtocolTest.cs

@@ -351,15 +351,108 @@ public async Task BroadcastMessage(int count)
             }
         }
 
+        [Fact(Timeout = Timeout)]
+        public async Task BroadcastGuarantee()
+        {
+            // Make sure t1 and t2 is in same bucket of seed's routing table.
+            var privateKey0 = new PrivateKey(new byte[]
+            {
+                0x1a, 0x55, 0x30, 0x84, 0xe8, 0x9e, 0xee, 0x1e, 0x9f, 0xe2, 0xd1, 0x49, 0xe7, 0xa9,
+                0x53, 0xa9, 0xb4, 0xe4, 0xfe, 0x5a, 0xc1, 0x6c, 0x61, 0x9f, 0x54, 0x8f, 0x5e, 0xd9,
+                0x7f, 0xa3, 0xa0, 0x79,
+            });
+            var privateKey1 = new PrivateKey(new byte[]
+            {
+                0x8e, 0x26, 0x31, 0x4a, 0xee, 0x84, 0xd, 0x8a, 0xea, 0x7b, 0x6, 0xf8, 0x81, 0x5f,
+                0x69, 0xb3, 0x44, 0x46, 0xe0, 0x27, 0x65, 0x17, 0x1, 0x16, 0x58, 0x26, 0x69, 0x93,
+                0x48, 0xbb, 0xf, 0xb4,
+            });
+            var privateKey2 = new PrivateKey(new byte[]
+            {
+                0xd4, 0x6b, 0x4b, 0x38, 0xde, 0x39, 0x25, 0x3b, 0xd8, 0x1, 0x9d, 0x2, 0x2, 0x7a,
+                0x90, 0x9, 0x46, 0x2f, 0xc1, 0xd3, 0xd9, 0xa, 0xa6, 0xf4, 0xfa, 0x9a, 0x6, 0xa3,
+                0x60, 0xed, 0xf3, 0xd7,
+            });
+
+            var seed = CreateTestTransport(privateKey0);
+            var t1 = CreateTestTransport(privateKey1, true);
+            var t2 = CreateTestTransport(privateKey2);
+            await StartTestTransportAsync(seed);
+            await StartTestTransportAsync(t1);
+            await StartTestTransportAsync(t2);
+
+            try
+            {
+                await t1.BootstrapAsync(new[] { seed.AsPeer });
+                await t2.BootstrapAsync(new[] { seed.AsPeer });
+
+                Log.Debug(seed.Protocol.Trace());
+
+                Log.Debug("Bootstrap completed.");
+
+                var tcs = new CancellationTokenSource();
+                var task = t2.WaitForTestMessageWithData("foo", tcs.Token);
+
+                seed.BroadcastTestMessage(null, "foo");
+                Log.Debug("Broadcast \"foo\" completed.");
+
+                tcs.CancelAfter(TimeSpan.FromSeconds(5));
+                await task;
+
+                Assert.True(t2.ReceivedTestMessageOfData("foo"));
+
+                tcs = new CancellationTokenSource();
+                task = t2.WaitForTestMessageWithData("bar", tcs.Token);
+
+                seed.BroadcastTestMessage(null, "bar");
+                Log.Debug("Broadcast \"bar\" completed.");
+
+                tcs.CancelAfter(TimeSpan.FromSeconds(5));
+                await task;
+
+                Assert.True(t2.ReceivedTestMessageOfData("bar"));
+
+                tcs = new CancellationTokenSource();
+                task = t2.WaitForTestMessageWithData("baz", tcs.Token);
+
+                seed.BroadcastTestMessage(null, "baz");
+                Log.Debug("Broadcast \"baz\" completed.");
+
+                tcs.CancelAfter(TimeSpan.FromSeconds(5));
+                await task;
+
+                Assert.True(t2.ReceivedTestMessageOfData("baz"));
+
+                tcs = new CancellationTokenSource();
+                task = t2.WaitForTestMessageWithData("qux", tcs.Token);
+
+                seed.BroadcastTestMessage(null, "qux");
+                Log.Debug("Broadcast \"qux\" completed.");
+
+                tcs.CancelAfter(TimeSpan.FromSeconds(5));
+                await task;
+
+                Assert.True(t2.ReceivedTestMessageOfData("qux"));
+            }
+            finally
+            {
+                await seed.StopAsync(TimeSpan.Zero);
+                await t1.StopAsync(TimeSpan.Zero);
+                await t2.StopAsync(TimeSpan.Zero);
+            }
+        }
+
         private TestTransport CreateTestTransport(
             PrivateKey privateKey = null,
+            bool blockBroadcast = false,
             int? tableSize = null,
             int? bucketSize = null,
             TimeSpan? networkDelay = null)
         {
             return new TestTransport(
                 _transports,
                 privateKey ?? new PrivateKey(),
+                blockBroadcast,
                 tableSize,
                 bucketSize,
                 networkDelay);

Libplanet.Tests/Net/Protocols/TestTransport.cs

@@ -4,6 +4,7 @@
 using System.Collections.Immutable;
 using System.Linq;
 using System.Net;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Libplanet.Crypto;
@@ -25,18 +26,21 @@ internal class TestTransport : ITransport
         private readonly List<string> _ignoreTestMessageWithData;
         private readonly PrivateKey _privateKey;
         private readonly Random _random;
+        private readonly bool _blockBroadcast;
 
         private CancellationTokenSource _swarmCancellationTokenSource;
         private TimeSpan _networkDelay;
 
         public TestTransport(
             Dictionary<Address, TestTransport> transports,
             PrivateKey privateKey,
+            bool blockBroadcast,
             int? tableSize,
             int? bucketSize,
             TimeSpan? networkDelay)
         {
             _privateKey = privateKey;
+            _blockBroadcast = blockBroadcast;
             var loggerId = _privateKey.PublicKey.ToAddress().ToHex();
             _logger = Log.ForContext<TestTransport>()
                 .ForContext("Address", loggerId);
@@ -248,10 +252,14 @@ public void BroadcastTestMessage(Address? except, string data)
         public void BroadcastMessage(Address? except, Message message)
         {
             var peers = Protocol.PeersToBroadcast(except).ToList();
+            var peersString = peers.Aggregate(
+                new StringBuilder(),
+                (builder, peer) => builder.Append($"{peer.Address.ToHex()}, ")).ToString();
             _logger.Debug(
-                "Broadcasting test message {Data} to {Count} peers.",
+                "Broadcasting test message {Data} to {Count} peers which are: {Peers}.",
                 ((TestMessage)message).Data,
-                peers.Count);
+                peers.Count,
+                peersString);
             foreach (var peer in peers)
             {
                 _requests.Add(new Request()
@@ -369,16 +377,18 @@ public void ReplyMessage(Message message)
             });
         }
 
-        public async Task WaitForTestMessageWithData(string data)
+        public async Task WaitForTestMessageWithData(
+            string data,
+            CancellationToken token = default(CancellationToken))
         {
             if (!Running)
             {
                 throw new SwarmException("Start swarm before use.");
             }
 
-            while (!ReceivedTestMessageOfData(data))
+            while (!token.IsCancellationRequested && !ReceivedTestMessageOfData(data))
             {
-                await Task.Delay(10);
+                await Task.Delay(10, token);
             }
         }
 
@@ -389,7 +399,7 @@ public bool ReceivedTestMessageOfData(string data)
                 throw new SwarmException("Start swarm before use.");
             }
 
-            return ReceivedMessages.OfType<TestMessage>().Select(msg => msg.Data == data).Any();
+            return ReceivedMessages.OfType<TestMessage>().Any(msg => msg.Data == data);
         }
 
         private void ReceiveMessage(Message message)
@@ -415,7 +425,11 @@ private void ReceiveMessage(Message message)
                 {
                     _logger.Debug("Received test message with {Data}.", testMessage.Data);
                     _ignoreTestMessageWithData.Add(testMessage.Data);
-                    BroadcastTestMessage(testMessage.Remote.Address, testMessage.Data);
+                    // If this transport is blocked for testing, do not broadcast.
+                    if (!_blockBroadcast)
+                    {
+                        BroadcastTestMessage(testMessage.Remote.Address, testMessage.Data);
+                    }
                 }
             }
             else

Libplanet/Net/Protocols/RoutingTable.cs

@@ -7,6 +7,9 @@ namespace Libplanet.Net.Protocols
 {
     internal class RoutingTable
     {
+        // FIXME: This would be configurable.
+        private const int _minPeersToBroadcast = 10;
+
         private readonly Address _address;
         private readonly int _tableSize;
         private readonly int _bucketSize;
@@ -80,9 +83,18 @@ private IEnumerable<KBucket> NonEmptyBuckets
 
         public IEnumerable<BoundPeer> PeersToBroadcast(Address? except)
         {
-            return NonEmptyBuckets
+            List<BoundPeer> peers = NonEmptyBuckets
                 .Select(bucket => bucket.GetRandomPeer(except))
-                .Where(peer => !(peer is null));
+                .Where(peer => !(peer is null)).ToList();
+            var count = peers.Count;
+            if (count < _minPeersToBroadcast)
+            {
+                peers.AddRange(Peers
+                    .Where(peer => except is null || !peer.Address.Equals(except.Value))
+                    .Take(_minPeersToBroadcast - count));
+            }
+
+            return peers;
         }
 
         public IEnumerable<BoundPeer> PeersToRefresh(TimeSpan maxAge)