<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Common.Core;
using Microsoft.R.Host.Client;
using Microsoft.R.Support.Settings;
using Microsoft.VisualStudio.R.Package.Shell;
using Microsoft.VisualStudio.Shell;
using Microsoft.VisualStudio.Shell.Interop;
using Microsoft.VisualStudio.Threading;

namespace Microsoft.VisualStudio.R.Package.Repl
{
    using Task = System.Threading.Tasks.Task;

    internal sealed class RSession : IRSession, IRCallbacks
    {
        private readonly RHost _host;
        private readonly TaskCompletionSource<object> _initializationTcs;
        private readonly ConcurrentQueue<RSessionRequestSource> _pendingRequestSources = new ConcurrentQueue<RSessionRequestSource>();
        private readonly Stack<RSessionRequestSource> _currentRequestSources = new Stack<RSessionRequestSource>();
        private IRExpressionEvaluator _expressionEvaluator;

        public event EventHandler<RBeforeRequestEventArgs> BeforeRequest;
        public event EventHandler<RResponseEventArgs> Response;
        public event EventHandler<RErrorEventArgs> Error;

        /// <summary>
        /// ReadConsole requires a task even if there are no pending requests
        /// </summary>
        private TaskCompletionSource<string> _nextRequestTcs;
        private IReadOnlyCollection<IRContext> _contexts;

        public string Prompt { get; private set; } = "> ";
        public int MaxLength { get; private set; } = 0x1000;

        public RSession()
        {
            _host = new RHost(this);
            _initializationTcs = new TaskCompletionSource<object>();
        }

        public void Dispose()
        {
            _host.Dispose();
        }

        public Task<IRSessionInteraction> BeginInteractionAsync(bool isVisible = true)
        {
            var requestTcs = GetRequestTcs();
            RSessionRequestSource requestSource;
            if (requestTcs == null)
            {
                requestSource = new RSessionRequestSource(isVisible, _contexts);
                _pendingRequestSources.Enqueue(requestSource);
                
            }
            else
            {
                requestSource = new RSessionRequestSource(isVisible, _contexts, requestTcs);
                requestSource.BeginInteractionAsync(Prompt, MaxLength, _expressionEvaluator);
                _currentRequestSources.Push(requestSource);
            }

            return requestSource.CreateRequestTask;
        }

        public async Task InitializeAsync()
        {
            var psi = new ProcessStartInfo { WorkingDirectory = RToolsSettings.GetBinariesFolder() };
            var hostTask = _host.CreateAndRun(psi);
            await Task.WhenAny(_initializationTcs.Task, hostTask);

            // If hostTask is the one that has completed by now, it has probably faulted, so
            // await it to let the exception bubble up - WhenAny always completes successfully.
            if (hostTask.IsCompleted) {
                await hostTask;
            }
        }

        private TaskCompletionSource<string> GetRequestTcs()
        {
            SpinWait spin = new SpinWait();
            while (true)
            {
                var requestTsc = Interlocked.Exchange(ref _nextRequestTcs, null);
                if (requestTsc != null)
                {
                    return requestTsc;
                }

                if (_pendingRequestSources.Count > 0)
                {
                    return null;<commit_msg>Use Task.Unwrap to simplify code.<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Common.Core;
using Microsoft.R.Host.Client;
using Microsoft.R.Support.Settings;
using Microsoft.VisualStudio.R.Package.Shell;
using Microsoft.VisualStudio.Shell;
using Microsoft.VisualStudio.Shell.Interop;
using Microsoft.VisualStudio.Threading;

namespace Microsoft.VisualStudio.R.Package.Repl
{
    using Task = System.Threading.Tasks.Task;

    internal sealed class RSession : IRSession, IRCallbacks
    {
        private readonly RHost _host;
        private readonly TaskCompletionSource<object> _initializationTcs;
        private readonly ConcurrentQueue<RSessionRequestSource> _pendingRequestSources = new ConcurrentQueue<RSessionRequestSource>();
        private readonly Stack<RSessionRequestSource> _currentRequestSources = new Stack<RSessionRequestSource>();
        private IRExpressionEvaluator _expressionEvaluator;

        public event EventHandler<RBeforeRequestEventArgs> BeforeRequest;
        public event EventHandler<RResponseEventArgs> Response;
        public event EventHandler<RErrorEventArgs> Error;

        /// <summary>
        /// ReadConsole requires a task even if there are no pending requests
        /// </summary>
        private TaskCompletionSource<string> _nextRequestTcs;
        private IReadOnlyCollection<IRContext> _contexts;

        public string Prompt { get; private set; } = "> ";
        public int MaxLength { get; private set; } = 0x1000;

        public RSession()
        {
            _host = new RHost(this);
            _initializationTcs = new TaskCompletionSource<object>();
        }

        public void Dispose()
        {
            _host.Dispose();
        }

        public Task<IRSessionInteraction> BeginInteractionAsync(bool isVisible = true)
        {
            var requestTcs = GetRequestTcs();
            RSessionRequestSource requestSource;
            if (requestTcs == null)
            {
                requestSource = new RSessionRequestSource(isVisible, _contexts);
                _pendingRequestSources.Enqueue(requestSource);
                
            }
            else
            {
                requestSource = new RSessionRequestSource(isVisible, _contexts, requestTcs);
                requestSource.BeginInteractionAsync(Prompt, MaxLength, _expressionEvaluator);
                _currentRequestSources.Push(requestSource);
            }

            return requestSource.CreateRequestTask;
        }

        public async Task InitializeAsync()
        {
            var psi = new ProcessStartInfo { WorkingDirectory = RToolsSettings.GetBinariesFolder() };
            await Task.WhenAny(_initializationTcs.Task, _host.CreateAndRun(psi)).Unwrap();
        }

        private TaskCompletionSource<string> GetRequestTcs()
        {
            SpinWait spin = new SpinWait();
            while (true)
            {
                var requestTsc = Interlocked.Exchange(ref _nextRequestTcs, null);
                if (requestTsc != null)
                {
                    return requestTsc;
                }

                if (_pendingRequestSources.Count > 0)
                {
                    return null;
{'fraction_non_alphanumeric': 0.07289586305278174, 'fraction_numerical': 0.0017118402282453639, 'mean_word_length': 2.344942748091603, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 37, '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': '7200585', 'n_tokens_mistral': 1740, 'n_tokens_neox': 1669, 'n_words': 407}