<commit_before>﻿// ------------------------------------------------------------
//  Copyright (c) Microsoft Corporation.  All rights reserved.
//  Licensed under the MIT License (MIT). See License.txt in the repo root for license information.
// ------------------------------------------------------------

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Validation;

using Microsoft.Diagnostics.EventFlow.Configuration;
using System.Diagnostics;

namespace Microsoft.Diagnostics.EventFlow
{
    public class DiagnosticPipeline : IDisposable
    {
        private List<IDisposable> pipelineLinkDisposables;
        private List<IDisposable> inputSubscriptions;
        private List<Task> pipelineCompletionTasks;
        private bool disposed;
        private bool disposeDependencies;
        private CancellationTokenSource cancellationTokenSource;
        private IDataflowBlock pipelineHead;
        private DiagnosticPipelineConfiguration pipelineConfiguration;
        private volatile int eventsInProgress;

        public DiagnosticPipeline(
            IHealthReporter healthReporter,
            IReadOnlyCollection<IObservable<EventData>> inputs,
            IReadOnlyCollection<IFilter> globalFilters,
            IReadOnlyCollection<EventSink> sinks,
            DiagnosticPipelineConfiguration pipelineConfiguration = null,
            bool disposeDependencies = false)
        {
            Requires.NotNull(healthReporter, nameof(healthReporter));
            Requires.NotNull(inputs, nameof(inputs));
            Requires.Argument(inputs.Count > 0, nameof(inputs), "There must be at least one input");
            Requires.NotNull(sinks, nameof(sinks));
            Requires.Argument(sinks.Count > 0, nameof(sinks), "There must be at least one sink");

            this.eventsInProgress = 0;

            this.pipelineConfiguration = pipelineConfiguration ?? new DiagnosticPipelineConfiguration();

            // An estimatie how many batches of events to allow inside the pipeline.
            // We want to be able to process full buffer of events, but also have enough batches in play in case of high concurrency.
            int MaxNumberOfBatchesInProgress = Math.Max(
                5 * this.pipelineConfiguration.MaxConcurrency,
                this.pipelineConfiguration.PipelineBufferSize / this.pipelineConfiguration.MaxEventBatchSize);

            this.Inputs = inputs;
            this.Sinks = sinks;
            
            // Just play nice and make sure there is always something to enumerate on
            this.GlobalFilters = globalFilters ?? new IFilter[0];

            this.HealthReporter = healthReporter;
            this.cancellationTokenSource = new CancellationTokenSource();
            var propagateCompletion = new DataflowLinkOptions() { PropagateCompletion = true };
            this.pipelineLinkDisposables = new List<IDisposable>();
            this.pipelineCompletionTasks = new List<Task>();

            var inputBuffer = new BufferBlock<EventData>(
                new DataflowBlockOptions()
                {
                    BoundedCapacity = this.pipelineConfiguration.PipelineBufferSize,
                    CancellationToken = this.cancellationTokenSource.Token
                });
            this.pipelineHead = inputBuffer;
            this.pipelineCompletionTasks.Add(inputBuffer.Completion);

            var batcher = new BatchBlock<EventData>(
                    this.pipelineConfiguration.MaxEventBatchSize,
                    new GroupingDataflowBlockOptions()
                    {
                        BoundedCapacity = this.pipelineConfiguration.MaxEventBatchSize,
                        CancellationToken = this.cancellationTokenSource.Token
                    }
                );
            this.pipelineLinkDisposables.Add(inputBuffer.LinkTo(batcher, propagateCompletion));
            this.pipelineCompletionTasks.Add(batcher.Completion);

            this.pipelineLinkDisposables.Add(new Timer(
                (unused) => batcher.TriggerBatch(),
                state: null,
                dueTime: TimeSpan.FromMilliseconds(this.pipelineConfiguration.MaxBatchDelayMsec),
                period: TimeSpan.FromMilliseconds(this.pipelineConfiguration.MaxBatchDelayMsec)));

            ISourceBlock<EventData[]> sinkSource;
            FilterAction filterTransform;

            if (this.GlobalFilters.Count > 0)
            {
                filterTransform = new FilterAction(
                    this.GlobalFilters,
                    this.cancellationTokenSource.Token,
                    this.pipelineConfiguration.MaxEventBatchSize,
                    this.pipelineConfiguration.MaxConcurrency,
                    healthReporter,
                    this.OnEventsFilteredOut);
                var globalFiltersBlock = filterTransform.GetFilterBlock();
                this.pipelineLinkDisposables.Add(batcher.LinkTo(globalFiltersBlock, propagateCompletion));
                this.pipelineCompletionTasks.Add(globalFiltersBlock.Completion);
                sinkSource = globalFiltersBlock;
            }
            else
            {
                sinkSource = batcher;
            }

            if (sinks.Count > 1)
            {
                // After broadcasting we will effectively have (sinks.Count - 1) * batch.Length more events in the pipeline, 
                // because the broadcaster is cloning the events for the sake of each sink (filters-output combination).
                var eventCounter = new TransformBlock<EventData[], EventData[]>(
                    (batch) => { Interlocked.Add(ref this.eventsInProgress, (sinks.Count - 1) * batch.Length);  return batch; },
                    new ExecutionDataflowBlockOptions()
                    {
                        BoundedCapacity = MaxNumberOfBatchesInProgress,
                        CancellationToken = this.cancellationTokenSource.Token
                    });
                this.pipelineLinkDisposables.Add(sinkSource.LinkTo(eventCounter, propagateCompletion));
                this.pipelineCompletionTasks.Add(eventCounter.Completion);

                var broadcaster = new BroadcastBlock<EventData[]>(
                    (events) => events?.Select((e) => e.DeepClone()).ToArray(),
                    new DataflowBlockOptions()
                    {<commit_msg>Correct buffer size for global filters block<commit_after>﻿// ------------------------------------------------------------
//  Copyright (c) Microsoft Corporation.  All rights reserved.
//  Licensed under the MIT License (MIT). See License.txt in the repo root for license information.
// ------------------------------------------------------------

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Validation;

using Microsoft.Diagnostics.EventFlow.Configuration;
using System.Diagnostics;

namespace Microsoft.Diagnostics.EventFlow
{
    public class DiagnosticPipeline : IDisposable
    {
        private List<IDisposable> pipelineLinkDisposables;
        private List<IDisposable> inputSubscriptions;
        private List<Task> pipelineCompletionTasks;
        private bool disposed;
        private bool disposeDependencies;
        private CancellationTokenSource cancellationTokenSource;
        private IDataflowBlock pipelineHead;
        private DiagnosticPipelineConfiguration pipelineConfiguration;
        private volatile int eventsInProgress;

        public DiagnosticPipeline(
            IHealthReporter healthReporter,
            IReadOnlyCollection<IObservable<EventData>> inputs,
            IReadOnlyCollection<IFilter> globalFilters,
            IReadOnlyCollection<EventSink> sinks,
            DiagnosticPipelineConfiguration pipelineConfiguration = null,
            bool disposeDependencies = false)
        {
            Requires.NotNull(healthReporter, nameof(healthReporter));
            Requires.NotNull(inputs, nameof(inputs));
            Requires.Argument(inputs.Count > 0, nameof(inputs), "There must be at least one input");
            Requires.NotNull(sinks, nameof(sinks));
            Requires.Argument(sinks.Count > 0, nameof(sinks), "There must be at least one sink");

            this.eventsInProgress = 0;

            this.pipelineConfiguration = pipelineConfiguration ?? new DiagnosticPipelineConfiguration();

            // An estimatie how many batches of events to allow inside the pipeline.
            // We want to be able to process full buffer of events, but also have enough batches in play in case of high concurrency.
            int MaxNumberOfBatchesInProgress = Math.Max(
                5 * this.pipelineConfiguration.MaxConcurrency,
                this.pipelineConfiguration.PipelineBufferSize / this.pipelineConfiguration.MaxEventBatchSize);

            this.Inputs = inputs;
            this.Sinks = sinks;
            
            // Just play nice and make sure there is always something to enumerate on
            this.GlobalFilters = globalFilters ?? new IFilter[0];

            this.HealthReporter = healthReporter;
            this.cancellationTokenSource = new CancellationTokenSource();
            var propagateCompletion = new DataflowLinkOptions() { PropagateCompletion = true };
            this.pipelineLinkDisposables = new List<IDisposable>();
            this.pipelineCompletionTasks = new List<Task>();

            var inputBuffer = new BufferBlock<EventData>(
                new DataflowBlockOptions()
                {
                    BoundedCapacity = this.pipelineConfiguration.PipelineBufferSize,
                    CancellationToken = this.cancellationTokenSource.Token
                });
            this.pipelineHead = inputBuffer;
            this.pipelineCompletionTasks.Add(inputBuffer.Completion);

            var batcher = new BatchBlock<EventData>(
                    this.pipelineConfiguration.MaxEventBatchSize,
                    new GroupingDataflowBlockOptions()
                    {
                        BoundedCapacity = this.pipelineConfiguration.MaxEventBatchSize,
                        CancellationToken = this.cancellationTokenSource.Token
                    }
                );
            this.pipelineLinkDisposables.Add(inputBuffer.LinkTo(batcher, propagateCompletion));
            this.pipelineCompletionTasks.Add(batcher.Completion);

            this.pipelineLinkDisposables.Add(new Timer(
                (unused) => batcher.TriggerBatch(),
                state: null,
                dueTime: TimeSpan.FromMilliseconds(this.pipelineConfiguration.MaxBatchDelayMsec),
                period: TimeSpan.FromMilliseconds(this.pipelineConfiguration.MaxBatchDelayMsec)));

            ISourceBlock<EventData[]> sinkSource;
            FilterAction filterTransform;

            if (this.GlobalFilters.Count > 0)
            {
                filterTransform = new FilterAction(
                    this.GlobalFilters,
                    this.cancellationTokenSource.Token,
                    MaxNumberOfBatchesInProgress,
                    this.pipelineConfiguration.MaxConcurrency,
                    healthReporter,
                    this.OnEventsFilteredOut);
                var globalFiltersBlock = filterTransform.GetFilterBlock();
                this.pipelineLinkDisposables.Add(batcher.LinkTo(globalFiltersBlock, propagateCompletion));
                this.pipelineCompletionTasks.Add(globalFiltersBlock.Completion);
                sinkSource = globalFiltersBlock;
            }
            else
            {
                sinkSource = batcher;
            }

            if (sinks.Count > 1)
            {
                // After broadcasting we will effectively have (sinks.Count - 1) * batch.Length more events in the pipeline, 
                // because the broadcaster is cloning the events for the sake of each sink (filters-output combination).
                var eventCounter = new TransformBlock<EventData[], EventData[]>(
                    (batch) => { Interlocked.Add(ref this.eventsInProgress, (sinks.Count - 1) * batch.Length);  return batch; },
                    new ExecutionDataflowBlockOptions()
                    {
                        BoundedCapacity = MaxNumberOfBatchesInProgress,
                        CancellationToken = this.cancellationTokenSource.Token
                    });
                this.pipelineLinkDisposables.Add(sinkSource.LinkTo(eventCounter, propagateCompletion));
                this.pipelineCompletionTasks.Add(eventCounter.Completion);

                var broadcaster = new BroadcastBlock<EventData[]>(
                    (events) => events?.Select((e) => e.DeepClone()).ToArray(),
                    new DataflowBlockOptions()
                    {
{'fraction_non_alphanumeric': 0.08793063761221515, 'fraction_numerical': 0.0013811094912913374, 'mean_word_length': 2.0661020936250294, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 47, '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': '5631615', 'n_tokens_mistral': 2943, 'n_tokens_neox': 2881, 'n_words': 670}