diff --git a/EBA/Blockchains/Bitcoin/BitcoinOrchestrator.cs b/EBA/Blockchains/Bitcoin/BitcoinOrchestrator.cs index e71c5209..099abf1c 100644 --- a/EBA/Blockchains/Bitcoin/BitcoinOrchestrator.cs +++ b/EBA/Blockchains/Bitcoin/BitcoinOrchestrator.cs @@ -215,7 +215,7 @@ void RegisterFailed(long h) // TODO: refactor the following so that only options is passed to the buffer using var gBuffer = new PersistentGraphBuffer( - graphAgent: _host.Services.GetRequiredService(), + graphOrchestrator: _host.Services.GetRequiredService(), logger: _host.Services.GetRequiredService>(), pTxoLifeCyccleLogger: options.Bitcoin.Traverse.TrackTxo ?_host.Services.GetRequiredService>() : null, txoLifeCycleFilename: options.Bitcoin.Traverse.TrackTxo ? options.Bitcoin.Traverse.TxoFilename : null, diff --git a/EBA/Blockchains/Bitcoin/Utilities/TraverseFinalizer.cs b/EBA/Blockchains/Bitcoin/Utilities/TraverseFinalizer.cs index e9c89314..124755d5 100644 --- a/EBA/Blockchains/Bitcoin/Utilities/TraverseFinalizer.cs +++ b/EBA/Blockchains/Bitcoin/Utilities/TraverseFinalizer.cs @@ -20,16 +20,16 @@ public async Task UpdatePostTraverse(CancellationToken ct) var batches = await Batch.DeserializeBatchesAsync(_options.Bitcoin.MapSpends.BatchesFilename); _logger.LogInformation("Deserialized {n} batches from {filename}.", batches.Count, _options.Bitcoin.MapSpends.BatchesFilename); - _processStep = "[1/4] Block-to-batch mapping: "; + _processStep = "[1/4] Block-to-batch mapping:"; GetBlockHeightToBatchMapping(_options, batches, out var blockToBatch, out var blockNodes); - _processStep = "[2/4] Collecting Txo spending: "; + _processStep = "[2/4] Collecting Txo spending:"; await CreatePerBatchSpentUtxo(batches, blockToBatch, ct); - _processStep = "[3/4] Setting UTxO spending: "; + _processStep = "[3/4] Setting UTxO spending:"; await SetTxoSpentHeight(batches, ct); - _processStep = "[4/4] Setting supply amount: "; + _processStep = "[4/4] Setting supply amount:"; await SetSupplyAmount(batches, blockNodes, ct); } @@ -211,10 +211,10 @@ private async Task SetSupplyAmount(List batches, SortedDictionary x.BlockMetadata.TotalSupply); var totalSupplyNominalIndex = BlockNodeDescriptor.StaticMapper.GetPropertyCsvIndex(x => x.BlockMetadata.TotalSupplyNominal); + _logger.LogInformation("{s} Updating block node files with total supply information.", _processStep); + int counter = 0; foreach (var batch in batches) { - _logger.LogInformation("{s} Updating block node files with total supply information...", _processStep); - var blockNodesFilename = batch.GetFilename(BlockNode.Kind); using var writer = new StreamWriter(new GZipStream( File.Create(Helpers.AddPostfixToFilename(blockNodesFilename, "_supply_updated")), @@ -235,9 +235,17 @@ private async Task SetSupplyAmount(List batches, SortedDictionary, IDisposable +public class BitcoinGraphOrchestrator : IGraphOrchestrator, IDisposable { private readonly Options _options; private readonly IGraphDb _db; - private readonly ILogger _logger; + private readonly ILogger _logger; private bool _hasSerializedConstants = false; private bool _disposed = false; - public BitcoinGraphAgent( + public BitcoinGraphOrchestrator( Options options, IGraphDb graphDb, - ILogger logger) + ILogger logger) { _options = options; _db = graphDb; diff --git a/EBA/Graph/Bitcoin/Descriptors/BlockNodeDescriptor.cs b/EBA/Graph/Bitcoin/Descriptors/BlockNodeDescriptor.cs index 2a033af8..de471071 100644 --- a/EBA/Graph/Bitcoin/Descriptors/BlockNodeDescriptor.cs +++ b/EBA/Graph/Bitcoin/Descriptors/BlockNodeDescriptor.cs @@ -10,8 +10,12 @@ public class BlockNodeDescriptor : IElementDescriptor public static ElementMapper StaticMapper => _mapper; private static readonly ElementMapper _mapper = new( new MappingBuilder() - .MapSourceId(_idSpace, n => n.BlockMetadata.Height) + // using empty string as propName in header override since height property is already defined. + // Height is defined twice because id is treated string, but we want a long height as property, + // so only option currently is double-prop definition. + .Map(n => n.BlockMetadata.Height).WithCsvHeader(p => p.GetIdFieldCsvHeader("", _idSpace)) .Map(n => n.BlockMetadata.Height) + .Map(n => n.BlockMetadata.Hash) .Map(n => n.BlockMetadata.Confirmations) .Map(n => n.BlockMetadata.Version) @@ -68,12 +72,6 @@ public class BlockNodeDescriptor : IElementDescriptor .MapRange(PropertyMappingFactory.ToMappings( nameof(BlockNode.TripletTypeValueSum), n => n.TripletTypeValueSum)) - .MapRange(PropertyMappingFactory.ToMappings( - nameof(BlockNode.TripletTypeCount), (BlockNode n) => n.TripletTypeCount)) - - .MapRange(PropertyMappingFactory.ToMappings( - nameof(BlockNode.TripletTypeValueSum), (BlockNode n) => n.TripletTypeValueSum)) - .Map(n => n.BlockMetadata.TotalSupply) .Map(n => n.BlockMetadata.TotalSupplyNominal) diff --git a/EBA/Graph/Bitcoin/Descriptors/C2TEdgeDescriptor.cs b/EBA/Graph/Bitcoin/Descriptors/C2TEdgeDescriptor.cs index 4712803c..dd162ef1 100644 --- a/EBA/Graph/Bitcoin/Descriptors/C2TEdgeDescriptor.cs +++ b/EBA/Graph/Bitcoin/Descriptors/C2TEdgeDescriptor.cs @@ -9,7 +9,7 @@ public class C2TEdgeDescriptor : IElementDescriptor public static ElementMapper StaticMapper => _mapper; private static readonly ElementMapper _mapper = new( new MappingBuilder() - .MapSourceId(_idSpace, _ => CoinbaseNode.Kind) + .MapSourceId(CoinbaseNode.Kind.ToString(), _ => CoinbaseNode.Kind) .MapTargetId(TxNodeDescriptor.IdSpace, e => e.Target.Txid) .Map(e => e.Value) .Map(e => e.BlockHeight) diff --git a/EBA/Graph/Bitcoin/OffChain/EconomicAugmentor.cs b/EBA/Graph/Bitcoin/OffChain/EconomicAugmentor.cs index 25bda823..72838a1d 100644 --- a/EBA/Graph/Bitcoin/OffChain/EconomicAugmentor.cs +++ b/EBA/Graph/Bitcoin/OffChain/EconomicAugmentor.cs @@ -2,11 +2,11 @@ namespace EBA.Graph.Bitcoin.OffChain; -public class EconomicAugmentor(Options options, IGraphDb graphDb, ILogger logger) +public class EconomicAugmentor(Options options, IGraphDb graphDb, ILogger logger) { private readonly Options _options = options; private readonly IGraphDb _graphDb = graphDb; - private readonly ILogger _logger = logger; + private readonly ILogger _logger = logger; public async Task SetBlockMarketIndicators(CancellationToken ct) { diff --git a/EBA/Graph/Bitcoin/TraversalAlgorithms/ForestFire.cs b/EBA/Graph/Bitcoin/TraversalAlgorithms/ForestFire.cs index 8f905d80..bf14707f 100644 --- a/EBA/Graph/Bitcoin/TraversalAlgorithms/ForestFire.cs +++ b/EBA/Graph/Bitcoin/TraversalAlgorithms/ForestFire.cs @@ -4,11 +4,11 @@ public class ForestFire : ITraversalAlgorithm { private readonly Options _options; private readonly IGraphDb _graphDb; - private readonly ILogger _logger; + private readonly ILogger _logger; private int _maxHopReached = 0; - public ForestFire(Options options, IGraphDb graphDb, ILogger logger) + public ForestFire(Options options, IGraphDb graphDb, ILogger logger) { _options = options; _graphDb = graphDb; diff --git a/EBA/Graph/IGraphAgent.cs b/EBA/Graph/IGraphOrchestrator.cs similarity index 70% rename from EBA/Graph/IGraphAgent.cs rename to EBA/Graph/IGraphOrchestrator.cs index 6ae41399..1470dba4 100644 --- a/EBA/Graph/IGraphAgent.cs +++ b/EBA/Graph/IGraphOrchestrator.cs @@ -1,6 +1,6 @@ namespace EBA.Graph; -public interface IGraphAgent where T : GraphBase +public interface IGraphOrchestrator where T : GraphBase { public Task SampleAsync(CancellationToken ct); public Task SerializeAsync(T g, CancellationToken ct); diff --git a/EBA/Graph/Model/Property.cs b/EBA/Graph/Model/Property.cs index 6fa20a47..bc93ff72 100644 --- a/EBA/Graph/Model/Property.cs +++ b/EBA/Graph/Model/Property.cs @@ -50,6 +50,11 @@ public string GetIdFieldCsvHeader(string idGroup) return $"{Name}:ID({idGroup})"; } + public string GetIdFieldCsvHeader(string propName, string idGroup) + { + return $"{propName}:ID({idGroup})"; + } + public string GetSetter() { return $"{Name}:{GetReader()}"; diff --git a/EBA/Infrastructure/StartupSolutions/Startup.cs b/EBA/Infrastructure/StartupSolutions/Startup.cs index be877643..32e56039 100644 --- a/EBA/Infrastructure/StartupSolutions/Startup.cs +++ b/EBA/Infrastructure/StartupSolutions/Startup.cs @@ -72,7 +72,7 @@ private static void ConfigureServices(IServiceCollection services, Options optio services.AddSingleton(options); services.AddSingleton, BitcoinNeo4jDb>(); services.AddSingleton(); - services.AddSingleton(); + services.AddSingleton(); // Passing BitcoinAgent type as the generic argument // to AddHttpClient will cause registering it diff --git a/EBA/Orchestrator.cs b/EBA/Orchestrator.cs index 44d111ba..79bc0c49 100644 --- a/EBA/Orchestrator.cs +++ b/EBA/Orchestrator.cs @@ -100,8 +100,8 @@ private async Task BitcoinAugmentGraphAsync(Options options) { var host = await SetupAndGetHostAsync(options); await JsonSerializer.SerializeAsync(options, options.StatusFile, _cT); - var bitcoinGraphAgent = host.Services.GetRequiredService(); - await bitcoinGraphAgent.AddMarketData(_cT); + var bitcoinGraphOrchestrator = host.Services.GetRequiredService(); + await bitcoinGraphOrchestrator.AddMarketData(_cT); } private async Task BitcoinSampleGraphAsync(Options options) @@ -109,8 +109,8 @@ private async Task BitcoinSampleGraphAsync(Options options) var host = await SetupAndGetHostAsync(options); await JsonSerializer.SerializeAsync(options, options.StatusFile, _cT); - var bitcoinGraphAgent = host.Services.GetRequiredService(); - await bitcoinGraphAgent.SampleAsync(_cT); + var bitcoinGraphOrchestrator = host.Services.GetRequiredService(); + await bitcoinGraphOrchestrator.SampleAsync(_cT); } private async Task BitcoinAddressStatsAsync(Options options) diff --git a/EBA/PersistentObject/PersistentGraphBuffer.cs b/EBA/PersistentObject/PersistentGraphBuffer.cs index e6b4f43e..40c8bc0e 100644 --- a/EBA/PersistentObject/PersistentGraphBuffer.cs +++ b/EBA/PersistentObject/PersistentGraphBuffer.cs @@ -4,7 +4,7 @@ namespace EBA.PersistentObject; public class PersistentGraphBuffer : PersistentObjectBase, IDisposable, IAsyncDisposable { - private readonly Graph.Bitcoin.BitcoinGraphAgent? _graphAgent; + private readonly Graph.Bitcoin.BitcoinGraphOrchestrator? _graphOrchestrator; private readonly ILogger _logger; private readonly PersistentTxoLifeCycleBuffer? _pTxoLifeCycleBuffer = null; private readonly SemaphoreSlim _semaphore; @@ -20,7 +20,7 @@ public ReadOnlyCollection BlocksHeightInBuffer private readonly ConcurrentDictionary _blocksHeightsInBuffer = new(); public PersistentGraphBuffer( - Graph.Bitcoin.BitcoinGraphAgent? graphAgent, + Graph.Bitcoin.BitcoinGraphOrchestrator? graphOrchestrator, ILogger logger, ILogger? pTxoLifeCyccleLogger, string? txoLifeCycleFilename, @@ -30,7 +30,7 @@ public PersistentGraphBuffer( CancellationToken ct) : base(logger, ct) { - _graphAgent = graphAgent; + _graphOrchestrator = graphOrchestrator; _logger = logger; if (txoLifeCycleFilename != null && pTxoLifeCyccleLogger != null) @@ -59,8 +59,8 @@ public override async Task SerializeAsync( // on cancellation and recovery, but that can add additional complexities. var tasks = new List { }; - if (_graphAgent != null) - tasks.Add(_graphAgent.SerializeAsync(obj, default)); + if (_graphOrchestrator != null) + tasks.Add(_graphOrchestrator.SerializeAsync(obj, default)); if (_pTxoLifeCycleBuffer != null) tasks.Add(_pTxoLifeCycleBuffer.SerializeAsync(obj.Block.TxoLifecycle.Values, default)); @@ -107,7 +107,7 @@ public async Task WaitForBufferToEmptyAsync() if (disposing) { - _graphAgent?.Dispose(); + _graphOrchestrator?.Dispose(); } _disposed = true; @@ -120,7 +120,7 @@ public async Task WaitForBufferToEmptyAsync() { base.Dispose(true); - if (_graphAgent is IAsyncDisposable asyncDisposable) + if (_graphOrchestrator is IAsyncDisposable asyncDisposable) await asyncDisposable.DisposeAsync(); _disposed = true; diff --git a/website/docs/bitcoin/etl/s1-sync-bitcoin.mdx b/website/docs/bitcoin/etl/s1-sync-bitcoin.mdx index 772d6614..0d2deff3 100644 --- a/website/docs/bitcoin/etl/s1-sync-bitcoin.mdx +++ b/website/docs/bitcoin/etl/s1-sync-bitcoin.mdx @@ -72,8 +72,3 @@ EBA does **NOT** try to access Wallets defined in `Bitcoin Core`. Regardless, there are some known privacy concerns when running `Bitcoin Core` with a defined wallet while the REST API is enabled. See [Bitcoin docs](https://github.com/bitcoin/bitcoin/blob/master/doc/REST-interface.md#risks). - - - - - diff --git a/website/docs/bitcoin/etl/s2-traverse-bitcoin.md b/website/docs/bitcoin/etl/s2-traverse-bitcoin.md index b96ec50a..68690def 100644 --- a/website/docs/bitcoin/etl/s2-traverse-bitcoin.md +++ b/website/docs/bitcoin/etl/s2-traverse-bitcoin.md @@ -33,33 +33,35 @@ For this task, you may take the following steps. .\eba.exe bitcoin traverse --help ``` +
+ Optimizations and Scaling Potential + + Traversing Bitcoin blocks can take a considerable amount of time. + To accelerate this, + EBA heavily leverages multi-threading, + and all time-consuming operations are implemented to be non-blocking. + It also minimizes the latency between submitting API calls and + processing the returned data, + which allows data to be handled in parallel threads, + so it doesn't wait to encode and persist a block's graph elements before processing the next block. + However, there is a limit to how many concurrent requests EBA and Bitcoin Core can process optimally. + Therefore, despite these optimizations, + if both applications are running on the same machine, + their performance is ultimately bound by its I/O limits, + primarily the random read/write performance of the storage. + + + Since EBA processes each block independently, + one potential improvement is to deploy the application on a Kubernetes (k8s) cluster + (requires dockerizing both EBA and Bitcoin Core). + In this setup, each instance of EBA service could process a subset of blocks + while a load balancer directs its API calls to replicas of the Bitcoin Core services. + This horizontal scaling would significantly improve performance; + however, because this requires a k8s cluster and cloud or on-premises HPC resources + that may not be [widely accessible](/docs/gs/accessibility), + the specifics of such a deployment are not currently covered. +
-### Performance and Scalability - -Traversing Bitcoin blocks can take a considerable amount of time. -To accelerate this, -EBA heavily leverages multi-threading, -and all time-consuming operations are implemented to be non-blocking. -It also minimizes the latency between submitting API calls and -processing the returned data, -which allows data to be handled in parallel threads, -so it doesn't wait to encode and persist a block's graph elements before processing the next block. -However, there is a limit to how many concurrent requests EBA and Bitcoin Core can process optimally. -Therefore, despite these optimizations, -if both applications are running on the same machine, -their performance is ultimately bound by its I/O limits, -primarily the random read/write performance of the storage. - - -Since EBA processes each block independently, -one potential improvement is to deploy the application on a Kubernetes (k8s) cluster -(requires dockerizing both EBA and Bitcoin Core). -In this setup, each instance of EBA service could process a subset of blocks -while a load balancer directs its API calls to replicas of the Bitcoin Core services. -This horizontal scaling would significantly improve performance; -however, because this requires a k8s cluster and cloud or on-premises HPC resources -that may not be [widely accessible](/docs/gs/accessibility), -the specifics of such a deployment are not currently covered. @@ -149,3 +151,11 @@ node files. ```shell .\eba.exe bitcoin dedup --sorted-script-nodes-file sorted_nodes_Script.csv --sorted-tx-nodes-file sorted_nodes_Tx.csv ``` + + +## Post-Traverse + + +```shell +.\eba.exe bitcoin post-traverse --batches-filename batches.json +``` \ No newline at end of file diff --git a/website/docs/bitcoin/etl/s5a-import.md b/website/docs/bitcoin/etl/s5a-import.md index abd6760a..cec35496 100644 --- a/website/docs/bitcoin/etl/s5a-import.md +++ b/website/docs/bitcoin/etl/s5a-import.md @@ -5,6 +5,8 @@ sidebar_position: 4 slug: /bitcoin/etl/import --- +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; :::tip Do I need to run the bulk import process? @@ -83,10 +85,27 @@ meaning it does not support incremental updates to an existing graph. 4. Set an environment variable pointing to the directory containing the Bitcoin graph TSV files. - + + ```shell export GDIR="" ``` + + + + ```bash + $env:GDIR = "" + $env:NDIR = "" + ``` + + Verify that your data directory contains the correctly batched files. You can use the following command to inspect the file distribution by type @@ -136,6 +155,15 @@ meaning it does not support incremental updates to an existing graph. 6. Execute the import command. Note that we use regex patterns (e.g., .`*BitcoinS2S.tsv.gz`) to ingest the batched edge files automatically. + + ```shell sudo -u neo4j HEAP_SIZE=4G neo4j-admin database import full \ --overwrite-destination neo4j \ @@ -153,6 +181,28 @@ meaning it does not support incremental updates to an existing graph. --array-delimiter=";" \ --verbose ``` + + + + ```bash + & "$env:NDIR\bin\neo4j-admin.bat" database import full ` + --overwrite-destination neo4j ` + --nodes="$env:GDIR\Coinbase.csv.gz" ` + --nodes="$env:GDIR\header_nodes_Block.csv.gz,$env:GDIR\[0-9].*_nodes_Block_supply_updated.csv.gz" ` + --nodes="$env:GDIR\header_nodes_Script.csv.gz,$env:GDIR\unique_nodes_Script.csv.gz" ` + --nodes="$env:GDIR\header_nodes_Tx.csv.gz,$env:GDIR\unique_nodes_Tx.csv.gz" ` + --relationships="$env:GDIR\header_edges_Block_Confirms_Tx.csv.gz,$env:GDIR\[0-9].*_edges_Block_Confirms_Tx.csv.gz" ` + --relationships="$env:GDIR\header_edges_Coinbase_Mints_Tx.csv.gz,$env:GDIR\[0-9].*_edges_Coinbase_Mints_Tx.csv.gz" ` + --relationships="$env:GDIR\header_edges_Tx_Credits_Script.csv.gz,$env:GDIR\[0-9].*_edges_Tx_Credits_Script_with_txo_spent_height_set.csv.gz" ` + --relationships="$env:GDIR\header_edges_Script_Redeems_Tx.csv.gz,$env:GDIR\[0-9].*_edges_Script_Redeems_Tx.csv.gz" ` + --relationships="$env:GDIR\header_edges_Tx_Fee_Tx.csv.gz,$env:GDIR\[0-9].*_edges_Tx_Fee_Tx.csv.gz" ` + --relationships="$env:GDIR\header_edges_Tx_Transfers_Tx.csv.gz,$env:GDIR\[0-9].*_edges_Tx_Transfers_Tx.csv.gz" ` + --delimiter="\t" ` + --array-delimiter=";" ` + --verbose + ``` + + 7. Once the import concludes, restart the Neo4j service.