<commit_before>﻿using BitSharp.Common;
using BitSharp.Common.ExtensionMethods;
using BitSharp.Core.Domain;
using BitSharp.Core.Rules;
using BitSharp.Core.Storage;
using NLog;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Diagnostics;
using System.Linq;
using System.Threading;

namespace BitSharp.Core.Builders
{
    internal class ChainStateBuilder : IDisposable
    {
        private static readonly Logger logger = LogManager.GetCurrentClassLogger();

        private readonly IBlockchainRules rules;
        private readonly CoreStorage coreStorage;
        private readonly IStorageManager storageManager;

        private readonly BlockValidator blockValidator;

        private bool inTransaction;
        private readonly DisposeHandle<IChainStateCursor> chainStateCursorHandle;
        private readonly IChainStateCursor chainStateCursor;
        private ChainBuilder chain;
        private Chain rollbackChain;
        private readonly UtxoBuilder utxoBuilder;

        private readonly WorkerPool blockTxesDispatcher;
        private readonly ParallelConsumer<Tuple<UInt256, int, CompletionCount>> utxoReader;
        private readonly WorkerPool blockTxesSorter;

        private readonly ReaderWriterLockSlim commitLock;

        private readonly BuilderStats stats;

        public ChainStateBuilder(IBlockchainRules rules, CoreStorage coreStorage)
        {
            this.rules = rules;
            this.coreStorage = coreStorage;
            this.storageManager = coreStorage.StorageManager;

            this.stats = new BuilderStats();
            this.blockValidator = new BlockValidator(this.stats, this.coreStorage, this.rules);

            this.chainStateCursorHandle = this.storageManager.OpenChainStateCursor();
            this.chainStateCursor = this.chainStateCursorHandle.Item;

            this.chainStateCursor.BeginTransaction(readOnly: true);
            try
            {
                this.chain = new ChainBuilder(chainStateCursor.ReadChain());
            }
            finally
            {
                this.chainStateCursor.RollbackTransaction();
            }
            this.utxoBuilder = new UtxoBuilder();

            this.commitLock = new ReaderWriterLockSlim();

            this.blockTxesDispatcher = new WorkerPool("ChainStateBuilder.BlockTxesDispatcher", 1);
            this.utxoReader = new ParallelConsumer<Tuple<UInt256, int, CompletionCount>>("ChainStateBuilder.UtxoReaer", 4);
            this.blockTxesSorter = new WorkerPool("ChainStateBuilder.BlockTxesSorter", 1);
        }

        public void Dispose()
        {
            this.blockValidator.Dispose();
            this.chainStateCursorHandle.Dispose();
            this.stats.Dispose();
            this.commitLock.Dispose();
            this.blockTxesDispatcher.Dispose();
            this.utxoReader.Dispose();
            this.blockTxesSorter.Dispose();
        }

        public Chain Chain
        {
            get
            {
                return this.commitLock.DoRead(() =>
                    this.chain.ToImmutable());
            }
        }

        public BuilderStats Stats { get { return this.stats; } }

        public void AddBlock(ChainedHeader chainedHeader, IEnumerable<Transaction> transactions)
        {
            AddBlock(chainedHeader, transactions.Select((tx, txIndex) => new BlockTx(txIndex, depth: 0, hash: tx.Hash, pruned: false, transaction: tx)));
        }

