Remove BlockDownloader Service, rather let the wallet service download and manage blocks
MODIFY
MagicalCryptoWallet.Tests/P2pTests.cs
MagicalCryptoWallet.Tests/P2pTests.cs
@@ -57,7 +57,6 @@ namespace MagicalCryptoWallet.Tests
var blocksFolderPath = Path.Combine(SharedFixture.DataDir, $"Blocks{network}");
var connectionParameters = new NodeConnectionParameters();
AddressManager addressManager = null;
- BlockDownloader downloader = null;
try
{
addressManager = AddressManager.LoadPeerFile(addressManagerFilePath);
@@ -86,13 +85,12 @@ namespace MagicalCryptoWallet.Tests
RequiredServices = NodeServices.Network,
MinVersion = ProtocolVersion.WITNESS_VERSION
});
- downloader = new BlockDownloader(network, nodes, blocksFolderPath);
+ WalletService walletService = new WalletService(
+ KeyManager.CreateNew(out Mnemonic mnemonic, "password"),
+ new IndexDownloader(network, "foo.txt", new Uri("http://localhost:12345")),
+ nodes,
+ blocksFolderPath);
Assert.True(Directory.Exists(blocksFolderPath));
- downloader.Synchronize();
- foreach (var hash in blocksToDownload)
- {
- downloader.QueueToDownload(hash);
- }
try
{
@@ -127,35 +125,28 @@ namespace MagicalCryptoWallet.Tests
foreach (var hash in blocksToDownload)
{
- times = 0;
- while (downloader.GetBlock(hash) == null)
+ using (var cts = new CancellationTokenSource(TimeSpan.FromMinutes(3)))
{
- if (times > 1800) // 3 minutes
- {
- throw new TimeoutException($"{nameof(BlockDownloader)} test timed out.");
- }
- await Task.Delay(100);
- times++;
- }
+ var block = await walletService.GetOrDownloadBlockAsync(hash, cts.Token);
Assert.True(File.Exists(Path.Combine(blocksFolderPath, hash.ToString())));
Logger.LogInfo<P2pTests>($"Full block is downloaded: {hash}.");
}
}
+
+ }
finally
{
nodes.ConnectedNodes.Added -= ConnectedNodes_Added;
nodes.ConnectedNodes.Removed -= ConnectedNodes_Removed;
memPoolService.TransactionReceived -= MemPoolService_TransactionReceived;
- if(downloader != null)
- {
- await downloader.StopAsync();
- }
// So next test will download the block.
foreach (var hash in blocksToDownload)
{
- downloader?.TryRemove(hash);
+ await walletService?.DeleteBlockAsync(hash);
}
+ walletService?.Dispose();
+
if (Directory.Exists(blocksFolderPath))
{
Directory.Delete(blocksFolderPath, recursive: true);

MODIFY
MagicalCryptoWallet.Tests/RegTests.cs
MagicalCryptoWallet.Tests/RegTests.cs
@@ -156,73 +156,6 @@ namespace MagicalCryptoWallet.Tests
#region ServicesTests
- [Fact]
- public async Task DownloaderAsync()
- {
- var blocksToDownload = new HashSet<uint256>();
- for (int i = 0; i < 101; i++)
- {
- var hash = await Global.RpcClient.GetBlockHashAsync(i);
- blocksToDownload.Add(hash);
- }
- var blocksFolderPath = Path.Combine(SharedFixture.DataDir, nameof(DownloaderAsync), $"Blocks");
- BlockDownloader downloader = null;
- var nodes = new NodesGroup(Global.Config.Network,
- requirements: new NodeRequirement
- {
- RequiredServices = NodeServices.Network,
- MinVersion = ProtocolVersion.WITNESS_VERSION
- });
- try
- {
- nodes.ConnectedNodes.Add(Fixture.BackendRegTestNode.CreateNodeClient());
-
- downloader = new BlockDownloader(Global.Config.Network, nodes, blocksFolderPath);
- downloader.Synchronize();
- foreach (var hash in blocksToDownload)
- {
- downloader.QueueToDownload(hash);
- }
-
- nodes.Connect();
-
- foreach (var hash in blocksToDownload)
- {
- var times = 0;
- while (downloader.GetBlock(hash) == null)
- {
- if (times > 300) // 30 seconds
- {
- throw new TimeoutException($"{nameof(BlockDownloader)} test timed out.");
- }
- await Task.Delay(100);
- times++;
- }
- Assert.True(File.Exists(Path.Combine(blocksFolderPath, hash.ToString())));
- }
- Logger.LogInfo<P2pTests>($"All RegTest block is downloaded.");
- }
- finally
- {
- if(downloader != null)
- {
- await downloader.StopAsync();
- }
-
- // So next test will download the block.
- foreach (var hash in blocksToDownload)
- {
- downloader?.TryRemove(hash);
- }
- if (Directory.Exists(blocksFolderPath))
- {
- Directory.Delete(blocksFolderPath, recursive: true);
- }
-
- nodes?.Dispose();
- }
- }
-
[Fact]
public async Task MempoolAsync()
{
@@ -494,25 +427,22 @@ namespace MagicalCryptoWallet.Tests
});
nodes.ConnectedNodes.Add(Fixture.BackendRegTestNode.CreateNodeClient());
- // 2. Create block downloader service.
- var blocksFolderPath = Path.Combine(SharedFixture.DataDir, nameof(WalletTestsAsync), $"Blocks");
- BlockDownloader blockDownloader = new BlockDownloader(network, nodes, blocksFolderPath);
-
- // 3. Create mempool service.
+ // 2. Create mempool service.
var memPoolService = new MemPoolService();
memPoolService.TransactionReceived += WalletTestsAsync_MemPoolService_TransactionReceived;
Node node = Fixture.BackendRegTestNode.CreateNodeClient();
node.Behaviors.Add(new MemPoolBehavior(memPoolService));
- // 4. Create index downloader service.
+ // 3. Create index downloader service.
var indexFilePath = Path.Combine(SharedFixture.DataDir, nameof(WalletTestsAsync), $"Index{Global.RpcClient.Network}.dat");
var indexDownloader = new IndexDownloader(Global.RpcClient.Network, indexFilePath, new Uri(Fixture.BackendEndPoint));
- // 5. Create key manager service.
+ // 4. Create key manager service.
var keyManager = KeyManager.CreateNew(out Mnemonic mnemonic, "password");
- // 6. Create wallet service.
- var wallet = new WalletService(keyManager, blockDownloader, indexDownloader);
+ // 5. Create wallet service.
+ var blocksFolderPath = Path.Combine(SharedFixture.DataDir, nameof(WalletTestsAsync), $"Blocks");
+ var wallet = new WalletService(keyManager, indexDownloader, nodes, blocksFolderPath);
// Get some money, make it confirm.
var key = keyManager.GenerateNewKey("", KeyState.Clean, isInternal: false);
@@ -522,7 +452,6 @@ namespace MagicalCryptoWallet.Tests
try
{
nodes.Connect(); // Start connection service.
- blockDownloader.Synchronize(); // Start block downloader service.
node.VersionHandshake(); // Start mempool service.
indexDownloader.Synchronize(requestInterval: TimeSpan.FromSeconds(3)); // Start index downloader service.
@@ -538,9 +467,11 @@ namespace MagicalCryptoWallet.Tests
times++;
}
- await wallet.InitializeAsync(); // Initialize wallet service.
- Assert.Equal(1, blockDownloader.CountBlocks());
- wallet.Synchronize(); // Start wallet service.
+ using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)))
+ {
+ await wallet.InitializeAsync(cts.Token); // Initialize wallet service.
+ }
+ Assert.Equal(1, await wallet.CountBlocksAsync());
// Get some money, make it confirm.
key = keyManager.GenerateNewKey("", KeyState.Clean, isInternal: false);
@@ -557,14 +488,11 @@ namespace MagicalCryptoWallet.Tests
await Task.Delay(100);
times++;
}
- Assert.Equal(2, blockDownloader.CountBlocks());
+ Assert.Equal(2, await wallet.CountBlocksAsync());
}
finally
{
- if(wallet != null)
- {
- await wallet.StopAsync();
- }
+ wallet?.Dispose();
// Dispose index downloader service.
if (indexDownloader != null)
{
@@ -574,16 +502,6 @@ namespace MagicalCryptoWallet.Tests
// Dispose mempool service.
memPoolService.TransactionReceived -= WalletTestsAsync_MemPoolService_TransactionReceived;
- // Dispose downloader service
- if(blockDownloader != null)
- {
- await blockDownloader.StopAsync();
- }
- if (Directory.Exists(blocksFolderPath))
- {
- Directory.Delete(blocksFolderPath, recursive: true);
- }
-
// Dispose connection service.
nodes?.Dispose();
}

MODIFY
MagicalCryptoWallet/Services/IndexDownloader.cs
MagicalCryptoWallet/Services/IndexDownloader.cs
@@ -76,7 +76,10 @@ namespace MagicalCryptoWallet.Services
_running = 0;
var indexDir = Path.GetDirectoryName(IndexFilePath);
+ if (!string.IsNullOrEmpty(indexDir))
+ {
Directory.CreateDirectory(indexDir);
+ }
if (File.Exists(IndexFilePath))
{
if (Network == Network.RegTest)

MODIFY
MagicalCryptoWallet/Services/WalletService.cs
MagicalCryptoWallet/Services/WalletService.cs
@@ -17,17 +17,17 @@ using Nito.AsyncEx;
namespace MagicalCryptoWallet.Services
{
- public class WalletService
+ public class WalletService : IDisposable
{
public KeyManager KeyManager { get; }
- public BlockDownloader BlockDownloader { get; }
public IndexDownloader IndexDownloader { get; }
public NodesGroup Nodes { get; }
public string BlocksFolderPath { get; }
- private AsyncLock BlocksFolderLock { get; }
private AsyncLock HandleFiltersLock { get; }
+ private AsyncLock BlockDownloadLock { get; }
+ private AsyncLock BlockFolderLock { get; }
public SortedDictionary<Height, uint256> WalletBlocks { get; }
private HashSet<uint256> ProcessedBlocks { get; }
@@ -56,16 +56,17 @@ namespace MagicalCryptoWallet.Services
KnownCoins = new ConcurrentHashSet<SmartCoin>();
BlocksFolderPath = Guard.NotNullOrEmptyOrWhitespace(nameof(blocksFolderPath), blocksFolderPath, trim: true);
- BlocksFolderLock = new AsyncLock();
+ BlockFolderLock = new AsyncLock();
+ BlockDownloadLock = new AsyncLock();
_running = 0;
if (Directory.Exists(BlocksFolderPath))
{
- foreach (var blockFilePath in Directory.EnumerateFiles(BlocksFolderPath))
+ if(IndexDownloader.Network == Network.RegTest)
{
- var blockBytes = File.ReadAllBytes(blockFilePath);
- var block = new Block(blockBytes);
+ Directory.Delete(BlocksFolderPath, true);
+ Directory.CreateDirectory(BlocksFolderPath);
}
}
else
@@ -73,12 +74,11 @@ namespace MagicalCryptoWallet.Services
Directory.CreateDirectory(BlocksFolderPath);
}
- IndexDownloader.NewFilter += IndexDownloader_NewFilter;
- IndexDownloader.Reorged += IndexDownloader_Reorged;
- BlockDownloader.NewBlock += BlockDownloader_NewBlock;
+ IndexDownloader.NewFilter += IndexDownloader_NewFilterAsync;
+ IndexDownloader.Reorged += IndexDownloader_ReorgedAsync;
}
- private void BlockDownloader_NewBlock(object sender, Block block)
+ private void ProcessBlock(Block block)
{
using (WalletBlocksLock.Lock())
{
@@ -101,13 +101,13 @@ namespace MagicalCryptoWallet.Services
}
}
- private void IndexDownloader_Reorged(object sender, uint256 invalidBlockHash)
+ private async void IndexDownloader_ReorgedAsync(object sender, uint256 invalidBlockHash)
{
using (HandleFiltersLock.Lock())
using (WalletBlocksLock.Lock())
{
var elem = WalletBlocks.SingleOrDefault(x => x.Value == invalidBlockHash);
- BlockDownloader.TryRemove(invalidBlockHash);
+ await DeleteBlockAsync(invalidBlockHash);
WalletBlocks.RemoveByValue(invalidBlockHash);
ProcessedBlocks.Remove(invalidBlockHash);
if (elem.Key != null)
@@ -120,29 +120,20 @@ namespace MagicalCryptoWallet.Services
}
}
- private void IndexDownloader_NewFilter(object sender, FilterModel filterModel)
+ private async void IndexDownloader_NewFilterAsync(object sender, FilterModel filterModel)
{
using (HandleFiltersLock.Lock())
using (WalletBlocksLock.Lock())
{
if (filterModel.Filter != null && !WalletBlocks.ContainsValue(filterModel.BlockHash))
{
- var matchFound = filterModel.Filter.MatchAny(KeyManager.GetKeys().Select(x => x.GetP2wpkhScript().ToCompressedBytes()), filterModel.FilterKey);
- if (matchFound)
- {
- BlockDownloader.QueueToDownload(filterModel.BlockHash);
- WalletBlocks.AddOrReplace(filterModel.BlockHeight, filterModel.BlockHash);
- }
+ await ProcessFilterModelAsync(filterModel, CancellationToken.None);
}
}
}
- public async Task InitializeAsync()
- {
- if (!BlockDownloader.IsRunning)
+ public async Task InitializeAsync(CancellationToken cancel)
{
- throw new NotSupportedException($"{nameof(BlockDownloader)} is not running.");
- }
if (!IndexDownloader.IsRunning)
{
throw new NotSupportedException($"{nameof(IndexDownloader)} is not running.");
@@ -156,22 +147,17 @@ namespace MagicalCryptoWallet.Services
foreach (var filterModel in filters.Where(x => x.Filter != null && !WalletBlocks.ContainsValue(x.BlockHash))) // Filter can be null if there is no bech32 tx.
{
- var matchFound = filterModel.Filter.MatchAny(KeyManager.GetKeys().Select(x => x.GetP2wpkhScript().ToCompressedBytes()), filterModel.FilterKey);
- if (matchFound)
- {
- BlockDownloader.QueueToDownload(filterModel.BlockHash);
- WalletBlocks.AddOrReplace(filterModel.BlockHeight, filterModel.BlockHash);
+ await ProcessFilterModelAsync(filterModel, cancel);
+ }
}
}
- foreach (var relevantBlock in WalletBlocks)
+ private async Task ProcessFilterModelAsync(FilterModel filterModel, CancellationToken cancel)
{
- Block block = null;
- while ((block = BlockDownloader.GetBlock(relevantBlock.Value)) == null) // Wait until not downloaded.
+ var matchFound = filterModel.Filter.MatchAny(KeyManager.GetKeys().Select(x => x.GetP2wpkhScript().ToCompressedBytes()), filterModel.FilterKey);
+ if (matchFound)
{
- await Task.Delay(100);
- }
-
+ Block block = await GetOrDownloadBlockAsync(filterModel.BlockHash, cancel); // Wait until not downloaded.
var keys = KeyManager.GetKeys().ToList();
foreach (var tx in block.Transactions)
@@ -184,7 +170,7 @@ namespace MagicalCryptoWallet.Services
if (foundKey != default)
{
foundKey.KeyState = KeyState.Used;
- var coin = new SmartCoin(tx.GetHash(), i, output.ScriptPubKey, output.Value, tx.Inputs.ToTxoRefs().ToArray(), relevantBlock.Key, foundKey.Label, null);
+ var coin = new SmartCoin(tx.GetHash(), i, output.ScriptPubKey, output.Value, tx.Inputs.ToTxoRefs().ToArray(), filterModel.BlockHeight, foundKey.Label, null);
// Make sure there's always 21 clean keys generated and indexed.
while (KeyManager.GetKeys(KeyState.Clean, foundKey.IsInternal()).Count() < 21)
@@ -208,36 +194,99 @@ namespace MagicalCryptoWallet.Services
}
ProcessedBlocks.Add(block.GetHash());
- }
+
+ WalletBlocks.AddOrReplace(filterModel.BlockHeight, filterModel.BlockHash);
}
}
- public void Synchronize()
+ /// <exception cref="OperationCanceledException"></exception>
+ public async Task<Block> GetOrDownloadBlockAsync(uint256 hash, CancellationToken cancel)
+ {
+ // Try get the block
+ using (await BlockFolderLock.LockAsync())
+ {
+ foreach (var filePath in Directory.EnumerateFiles(BlocksFolderPath))
+ {
+ var fileName = Path.GetFileName(filePath);
+ if (hash == new uint256(fileName))
{
- Interlocked.Exchange(ref _running, 1);
+ var blockBytes = await File.ReadAllBytesAsync(filePath);
+ return new Block(blockBytes);
+ }
+ }
+ }
+ cancel.ThrowIfCancellationRequested();
- Task.Run(async () =>
+ // Download the block
+ Block block = null;
+ using (await BlockDownloadLock.LockAsync())
+ {
+ while(true)
{
+ cancel.ThrowIfCancellationRequested();
try
{
- while (IsRunning)
+ // If no connection, wait then continue.
+ while (Nodes.ConnectedNodes.Count == 0)
{
+ await Task.Delay(100);
+ }
+
+ Node node = Nodes.ConnectedNodes.RandomElement();
+ if (node == default(Node))
+ {
+ await Task.Delay(100);
+ continue;
+ }
+
+ if (!node.IsConnected && !(IndexDownloader.Network != Network.RegTest))
+ {
+ await Task.Delay(100);
+ continue;
+ }
+
try
{
- if (!BlockDownloader.IsRunning)
+ using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(32))) // ADSL 512 kbit/s 00:00:32
+ {
+ block = node.GetBlocks(new uint256[] { hash }, cts.Token)?.Single();
+ }
+
+ if (block == null)
{
- Logger.LogError<WalletService>($"{nameof(BlockDownloader)} is not running.");
- await Task.Delay(1000);
+ Logger.LogInfo<WalletService>($"Disconnected node, because couldn't parse received block.");
+ node.DisconnectAsync("Couldn't parse block.");
continue;
}
- if (!IndexDownloader.IsRunning)
+
+ if (!block.Check())
+ {
+ Logger.LogInfo<WalletService>($"Disconnected node, because block invalid block received.");
+ node.DisconnectAsync("Invalid block received.");
+ continue;
+ }
+ }
+ catch (TimeoutException)
+ {
+ Logger.LogInfo<WalletService>($"Disconnected node, because block download took too long.");
+ node.DisconnectAsync("Block download took too long.");
+ continue;
+ }
+ catch (OperationCanceledException)
{
- Logger.LogError<WalletService>($"{nameof(IndexDownloader)} is not running.");
- await Task.Delay(1000);
+ Logger.LogInfo<WalletService>($"Disconnected node, because block download took too long.");
+ node.DisconnectAsync("Block download took too long.");
+ continue;
+ }
+ catch (Exception ex)
+ {
+ Logger.LogDebug<WalletService>(ex);
+ Logger.LogInfo<WalletService>($"Disconnected node, because block download failed: {ex.Message}");
+ node.DisconnectAsync("Block download failed.");
continue;
}
- await Task.Delay(1000); // dummmy wait for now (TODO)
+ break; // If got this far break, then we have the block, it's valid. Break.
}
catch (Exception ex)
{
@@ -245,30 +294,68 @@ namespace MagicalCryptoWallet.Services
}
}
}
- finally
+ // Save the block
+ using (await BlockFolderLock.LockAsync())
{
- if (IsStopping)
- {
- Interlocked.Exchange(ref _running, 3);
+ var path = Path.Combine(BlocksFolderPath, hash.ToString());
+ await File.WriteAllBytesAsync(path, block.ToBytes());
+ }
+
+ return block;
}
+
+ /// <remarks>
+ /// Use it at reorgs.
+ /// </remarks>
+ public async Task DeleteBlockAsync(uint256 hash)
+ {
+ using (await BlockFolderLock.LockAsync())
+ {
+ var filePaths = Directory.EnumerateFiles(BlocksFolderPath);
+ var fileNames = filePaths.Select(x => Path.GetFileName(x));
+ var hashes = fileNames.Select(x => new uint256(x));
+
+ if (hashes.Contains(hash))
+ {
+ File.Delete(Path.Combine(BlocksFolderPath, hash.ToString()));
}
}
- );
}
- public async Task StopAsync()
+ public async Task<int> CountBlocksAsync()
{
- BlockDownloader.NewBlock -= BlockDownloader_NewBlock;
- IndexDownloader.NewFilter -= IndexDownloader_NewFilter;
- IndexDownloader.Reorged -= IndexDownloader_Reorged;
- if (IsRunning)
+ using (await BlockFolderLock.LockAsync())
{
- Interlocked.Exchange(ref _running, 2);
+ return Directory.EnumerateFiles(BlocksFolderPath).Count();
}
- while (IsStopping)
+ }
+
+ #region IDisposable Support
+
+ private volatile bool _disposedValue = false; // To detect redundant calls
+
+ protected virtual void Dispose(bool disposing)
+ {
+ if (!_disposedValue)
{
- await Task.Delay(50);
+ if (disposing)
+ {
+ IndexDownloader.NewFilter -= IndexDownloader_NewFilterAsync;
+ IndexDownloader.Reorged -= IndexDownloader_ReorgedAsync;
+ }
+
+ _disposedValue = true;
}
}
+
+ // This code added to correctly implement the disposable pattern.
+ public void Dispose()
+ {
+ // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
+ Dispose(true);
+ // GC.SuppressFinalize(this);
+ }
+
+ #endregion
}
}

Remove BlockDownloader Service, rather let the wallet service download and manage blocks