<commit_before>﻿using BitSharp.Blockchain;
using BitSharp.Common;
using BitSharp.Common.ExtensionMethods;
using BitSharp.Data;
using BitSharp.Storage;
using Ninject;
using Ninject.Parameters;
using NLog;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace BitSharp.Daemon
{
    internal class ChainStateWorker : Worker
    {
        public event Action OnChainStateChanged;
        public event Action OnChainStateBuilderChanged;

        private static readonly TimeSpan MAX_BUILDER_LIFETIME = TimeSpan.FromMinutes(10);

        private readonly Logger logger;
        private Func<Chain> getTargetChain;
        private readonly IKernel kernel;
        private readonly IBlockchainRules rules;
        private readonly BlockchainCalculator calculator;
        private readonly TransactionCache transactionCache;
        private readonly BlockRollbackCache blockRollbackCache;
        private readonly InvalidBlockCache invalidBlockCache;

        private ChainState chainState;
        private readonly ReaderWriterLockSlim chainStateLock;

        private ChainStateBuilder chainStateBuilder;
        private DateTime chainStateBuilderTime;

        private readonly TimeSpan[] blockTimes;
        private int blockTimesIndex;

        private readonly PruningWorker pruningWorker;

        public ChainStateWorker(WorkerConfig workerConfig, Func<Chain> getTargetChain, Logger logger, IKernel kernel, IBlockchainRules rules, TransactionCache transactionCache, BlockRollbackCache blockRollbackCache, InvalidBlockCache invalidBlockCache)
            : base("ChainStateWorker", workerConfig.initialNotify, workerConfig.minIdleTime, workerConfig.maxIdleTime, logger)
        {
            this.logger = logger;
            this.getTargetChain = getTargetChain;
            this.kernel = kernel;
            this.rules = rules;
            this.transactionCache = transactionCache;
            this.blockRollbackCache = blockRollbackCache;
            this.invalidBlockCache = invalidBlockCache;

            this.calculator = kernel.Get<BlockchainCalculator>(new ConstructorArgument("shutdownToken", this.ShutdownToken.Token));

            this.chainState = ChainState.CreateForGenesisBlock(this.rules.GenesisChainedBlock);
            this.chainStateLock = new ReaderWriterLockSlim();

            this.blockTimes = new TimeSpan[10000];
            this.blockTimesIndex = -1;

            this.pruningWorker = kernel.Get<PruningWorker>(
                new ConstructorArgument("workerConfig", new WorkerConfig(initialNotify: false, minIdleTime: TimeSpan.FromSeconds(30), maxIdleTime: TimeSpan.FromMinutes(5))),
                new ConstructorArgument("getChainState", (Func<ChainState>)(() => this.chainState)),
                new ConstructorArgument("getChainStateBuilder", (Func<ChainStateBuilder>)(() => this.chainStateBuilder)));
        }

        public ChainState ChainState { get { return this.chainState; } }

        internal ChainStateBuilder ChainStateBuilder { get { return this.chainStateBuilder; } }

        public TimeSpan AverageBlockProcessingTime()
        {
            return TimeSpan.FromTicks((long)(this.blockTimes.Where(x => x.Ticks > 0).AverageOrDefault(x => x.Ticks) ?? 0));
        }

        protected override void SubDispose()
        {
            new IDisposable[]
            {
                this.chainStateBuilder,
                this.pruningWorker,
            }.DisposeList();
        }

        protected override void SubStart()
        {
            this.pruningWorker.Start();
        }

        protected override void SubStop()
        {
            this.pruningWorker.Stop();
        }

        protected override void WorkAction()
        {
            try
            {
                var chainStateLocal = this.chainState;

                if (this.chainStateBuilder != null
                    && this.chainStateBuilder.LastBlockHash != chainStateLocal.LastBlockHash
                    && DateTime.UtcNow - this.chainStateBuilderTime > MAX_BUILDER_LIFETIME)
                {
                    this.calculator.LogBlockchainProgress(this.chainStateBuilder);

                    // ensure rollback information is fully saved, back to pruning limit, before considering new chain state committed
                    var blocksPerDay = 144;
                    var pruneBuffer = blocksPerDay * 7;
                    foreach (var block in this.chainStateBuilder.Chain.Blocks.Reverse().Take(pruneBuffer))
                    {
                        if (block.Height > 0 && !this.blockRollbackCache.ContainsKey(block.BlockHash))
                            throw new InvalidOperationException();
                    }
                    this.blockRollbackCache.Flush();

                    //TODO keep the builder open favors performance when catching up
                    //TODO once caught up, it should switch over to quickly returning committed utxo's as new blocks come in
                    //TODO should be configurable, as doing this requires keeping two copies of the utxo on disk at all times
                    var newChain = this.chainStateBuilder.Chain.ToImmutable();
                    var newUtxo = this.chainStateBuilder.Utxo.ToImmutable(newChain.LastBlock.BlockHash);
                    //this.chainStateBuilder.Dispose();
                    //this.chainStateBuilder = null;
                    this.chainStateBuilderTime = DateTime.UtcNow;

                    UpdateCurrentBlockchain(new ChainState(newChain, newUtxo));
                    chainStateLocal = this.chainState;

                    this.pruningWorker.ForceWorkAndWait();
                }

                if (this.chainStateBuilder == null)
                {
                    this.chainStateBuilderTime = DateTime.UtcNow;
                    this.chainStateBuilder =
                        new ChainStateBuilder
                        (
                            chainStateLocal.Chain.ToBuilder(),
                            new UtxoBuilder(chainStateLocal.Utxo, this.logger, this.kernel, this.transactionCache)
                        );
                }

                // try to advance the blockchain with the new winning block
                using (var cancelToken = new CancellationTokenSource())
                {
                    this.calculator.CalculateBlockchainFromExisting(this.chainStateBuilder, getTargetChain, cancelToken.Token,
                        (blockTime) =>
                        {
                            this.blockTimesIndex = (this.blockTimesIndex + 1) % this.blockTimes.Length;
                            this.blockTimes[this.blockTimesIndex] = blockTime;

                            this.pruningWorker.NotifyWork();

                            var handler = this.OnChainStateBuilderChanged;
                            if (handler != null)
                                handler();

                            if (DateTime.UtcNow - this.chainStateBuilderTime > MAX_BUILDER_LIFETIME)
                            {
                                this.NotifyWork();
                                cancelToken.Cancel();
                            }
                        });
                }
            }
            catch (Exception e)
            {
                if (!(e is MissingDataException))
                {
                    this.logger.Warn(
                        string.Join("\n",
                            new string('-', 200),
                            "ChainStateWorker failure: {0}",
                            "{1}",
                            new string('-', 200)
                        )
                        .Format2(e.Message, e));
                }

                if (this.chainStateBuilder != null && !this.chainStateBuilder.IsConsistent)
                {
                    this.chainStateBuilder.Dispose();
                    this.chainStateBuilder = null;

                    var builderHandler = this.OnChainStateBuilderChanged;
                    if (builderHandler != null)
                        builderHandler();
                }

                if (e is ValidationException)
                {
                    var validationException = (ValidationException)e;
                    this.invalidBlockCache[validationException.BlockHash] = validationException.Message;
                }

                // try again on failure
                this.NotifyWork();
            }

            // collect after processing
            //GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, blocking: true);<commit_msg>cleanup chain state error logging<commit_after>﻿using BitSharp.Blockchain;
using BitSharp.Common;
using BitSharp.Common.ExtensionMethods;
using BitSharp.Data;
using BitSharp.Storage;
using Ninject;
using Ninject.Parameters;
using NLog;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace BitSharp.Daemon
{
    internal class ChainStateWorker : Worker
    {
        public event Action OnChainStateChanged;
        public event Action OnChainStateBuilderChanged;

        private static readonly TimeSpan MAX_BUILDER_LIFETIME = TimeSpan.FromMinutes(10);

        private readonly Logger logger;
        private Func<Chain> getTargetChain;
        private readonly IKernel kernel;
        private readonly IBlockchainRules rules;
        private readonly BlockchainCalculator calculator;
        private readonly TransactionCache transactionCache;
        private readonly BlockRollbackCache blockRollbackCache;
        private readonly InvalidBlockCache invalidBlockCache;

        private ChainState chainState;
        private readonly ReaderWriterLockSlim chainStateLock;

        private ChainStateBuilder chainStateBuilder;
        private DateTime chainStateBuilderTime;

        private readonly TimeSpan[] blockTimes;
        private int blockTimesIndex;

        private readonly PruningWorker pruningWorker;

        public ChainStateWorker(WorkerConfig workerConfig, Func<Chain> getTargetChain, Logger logger, IKernel kernel, IBlockchainRules rules, TransactionCache transactionCache, BlockRollbackCache blockRollbackCache, InvalidBlockCache invalidBlockCache)
            : base("ChainStateWorker", workerConfig.initialNotify, workerConfig.minIdleTime, workerConfig.maxIdleTime, logger)
        {
            this.logger = logger;
            this.getTargetChain = getTargetChain;
            this.kernel = kernel;
            this.rules = rules;
            this.transactionCache = transactionCache;
            this.blockRollbackCache = blockRollbackCache;
            this.invalidBlockCache = invalidBlockCache;

            this.calculator = kernel.Get<BlockchainCalculator>(new ConstructorArgument("shutdownToken", this.ShutdownToken.Token));

            this.chainState = ChainState.CreateForGenesisBlock(this.rules.GenesisChainedBlock);
            this.chainStateLock = new ReaderWriterLockSlim();

            this.blockTimes = new TimeSpan[10000];
            this.blockTimesIndex = -1;

            this.pruningWorker = kernel.Get<PruningWorker>(
                new ConstructorArgument("workerConfig", new WorkerConfig(initialNotify: false, minIdleTime: TimeSpan.FromSeconds(30), maxIdleTime: TimeSpan.FromMinutes(5))),
                new ConstructorArgument("getChainState", (Func<ChainState>)(() => this.chainState)),
                new ConstructorArgument("getChainStateBuilder", (Func<ChainStateBuilder>)(() => this.chainStateBuilder)));
        }

        public ChainState ChainState { get { return this.chainState; } }

        internal ChainStateBuilder ChainStateBuilder { get { return this.chainStateBuilder; } }

        public TimeSpan AverageBlockProcessingTime()
        {
            return TimeSpan.FromTicks((long)(this.blockTimes.Where(x => x.Ticks > 0).AverageOrDefault(x => x.Ticks) ?? 0));
        }

        protected override void SubDispose()
        {
            new IDisposable[]
            {
                this.chainStateBuilder,
                this.pruningWorker,
            }.DisposeList();
        }

        protected override void SubStart()
        {
            this.pruningWorker.Start();
        }

        protected override void SubStop()
        {
            this.pruningWorker.Stop();
        }

        protected override void WorkAction()
        {
            try
            {
                var chainStateLocal = this.chainState;

                if (this.chainStateBuilder != null
                    && this.chainStateBuilder.LastBlockHash != chainStateLocal.LastBlockHash
                    && DateTime.UtcNow - this.chainStateBuilderTime > MAX_BUILDER_LIFETIME)
                {
                    this.calculator.LogBlockchainProgress(this.chainStateBuilder);

                    // ensure rollback information is fully saved, back to pruning limit, before considering new chain state committed
                    var blocksPerDay = 144;
                    var pruneBuffer = blocksPerDay * 7;
                    foreach (var block in this.chainStateBuilder.Chain.Blocks.Reverse().Take(pruneBuffer))
                    {
                        if (block.Height > 0 && !this.blockRollbackCache.ContainsKey(block.BlockHash))
                            throw new InvalidOperationException();
                    }
                    this.blockRollbackCache.Flush();

                    //TODO keep the builder open favors performance when catching up
                    //TODO once caught up, it should switch over to quickly returning committed utxo's as new blocks come in
                    //TODO should be configurable, as doing this requires keeping two copies of the utxo on disk at all times
                    var newChain = this.chainStateBuilder.Chain.ToImmutable();
                    var newUtxo = this.chainStateBuilder.Utxo.ToImmutable(newChain.LastBlock.BlockHash);
                    //this.chainStateBuilder.Dispose();
                    //this.chainStateBuilder = null;
                    this.chainStateBuilderTime = DateTime.UtcNow;

                    UpdateCurrentBlockchain(new ChainState(newChain, newUtxo));
                    chainStateLocal = this.chainState;

                    this.pruningWorker.ForceWorkAndWait();
                }

                if (this.chainStateBuilder == null)
                {
                    this.chainStateBuilderTime = DateTime.UtcNow;
                    this.chainStateBuilder =
                        new ChainStateBuilder
                        (
                            chainStateLocal.Chain.ToBuilder(),
                            new UtxoBuilder(chainStateLocal.Utxo, this.logger, this.kernel, this.transactionCache)
                        );
                }

                // try to advance the blockchain with the new winning block
                using (var cancelToken = new CancellationTokenSource())
                {
                    this.calculator.CalculateBlockchainFromExisting(this.chainStateBuilder, getTargetChain, cancelToken.Token,
                        (blockTime) =>
                        {
                            this.blockTimesIndex = (this.blockTimesIndex + 1) % this.blockTimes.Length;
                            this.blockTimes[this.blockTimesIndex] = blockTime;

                            this.pruningWorker.NotifyWork();

                            var handler = this.OnChainStateBuilderChanged;
                            if (handler != null)
                                handler();

                            if (DateTime.UtcNow - this.chainStateBuilderTime > MAX_BUILDER_LIFETIME)
                            {
                                this.NotifyWork();
                                cancelToken.Cancel();
                            }
                        });
                }
            }
            catch (Exception e)
            {
                if (!(e is MissingDataException))
                {
                    this.logger.WarnException("ChainStateWorker exception", e);
                }

                if (this.chainStateBuilder != null && !this.chainStateBuilder.IsConsistent)
                {
                    this.chainStateBuilder.Dispose();
                    this.chainStateBuilder = null;

                    var builderHandler = this.OnChainStateBuilderChanged;
                    if (builderHandler != null)
                        builderHandler();
                }

                if (e is ValidationException)
                {
                    var validationException = (ValidationException)e;
                    this.invalidBlockCache[validationException.BlockHash] = validationException.Message;
                }

                // try again on failure
                this.NotifyWork();
            }

            // collect after processing
            //GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, blocking: true);
{'fraction_non_alphanumeric': 0.07152850647079398, 'fraction_numerical': 0.002739885740935059, 'mean_word_length': 1.7452392382781245, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '651364', 'n_tokens_mistral': 4055, 'n_tokens_neox': 3784, 'n_words': 872}