Merge pull request #3399 from AElfProject/feature/grpc-bidirectional-streaming

Feature/grpc bidirectional streaming

Author: jason-aelf

azure-pipelines.yml

@@ -9,7 +9,7 @@ jobs:
     parameters:
       parts: 3
       n: 2
-      codecoverage: true
+      codecoverage: false
   - template: templates/build-template-window.yml
     parameters:
       parts: 3

protobuf/peer_service.proto

@@ -7,25 +7,27 @@ import "aelf/core.proto";
 
 service PeerService {
 
-    rpc Ping (PingRequest) returns (PongReply) {}
-    rpc CheckHealth (HealthCheckRequest) returns (HealthCheckReply) {}
-    
-    rpc RequestBlock (BlockRequest) returns (BlockReply) {}
-    rpc RequestBlocks (BlocksRequest) returns (BlockList) {}
+  rpc Ping (PingRequest) returns (PongReply) {}
+  rpc CheckHealth (HealthCheckRequest) returns (HealthCheckReply) {}
 
-    rpc BlockBroadcastStream (stream BlockWithTransactions) returns (VoidReply) {}
-    
-    rpc TransactionBroadcastStream (stream aelf.Transaction) returns (VoidReply) {}
-    rpc AnnouncementBroadcastStream (stream BlockAnnouncement) returns (VoidReply) {}
+  rpc RequestBlock (BlockRequest) returns (BlockReply) {}
+  rpc RequestBlocks (BlocksRequest) returns (BlockList) {}
 
-    rpc LibAnnouncementBroadcastStream (stream LibAnnouncement) returns (VoidReply) {}
+  rpc BlockBroadcastStream (stream BlockWithTransactions) returns (VoidReply) {}
 
-    rpc GetNodes (NodesRequest) returns (NodeList) {}
+  rpc TransactionBroadcastStream (stream aelf.Transaction) returns (VoidReply) {}
+  rpc AnnouncementBroadcastStream (stream BlockAnnouncement) returns (VoidReply) {}
 
-    rpc DoHandshake (HandshakeRequest) returns (HandshakeReply) {}
-    rpc ConfirmHandshake (ConfirmHandshakeRequest) returns (VoidReply) {}
+  rpc LibAnnouncementBroadcastStream (stream LibAnnouncement) returns (VoidReply) {}
 
-    rpc Disconnect (DisconnectReason) returns (VoidReply) {}
+  rpc RequestByStream (stream StreamMessage) returns (stream StreamMessage) {}
+
+  rpc GetNodes (NodesRequest) returns (NodeList) {}
+
+  rpc DoHandshake (HandshakeRequest) returns (HandshakeReply) {}
+  rpc ConfirmHandshake (ConfirmHandshakeRequest) returns (VoidReply) {}
+
+  rpc Disconnect (DisconnectReason) returns (VoidReply) {}
 }
 
 // **** No reply *****
@@ -45,3 +47,34 @@ message HealthCheckRequest {
 message HealthCheckReply {
 }
 
+message StreamMessage {
+  StreamType stream_type = 1;
+  MessageType message_type = 2;
+  string request_id = 3;
+  bytes message = 4;
+  map<string, string> meta = 5;
+}
+
+enum StreamType {
+  UNKNOWN = 0;
+  REQUEST = 1;
+  REPLY = 2;
+}
+
+enum MessageType {
+  ANY = 0;
+
+  HAND_SHAKE = 1;
+  PING = 2;
+  CONFIRM_HAND_SHAKE = 3;
+  HEALTH_CHECK = 4;
+  REQUEST_BLOCK = 5;
+  REQUEST_BLOCKS = 6;
+  GET_NODES = 7;
+
+  BLOCK_BROADCAST = 8;
+  TRANSACTION_BROADCAST = 9;
+  ANNOUNCEMENT_BROADCAST = 10;
+  LIB_ANNOUNCEMENT_BROADCAST = 11;
+  DISCONNECT = 12;
+}

src/AElf.Kernel.Core/Blockchain/Application/IBlockValidationProvider.cs

@@ -157,7 +157,7 @@ await _transactionBlockIndexService.ValidateTransactionBlockIndexExistsInBranchA
                     block.Header.PreviousBlockHash);
             if (!blockIndexExists)
                 continue;
