Merge pull request #1451 from longfin/bugfix/trnasport-throughput

Non-blocking NetMQTransport.ProcessMessage & Release 0.15.1
planetarium · Aug 27, 2021 · 61fb4ac · 61fb4ac
2 parents 2dcfb80 + 9ab824f
commit 61fb4ac
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 16 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -4,7 +4,11 @@ Libplanet changelog
 Version 0.15.1
 --------------
 
-To be released.
+Released on August 28, 2021.
+
+ -  `NetMQTransport` became to process message in non blocking way.  [[#1451]]
+
+ [#1451]: https://github.com/planetarium/libplanet/pull/1451
 
 
 Version 0.15.0

diff --git a/Libplanet/Net/Swarm.MessageHandlers.cs b/Libplanet/Net/Swarm.MessageHandlers.cs
@@ -46,9 +46,9 @@ private void ProcessMessageHandler(object target, Message message)
                 {
                     const string msg =
                         "Received a " + nameof(GetBlockHashes) + " message " +
-                        "(locator: {Locator}, stop: {Stop}).";
+                        "(stop: {Stop}).";
                     BlockHash[] locatorArray = getBlockHashes.Locator.ToArray();
-                    _logger.Debug(msg, locatorArray, getBlockHashes.Stop);
+                    _logger.Debug(msg, getBlockHashes.Stop);
                     BlockChain.FindNextHashes(
                         getBlockHashes.Locator,
                         getBlockHashes.Stop,
@@ -58,13 +58,14 @@ private void ProcessMessageHandler(object target, Message message)
                         out IReadOnlyList<BlockHash> hashes
                     );
                     const string resultMsg =
-                        "Found hashes after the branchpoint (locator: {Locator}, stop: {Stop}): " +
+                        "Found hashes after the branchpoint (stop: {Stop}): " +
                         "{Hashes} (offset: {Offset}.";
-                    _logger.Debug(resultMsg, locatorArray, getBlockHashes.Stop, hashes, offset);
+                    _logger.Debug(resultMsg, getBlockHashes.Stop, hashes, offset);
                     var reply = new BlockHashes(offset, hashes)
                     {
                         Identity = getBlockHashes.Identity,
                     };
+
                     Transport.ReplyMessage(reply);
                     break;
                 }

diff --git a/Libplanet/Net/Transports/NetMQTransport.cs b/Libplanet/Net/Transports/NetMQTransport.cs
@@ -566,18 +566,21 @@ private void ReceiveMessage(object sender, NetMQSocketEventArgs e)
                 MessageHistory.Enqueue(message);
                 LastMessageTimestamp = DateTimeOffset.UtcNow;
 
-                try
-                {
-                    ProcessMessageHandler?.Invoke(this, message);
-                }
-                catch (Exception exc)
+                Task.Run(() =>
                 {
-                    _logger.Error(
-                        exc,
-                        "Something went wrong during message parsing: {0}",
-                        exc);
-                    throw;
-                }
+                    try
+                    {
+                        ProcessMessageHandler?.Invoke(this, message);
+                    }
+                    catch (Exception exc)
+                    {
+                        _logger.Error(
+                            exc,
+                            "Something went wrong during message parsing: {0}",
+                            exc);
+                        throw;
+                    }
+                });
             }
             catch (DifferentAppProtocolVersionException dapve)
             {