Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ public class BuiltInStepsModule : Module
typeof(InitializeBlockProducer),
typeof(InitializeBlockTree),
typeof(InitializeNetwork),
typeof(StartNetwork),
typeof(InitializePlugins),
typeof(InitializePrecompiles),
typeof(InitTxTypesAndRlp),
Expand Down
98 changes: 0 additions & 98 deletions src/Nethermind/Nethermind.Init/Steps/InitializeNetwork.cs
Original file line number Diff line number Diff line change
Expand Up @@ -52,17 +52,13 @@ public class InitializeNetwork : IStep
{
private readonly IApiWithNetwork _api;
private readonly INodeStatsManager _nodeStatsManager;
private readonly ISynchronizer _synchronizer;
private readonly ISyncPeerPool _syncPeerPool;
private readonly IForkInfo _forkInfo;
private readonly NodeSourceToDiscV4Feeder _enrDiscoveryAppFeeder;
private readonly INetworkStorage _peerStorage;
private readonly IDiscoveryApp _discoveryApp;
private readonly Lazy<IPeerPool> _peerPool;

private readonly INetworkConfig _networkConfig;
private readonly ISyncConfig _syncConfig;
private readonly IInitConfig _initConfig;

private readonly ILogger _logger;

Expand All @@ -85,16 +81,12 @@ ILogManager logManager
{
_api = api;
_nodeStatsManager = nodeStatsManager;
_synchronizer = synchronizer;
_syncPeerPool = syncPeerPool;
_enrDiscoveryAppFeeder = enrDiscoveryAppFeeder;
_discoveryApp = discoveryApp;
_peerPool = peerPool;
_forkInfo = forkInfo;
_peerStorage = peerStorage;
_networkConfig = networkConfig;
_syncConfig = syncConfig;
_initConfig = initConfig;

_logger = logManager.GetClassLogger();
}
Expand Down Expand Up @@ -150,41 +142,6 @@ await InitPeer().ContinueWith(initPeerTask =>
return;
}

await StartSync().ContinueWith(initNetTask =>
{
if (initNetTask.IsFaulted)
{
_logger.Error("Unable to start the synchronizer.", initNetTask.Exception);
}
});

if (cancellationToken.IsCancellationRequested)
{
return;
}

await StartDiscovery().ContinueWith(initDiscoveryTask =>
{
if (initDiscoveryTask.IsFaulted)
{
_logger.Error("Unable to start the discovery protocol.", initDiscoveryTask.Exception);
}
});

try
{
if (cancellationToken.IsCancellationRequested)
{
return;
}

StartPeer();
}
catch (Exception e)
{
_logger.Error("Unable to start the peer manager.", e);
}

if (_api.Enode is null)
{
throw new InvalidOperationException("Cannot initialize network without knowing own enode");
Expand All @@ -199,61 +156,6 @@ await StartDiscovery().ContinueWith(initDiscoveryTask =>
ThisNodeInfo.AddInfo("Node address :", $"{_api.Enode.Address} (do not use as an account)");
}

private Task StartDiscovery()
{
if (!_initConfig.DiscoveryEnabled)
{
if (_logger.IsWarn) _logger.Warn($"Skipping discovery init due to {nameof(IInitConfig.DiscoveryEnabled)} set to false");
return Task.CompletedTask;
}

if (_logger.IsDebug) _logger.Debug("Starting discovery process.");
_ = _discoveryApp.StartAsync();
if (_logger.IsDebug) _logger.Debug("Discovery process started.");
return Task.CompletedTask;
}

private void StartPeer()
{
if (_api.PeerManager is null) throw new StepDependencyException(nameof(_api.PeerManager));
if (_api.SessionMonitor is null) throw new StepDependencyException(nameof(_api.SessionMonitor));

if (!_api.Config<IInitConfig>().PeerManagerEnabled)
{
if (_logger.IsWarn) _logger.Warn($"Skipping peer manager init due to {nameof(IInitConfig.PeerManagerEnabled)} set to false");
}

if (_logger.IsDebug) _logger.Debug("Initializing peer manager");
_peerPool.Value.Start();
_api.PeerManager.Start();
_api.SessionMonitor.Start();
if (_logger.IsDebug) _logger.Debug("Peer manager initialization completed");
}

private Task StartSync()
{
if (_api.BlockTree is null) throw new StepDependencyException(nameof(_api.BlockTree));

if (_syncConfig.NetworkingEnabled)
{
_syncPeerPool.Start();

if (_syncConfig.SynchronizationEnabled)
{
if (_logger.IsDebug) _logger.Debug($"Starting synchronization from block {_api.BlockTree.Head?.Header.ToString(BlockHeader.Format.Short)}.");
_synchronizer.Start();
}
else
{
if (_logger.IsWarn) _logger.Warn($"Skipping blockchain synchronization init due to {nameof(ISyncConfig.SynchronizationEnabled)} set to false");
}
}
else if (_logger.IsWarn) _logger.Warn($"Skipping connecting to peers due to {nameof(ISyncConfig.NetworkingEnabled)} set to false");


return Task.CompletedTask;
}

private async Task InitPeer()
{
if (_api.BlockTree is null) throw new StepDependencyException(nameof(_api.BlockTree));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@

namespace Nethermind.Init.Steps;

[RunnerStepDependencies(typeof(InitializeNetwork), typeof(SetupKeyStore), typeof(InitializeBlockchain), typeof(InitializePlugins), typeof(InitializeBlockProducer), typeof(RegisterRpcModules))]
[RunnerStepDependencies(typeof(SetupKeyStore), typeof(InitializeBlockchain), typeof(InitializePlugins), typeof(InitializeBlockProducer), typeof(RegisterRpcModules))]

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It doesn't still need Initialize? As Start is separated.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

RegisterRpcModules depends in Initialize

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok so it is there implicitly, could be left explicitly but whatever.

public class RegisterPluginRpcModules(
IRpcModuleProvider rpcModuleProvider,
IInitConfig initConfig,
Expand All @@ -28,9 +28,11 @@ public virtual async Task Execute(CancellationToken cancellationToken)
{
// Ugly temporary hack to not receive engine API messages before end of processing of all blocks after restart.
// Then we will wait 5s more to ensure everything is processed
while (!blockProcessingQueue!.IsEmpty)
await Task.Delay(100);
await Task.Delay(5000);
while (!blockProcessingQueue.IsEmpty && !cancellationToken.IsCancellationRequested)
await Task.Delay(100, cancellationToken);

await Task.Delay(5000, cancellationToken);
cancellationToken.ThrowIfCancellationRequested();
Comment thread
LukaszRozmej marked this conversation as resolved.
Outdated
}

foreach (INethermindPlugin plugin in plugins)
Expand Down
125 changes: 125 additions & 0 deletions src/Nethermind/Nethermind.Init/Steps/StartNetwork.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,125 @@
// SPDX-FileCopyrightText: 2025 Demerzel Solutions Limited
// SPDX-License-Identifier: LGPL-3.0-only

using System;
using System.Threading;
using System.Threading.Tasks;
using Nethermind.Api;
using Nethermind.Api.Steps;
using Nethermind.Blockchain.Synchronization;
using Nethermind.Core;
using Nethermind.Logging;
using Nethermind.Network;
using Nethermind.Synchronization;
using Nethermind.Synchronization.Peers;

namespace Nethermind.Init.Steps;

[RunnerStepDependencies(typeof(RegisterPluginRpcModules))]
public class StartNetwork(
INethermindApi api,
ISyncPeerPool syncPeerPool,
ISynchronizer synchronizer,
IDiscoveryApp discoveryApp,
Lazy<IPeerPool> peerPool,
ISyncConfig syncConfig,
IInitConfig initConfig,
ILogManager logManager) : IStep
{

ILogger _logger = logManager.GetClassLogger();

public async Task Execute(CancellationToken cancellationToken)
{
await StartSync().ContinueWith(initNetTask =>
{
if (initNetTask.IsFaulted)
{
_logger.Error("Unable to start the synchronizer.", initNetTask.Exception);
}
});

if (cancellationToken.IsCancellationRequested)
{
return;
}

await StartDiscovery().ContinueWith(initDiscoveryTask =>
{
if (initDiscoveryTask.IsFaulted)
{
_logger.Error("Unable to start the discovery protocol.", initDiscoveryTask.Exception);
}
});

try
{
if (cancellationToken.IsCancellationRequested)
{
return;
}

StartPeer();
}
catch (Exception e)
{
_logger.Error("Unable to start the peer manager.", e);
}
}


private Task StartDiscovery()
{
if (!initConfig.DiscoveryEnabled)
{
if (_logger.IsWarn) _logger.Warn($"Skipping discovery init due to {nameof(IInitConfig.DiscoveryEnabled)} set to false");
return Task.CompletedTask;
}

if (_logger.IsDebug) _logger.Debug("Starting discovery process.");
_ = discoveryApp.StartAsync();
if (_logger.IsDebug) _logger.Debug("Discovery process started.");
return Task.CompletedTask;
}

private void StartPeer()
{
if (api.PeerManager is null) throw new StepDependencyException(nameof(api.PeerManager));
if (api.SessionMonitor is null) throw new StepDependencyException(nameof(api.SessionMonitor));

if (!api.Config<IInitConfig>().PeerManagerEnabled)
{
if (_logger.IsWarn) _logger.Warn($"Skipping peer manager init due to {nameof(IInitConfig.PeerManagerEnabled)} set to false");
}

if (_logger.IsDebug) _logger.Debug("Initializing peer manager");
peerPool.Value.Start();
api.PeerManager.Start();
api.SessionMonitor.Start();
if (_logger.IsDebug) _logger.Debug("Peer manager initialization completed");
}

private Task StartSync()
{
if (api.BlockTree is null) throw new StepDependencyException(nameof(api.BlockTree));

if (syncConfig.NetworkingEnabled)
{
syncPeerPool.Start();

if (syncConfig.SynchronizationEnabled)
{
if (_logger.IsDebug) _logger.Debug($"Starting synchronization from block {api.BlockTree.Head?.Header.ToString(BlockHeader.Format.Short)}.");
synchronizer.Start();
}
else
{
if (_logger.IsWarn) _logger.Warn($"Skipping blockchain synchronization init due to {nameof(ISyncConfig.SynchronizationEnabled)} set to false");
}
}
else if (_logger.IsWarn) _logger.Warn($"Skipping connecting to peers due to {nameof(ISyncConfig.NetworkingEnabled)} set to false");


return Task.CompletedTask;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@

namespace Nethermind.Runner.Ethereum.Steps;

[RunnerStepDependencies(typeof(InitializeNetwork), typeof(RegisterRpcModules), typeof(RegisterPluginRpcModules))]
[RunnerStepDependencies(typeof(StartNetwork), typeof(RegisterRpcModules), typeof(RegisterPluginRpcModules))]
public class StartRpc(INethermindApi api, IJsonRpcServiceConfigurer[] serviceConfigurers, IWebSocketsManager webSocketsManager) : IStep
{
public async Task Execute(CancellationToken cancellationToken)
Expand Down