-            Logger.LogDebug("Transaction: {TransactionId} repackaged", transactionId);
+            Logger.LogDebug("Transaction: {TransactionId} repackaged", transactionId.ToHex());
             return false;
         }
 

src/AElf.Kernel.Core/SmartContract/Application/IBlockchainStateService.cs

@@ -158,7 +158,7 @@ public async Task MergeBlockStateAsync(long lastIrreversibleBlockHeight, Hash la
 
         Logger.LogDebug(
             "Start merge lib height: {LastIrreversibleBlockHeight}, lib block hash: {LastIrreversibleBlockHash}, merge count: {BlockIndexesCount}",
-            lastIrreversibleBlockHeight, lastIrreversibleBlockHash, blockIndexes.Count);
+            lastIrreversibleBlockHeight, lastIrreversibleBlockHash.ToHex(), blockIndexes.Count);
 
         foreach (var blockIndex in blockIndexes)
             try

src/AElf.Kernel.SmartContractExecution/Application/BlockchainExecutingService.cs

@@ -112,7 +112,7 @@ await _transactionResultService.GetTransactionResultAsync(transactionId, blockHa
                 == null)
             {
                 Logger.LogWarning(
-                    $"Fail to load transaction result. block hash : {blockHash}, tx id: {transactionId}");
+                    "Fail to load transaction result. block hash : {blockHash}, tx id: {transactionId}", blockHash.ToHex(), transactionId.ToHex());
 
                 return null;
             }

src/AElf.Kernel.Types/Block/BlockHeader.cs

@@ -35,7 +35,7 @@ public byte[] GetHashBytes()
     private byte[] GetSignatureData()
     {
         if (!VerifyFields())
-            throw new InvalidOperationException($"Invalid block header: {this}.");
+            throw new InvalidOperationException($"Invalid block header: PreviousBlockHash={PreviousBlockHash?.ToHex()}, mtr={MerkleTreeRootOfTransactions?.ToHex()}, ChainId={ChainId}, Height={Height}, Time={Time}.");
 
         if (Signature.IsEmpty)
             return this.ToByteArray();

src/AElf.Kernel.Types/KernelConstants.cs

@@ -1,3 +1,4 @@
+using System;
 using Google.Protobuf.WellKnownTypes;
 
 namespace AElf.Kernel;
@@ -16,4 +17,5 @@ public static class KernelConstants
     public const string SignaturePlaceholder = "SignaturePlaceholder";
     public const string BlockExecutedDataKey = "BlockExecutedData";
     public static Duration AllowedFutureBlockTimeSpan = new() { Seconds = 4 };
+    public static string SupportStreamMinVersion = "1.4.0.0";
 }

src/AElf.Launcher/AElf.Launcher.csproj

@@ -5,8 +5,8 @@
         <ServerGarbageCollection>true</ServerGarbageCollection>
     </PropertyGroup>
     <ItemGroup>
-        <ProjectReference Include="..\AElf.Blockchains.MainChain\AElf.Blockchains.MainChain.csproj"/>
-        <ProjectReference Include="..\AElf.Blockchains.SideChain\AElf.Blockchains.SideChain.csproj"/>
+        <ProjectReference Include="..\AElf.Blockchains.MainChain\AElf.Blockchains.MainChain.csproj" />
+        <ProjectReference Include="..\AElf.Blockchains.SideChain\AElf.Blockchains.SideChain.csproj" />
         <None Update="Dockerfile">
             <CopyToOutputDirectory>Always</CopyToOutputDirectory>
         </None>
@@ -15,9 +15,9 @@
         </None>
     </ItemGroup>
     <ItemGroup>
-        <PackageReference Include="Volo.Abp.AspNetCore.Mvc" Version="5.2.2"/>
-        <PackageReference Include="Volo.Abp.Autofac" Version="5.2.2"/>
-        <FrameworkReference Include="Microsoft.AspNetCore.App"/>
+        <PackageReference Include="Volo.Abp.AspNetCore.Mvc" Version="5.2.2" />
+        <PackageReference Include="Volo.Abp.Autofac" Version="5.2.2" />
+        <FrameworkReference Include="Microsoft.AspNetCore.App" />
     </ItemGroup>
     <ItemGroup>
         <Content Include="appsettings.Development.json">

src/AElf.OS.Core/Network/Application/INetworkService.cs

