<commit_before>﻿using System;
using System.Threading;
using System.Threading.Tasks;

namespace ALinq
{
    internal class Signal<T>
    {
        private TaskCompletionSource<T> _tcs = new TaskCompletionSource<T>();

        public void Set(T value = default(T))
        {
            _tcs.TrySetResult(value);
        }

        public async Task<T> Wait()
        {
            var result = await _tcs.Task.ConfigureAwait(false);
            Interlocked.Exchange(ref _tcs, new TaskCompletionSource<T>());
            return result;
        }
    }

    internal sealed class ConcurrentAsyncEnumerator<T> : IAsyncEnumerator<T>
    {
        private readonly Func<ConcurrentAsyncEnumerator<T>, Task> _producer;
        private readonly Signal<bool> _producerGate = new Signal<bool>();
        private readonly Signal<bool> _producerResult = new Signal<bool>();
        private Task _producerTask;
        private T _current;

        T IAsyncEnumerator<T>.Current
        {
            get { return _current; }
        }

        object IAsyncEnumerator.Current
        {
            get { return _current; }
        }

        async Task<bool> IAsyncEnumerator.MoveNext()
        {
            try
            {
                if(_producerTask == null)
                {
                    _producerTask = _producer(this);
                }

                var hasNext = await SignalProducer().ConfigureAwait(false);
                if(!hasNext)
                {
                    // Marshal exceptions and cancellation back to the consumer
                    await _producerTask.ConfigureAwait(false);
                    return false;
                }
                return true;
            }
            catch (InvalidOperationException)
            {
                return false;
            }
        }

        private async Task<bool> SignalProducer()
        {
            _producerGate.Set();
            return await _producerResult.Wait().ConfigureAwait(false);
        }

        private async Task Yield(T value)
        {
            // Wait for the consumer to pull (call MoveNext()).
            await _producerGate.Wait().ConfigureAwait(false);
            _current = value;
            _producerResult.Set(true);
        }

        private async Task EndOfStream()
        {
            // Wait for the consumer to pull (call MoveNext()).
            await _producerGate.Wait().ConfigureAwait(false);
            _producerResult.Set();
        }

        internal ConcurrentAsyncEnumerator(Func<ConcurrentAsyncProducer<T>, Task> producerFunc)
        {
            _producer = async enumerator =>
            {
                var producer = new ConcurrentAsyncProducer<T>(item => enumerator.Yield(item));

                try
                {
                    await producerFunc(producer);
                }
                finally
                {<commit_msg>Keep the state machine on hold while the consumer is using the Current instance after the MoveNext has been called for the first time. See the `ConcurrentAsyncEnumerator<T>.Yield` implementation for more details.<commit_after>﻿using System;
using System.Threading;
using System.Threading.Tasks;

namespace ALinq
{
    internal class Signal<T>
    {
        private TaskCompletionSource<T> _tcs = new TaskCompletionSource<T>();

        public void Set(T value = default(T))
        {
            _tcs.TrySetResult(value);
        }

        public async Task<T> Wait()
        {
            var result = await _tcs.Task.ConfigureAwait(false);
            Interlocked.Exchange(ref _tcs, new TaskCompletionSource<T>());
            return result;
        }
    }

    internal sealed class ConcurrentAsyncEnumerator<T> : IAsyncEnumerator<T>
    {
        private readonly Func<ConcurrentAsyncEnumerator<T>, Task> _producer;
        private readonly Signal<bool> _moveNextSignal = new Signal<bool>();
        private readonly Signal<bool> _producerResult = new Signal<bool>();
        private Task _producerTask;
        private T _current;

        T IAsyncEnumerator<T>.Current
        {
            get { return _current; }
        }

        object IAsyncEnumerator.Current
        {
            get { return _current; }
        }

        async Task<bool> IAsyncEnumerator.MoveNext()
        {
            try
            {
                if(_producerTask == null)
                {
                    _producerTask = _producer(this);
                }

                var hasNext = await SignalProducer().ConfigureAwait(false);
                if(!hasNext)
                {
                    // Marshal exceptions and cancellation back to the consumer
                    await _producerTask.ConfigureAwait(false);
                    return false;
                }
                return true;
            }
            catch (InvalidOperationException)
            {
                return false;
            }
        }

        private async Task<bool> SignalProducer()
        {
            _moveNextSignal.Set();
            return await _producerResult.Wait().ConfigureAwait(false);
        }

        private async Task Yield(T value)
        {
            // Wait for the consumer to call MoveNext() the first time.
            // Not that after the first time, this will be always signalled.
            // This is only necessary to synchronize the enumerator when the 
            // producer is started.
            await _moveNextSignal.Wait().ConfigureAwait(false);
            _current = value;
            _producerResult.Set(true);

            // Now, wait for the consumer to call the next MoveNext().
            // This will guarantee that _current instance is valid until
            // the consumer(s) are done.
            await _moveNextSignal.Wait().ConfigureAwait(false);
            _moveNextSignal.Set();
        }

        private async Task EndOfStream()
        {
            // Wait for the consumer to pull (call MoveNext()).
            await _moveNextSignal.Wait().ConfigureAwait(false);
            _producerResult.Set();
            // There no need here to weird stuff with _moveNextSignal :)
        }

        internal ConcurrentAsyncEnumerator(Func<ConcurrentAsyncProducer<T>, Task> producerFunc)
        {
            _producer = async enumerator =>
            {
                var producer = new ConcurrentAsyncProducer<T>(item => enumerator.Yield(item));

                try
                {
                    await producerFunc(producer);
                }
                finally
                {
{'fraction_non_alphanumeric': 0.07840146430750458, 'fraction_numerical': 0.0, 'mean_word_length': 1.5876085240726125, 'pattern_counts': {'":': 0, '<': 42, '<?xml version=': 0, '>': 46, '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': '5146883', 'n_tokens_mistral': 1659, 'n_tokens_neox': 1606, 'n_words': 415}