        public void AddBlock(ChainedHeader chainedHeader, IEnumerable<BlockTx> blockTxes)
        {
            this.commitLock.DoWrite(() =>
            {
                this.BeginTransaction(readOnly: true);
                try
                {
                    this.blockValidator.ValidateTransactions(chainedHeader,
                        loadingTxes =>
                        {
                            using (var deferredChainStateCursor = new DeferredChainStateCursor(this.chainStateCursor))
                            using (var blockTxesReadQueue = new ConcurrentBlockingQueue<Tuple<UInt256, int, CompletionCount>>())
                            using (var blockTxesCalculateQueue = new ConcurrentBlockingQueue<BlockTx>())
                            using (var txLoadedEvent = new AutoResetEvent(false))
                            {
                                this.stats.calculateUtxoDurationMeasure.Measure(() =>
                                {
                                    var pendingSortedTxes = new ConcurrentQueue<Tuple<BlockTx, CompletionCount>>();
                                    var pendingSortedTxesCompletedAdding = false;

                                    using (var chainState = new ChainState(this.chain.ToImmutable(), this.storageManager))
                                    using (this.blockTxesDispatcher.Start(() =>
                                    {
                                        try
                                        {
                                            foreach (var blockTx in blockTxes)
                                            {
                                                var inputCount = !blockTx.IsCoinbase ? blockTx.Transaction.Inputs.Length : 0;
                                                var completionCount = new CompletionCount(inputCount + 1);
                                                pendingSortedTxes.Enqueue(Tuple.Create(blockTx, completionCount));

                                                if (!blockTx.IsCoinbase)
                                                {
                                                    blockTxesReadQueue.AddRange(blockTx.Transaction.Inputs.Select(
                                                       (txInput, inputIndex) => Tuple.Create(txInput.PreviousTxOutputKey.TxHash, inputIndex, completionCount)));
                                                }

                                                blockTxesReadQueue.Add(Tuple.Create(blockTx.Hash, inputCount, completionCount));
                                            }
                                        }
                                        finally
                                        {
                                            blockTxesReadQueue.CompleteAdding();
                                            pendingSortedTxesCompletedAdding = true;
                                            txLoadedEvent.Set();
                                        }
                                    }))
                                    using (this.utxoReader.Start(blockTxesReadQueue,
                                        tuple =>
                                        {
                                            var txHash = tuple.Item1;
                                            var inputIndex = tuple.Item2;
                                            var completionCount = tuple.Item3;

                                            UnspentTx unspentTx;
                                            deferredChainStateCursor.WarmUnspentTx(txHash, () =>
                                            {
                                                var result = Tuple.Create(chainState.TryGetUnspentTx(txHash, out unspentTx), unspentTx);
                                                this.stats.utxoReadRateMeasure.Tick();
                                                return result;
                                            });<commit_msg>use consumer for block txes dispatcher as well<commit_after>﻿using BitSharp.Common;
using BitSharp.Common.ExtensionMethods;
using BitSharp.Core.Domain;
using BitSharp.Core.Rules;
using BitSharp.Core.Storage;
using NLog;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Diagnostics;
using System.Linq;
using System.Threading;

namespace BitSharp.Core.Builders
{
    internal class ChainStateBuilder : IDisposable
    {
        private static readonly Logger logger = LogManager.GetCurrentClassLogger();

        private readonly IBlockchainRules rules;
        private readonly CoreStorage coreStorage;
        private readonly IStorageManager storageManager;

        private readonly BlockValidator blockValidator;

        private bool inTransaction;
        private readonly DisposeHandle<IChainStateCursor> chainStateCursorHandle;
        private readonly IChainStateCursor chainStateCursor;
        private ChainBuilder chain;
        private Chain rollbackChain;
        private readonly UtxoBuilder utxoBuilder;

        private readonly ParallelConsumer<BlockTx> blockTxesDispatcher;
        private readonly ParallelConsumer<Tuple<UInt256, int, CompletionCount>> utxoReader;
        private readonly WorkerPool blockTxesSorter;

        private readonly ReaderWriterLockSlim commitLock;

        private readonly BuilderStats stats;

        public ChainStateBuilder(IBlockchainRules rules, CoreStorage coreStorage)
        {
            this.rules = rules;
            this.coreStorage = coreStorage;
            this.storageManager = coreStorage.StorageManager;

            this.stats = new BuilderStats();
            this.blockValidator = new BlockValidator(this.stats, this.coreStorage, this.rules);

            this.chainStateCursorHandle = this.storageManager.OpenChainStateCursor();
            this.chainStateCursor = this.chainStateCursorHandle.Item;

            this.chainStateCursor.BeginTransaction(readOnly: true);
            try
            {
                this.chain = new ChainBuilder(chainStateCursor.ReadChain());
            }
            finally
            {
                this.chainStateCursor.RollbackTransaction();
            }
            this.utxoBuilder = new UtxoBuilder();

            this.commitLock = new ReaderWriterLockSlim();

            this.blockTxesDispatcher = new ParallelConsumer<BlockTx>("ChainStateBuilder.BlockTxesDispatcher", 1);
            this.utxoReader = new ParallelConsumer<Tuple<UInt256, int, CompletionCount>>("ChainStateBuilder.UtxoReaer", 4);
            this.blockTxesSorter = new WorkerPool("ChainStateBuilder.BlockTxesSorter", 1);
        }

        public void Dispose()
        {
            this.blockValidator.Dispose();
            this.chainStateCursorHandle.Dispose();
            this.stats.Dispose();
            this.commitLock.Dispose();
            this.blockTxesDispatcher.Dispose();
            this.utxoReader.Dispose();
            this.blockTxesSorter.Dispose();
        }

        public Chain Chain
        {
            get
            {
                return this.commitLock.DoRead(() =>
                    this.chain.ToImmutable());
            }
        }

        public BuilderStats Stats { get { return this.stats; } }

        public void AddBlock(ChainedHeader chainedHeader, IEnumerable<Transaction> transactions)
        {
            AddBlock(chainedHeader, transactions.Select((tx, txIndex) => new BlockTx(txIndex, depth: 0, hash: tx.Hash, pruned: false, transaction: tx)));
        }

        public void AddBlock(ChainedHeader chainedHeader, IEnumerable<BlockTx> blockTxes)
        {
            this.commitLock.DoWrite(() =>
            {
                this.BeginTransaction(readOnly: true);
                try
                {
                    this.blockValidator.ValidateTransactions(chainedHeader,
                        loadingTxes =>
                        {
                            using (var deferredChainStateCursor = new DeferredChainStateCursor(this.chainStateCursor))
                            using (var blockTxesReadQueue = new ConcurrentBlockingQueue<Tuple<UInt256, int, CompletionCount>>())
                            using (var blockTxesCalculateQueue = new ConcurrentBlockingQueue<BlockTx>())
                            using (var txLoadedEvent = new AutoResetEvent(false))
                            {
                                this.stats.calculateUtxoDurationMeasure.Measure(() =>
                                {
                                    var pendingSortedTxes = new ConcurrentQueue<Tuple<BlockTx, CompletionCount>>();
                                    var pendingSortedTxesCompletedAdding = false;

                                    using (var chainState = new ChainState(this.chain.ToImmutable(), this.storageManager))
                                    using (this.blockTxesDispatcher.Start(blockTxes,
                                        blockTx =>
                                        {
                                            var inputCount = !blockTx.IsCoinbase ? blockTx.Transaction.Inputs.Length : 0;
                                            var completionCount = new CompletionCount(inputCount + 1);
                                            pendingSortedTxes.Enqueue(Tuple.Create(blockTx, completionCount));

                                            if (!blockTx.IsCoinbase)
                                            {
                                                blockTxesReadQueue.AddRange(blockTx.Transaction.Inputs.Select(
                                                    (txInput, inputIndex) => Tuple.Create(txInput.PreviousTxOutputKey.TxHash, inputIndex, completionCount)));
                                            }

                                            blockTxesReadQueue.Add(Tuple.Create(blockTx.Hash, inputCount, completionCount));
                                        },
                                        _ =>
                                        {
                                            blockTxesReadQueue.CompleteAdding();
                                            pendingSortedTxesCompletedAdding = true;
                                            txLoadedEvent.Set();
                                        }))
                                    using (this.utxoReader.Start(blockTxesReadQueue,
                                        tuple =>
                                        {
                                            var txHash = tuple.Item1;
                                            var inputIndex = tuple.Item2;
                                            var completionCount = tuple.Item3;

                                            UnspentTx unspentTx;
                                            deferredChainStateCursor.WarmUnspentTx(txHash, () =>
                                            {
                                                var result = Tuple.Create(chainState.TryGetUnspentTx(txHash, out unspentTx), unspentTx);
                                                this.stats.utxoReadRateMeasure.Tick();
                                                return result;
                                            });
{'fraction_non_alphanumeric': 0.06631998379144999, 'fraction_numerical': 0.0024312825015195515, 'mean_word_length': 1.3463793376643955, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 48, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 122, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1448531', 'n_tokens_mistral': 3112, 'n_tokens_neox': 2979, 'n_words': 599}