@@ -1,6 +1,7 @@
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using AElf.Kernel;
+using AElf.OS.Network.Infrastructure;
 using AElf.OS.Network.Types;
 using AElf.Types;
 
@@ -28,4 +29,5 @@ Task<bool> RemovePeerByPubkeyAsync(string peerPubkey,
     Task CheckPeersHealthAsync();
     void CheckNtpDrift();
     bool IsPeerPoolFull();
+    Task<List<NodeInfo>> GetNodesAsync(IPeer peer);
 }

src/AElf.OS.Core/Network/Application/NetworkService.cs

@@ -142,7 +142,7 @@ public Task BroadcastAnnounceAsync(BlockHeader blockHeader)
                         Logger.LogInformation(ex, $"Could not broadcast announcement to {peer} " +
                                                   $"- status {peer.ConnectionStatus}.");
 
-                        await HandleNetworkException(peer, ex);
+                        await HandleNetworkExceptionAsync(peer, ex);
                     }
                 });
             }
@@ -171,7 +171,7 @@ public Task BroadcastTransactionAsync(Transaction transaction)
                         Logger.LogWarning(ex, $"Could not broadcast transaction to {peer} " +
                                               $"- status {peer.ConnectionStatus}.");
 
-                        await HandleNetworkException(peer, ex);
+                        await HandleNetworkExceptionAsync(peer, ex);
                     }
                 });
             }
@@ -201,7 +201,7 @@ public Task BroadcastLibAnnounceAsync(Hash libHash, long libHeight)
                     {
                         Logger.LogWarning(ex, $"Could not broadcast lib announcement to {peer} " +
                                               $"- status {peer.ConnectionStatus}.");
-                        await HandleNetworkException(peer, ex);
+                        await HandleNetworkExceptionAsync(peer, ex);
                     }
                 });
             }
@@ -256,7 +256,6 @@ public async Task<Response<List<BlockWithTransactions>>> GetBlocksAsync(Hash pre
 
         if (peer == null)
             throw new InvalidOperationException($"Could not find peer {peerPubkey}.");
-
         var response = await Request(peer, p => p.GetBlocksAsync(previousBlock, count));
 
         if (response.Success && response.Payload != null
@@ -343,7 +342,7 @@ private void EnqueueBlock(IPeer peer, BlockWithTransactions blockWithTransaction
                 if (ex != null)
                 {
                     Logger.LogWarning(ex, $"Could not broadcast block to {peer} - status {peer.ConnectionStatus}.");
-                    await HandleNetworkException(peer, ex);
+                    await HandleNetworkExceptionAsync(peer, ex);
                 }
             });
         }
@@ -372,13 +371,33 @@ private async Task<Response<T>> Request<T>(IPeer peer, Func<IPeer, Task<T>> func
             if (ex.ExceptionType == NetworkExceptionType.HandlerException)
                 return new Response<T>(default);
 
-            await HandleNetworkException(peer, ex);
+            await HandleNetworkExceptionAsync(peer, ex);
         }
 
         return new Response<T>();
     }
 
-    private async Task HandleNetworkException(IPeer peer, NetworkException exception)
+    public async Task<List<NodeInfo>> GetNodesAsync(IPeer peer)
+    {
+        try
+        {
+            var nodeList = await peer.GetNodesAsync();
+
+            if (nodeList?.Nodes == null)
+                return new List<NodeInfo>();
+
+            Logger.LogDebug("get nodes: {nodeList} from peer: {peer}.", nodeList, peer);
+            return nodeList.Nodes.ToList();
+        }
+        catch (Exception e)
+        {
+            if (e is NetworkException exception) await HandleNetworkExceptionAsync(peer, exception);
+            Logger.LogWarning(e, "get nodes failed. peer={peer}", peer);
+            return new List<NodeInfo>();
+        }
+    }
+
+    private async Task HandleNetworkExceptionAsync(IPeer peer, NetworkException exception)
     {
         if (exception.ExceptionType == NetworkExceptionType.Unrecoverable)
         {
@@ -398,8 +417,9 @@ private async Task RecoverPeerAsync(IPeer peer)
             return;
 
         var success = await peer.TryRecoverAsync();
-
-        if (!success)
+        if (success)
+            await _networkServer.BuildStreamForPeerAsync(peer);
+        else
             await _networkServer.TrySchedulePeerReconnectionAsync(peer);
     }