<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace JsonRpc.Standard.Server
{
    /// <summary>
    /// Used to control the lifecycle of a JSON RPC service host.
    /// </summary>
    public interface IJsonRpcServiceHost
    {
        /// <summary>
        /// Asynchronously starts the JSON RPC service host.
        /// </summary>
        /// <param name="cancellationToken">The token used to shut down the service host.</param>
        /// <returns>A task that indicates the host state.</returns>
        /// <exception cref="InvalidOperationException">The service host is already running.</exception>
        Task RunAsync(CancellationToken cancellationToken);

        /// <summary>
        /// Requests to stop the JSON RPC service host.
        /// </summary>
        /// <remarks>This method will do nothing if the service is not started.</remarks>
        void Stop();
    }

    /// <summary>
    /// Provides options for <see cref="JsonRpcServiceHost"/>.
    /// </summary>
    [Flags]
    public enum JsonRpcServiceHostOptions
    {
        None = 0,
        /// <summary>
        /// Makes the response sequence consistent with the request order.
        /// </summary>
        ConsistentResponseSequence
    }

    /// <summary>
    /// Used to host a JSON RPC services.
    /// </summary>
    public class JsonRpcServiceHost : IJsonRpcServiceHost
    {
        private volatile CancellationTokenSource cts = null;
        private int isRunning = 0;
        private volatile SemaphoreSlim responseQueueSemaphore = null;
        private readonly LinkedList<ResponseSlot> responseQueue = new LinkedList<ResponseSlot>();

        public JsonRpcServiceHost(MessageReader reader, MessageWriter writer,
            IRpcMethodResolver resolver, JsonRpcServiceHostOptions options)
        {
            if (reader == null) throw new ArgumentNullException(nameof(reader));
            if (writer == null) throw new ArgumentNullException(nameof(writer));
            if (resolver == null) throw new ArgumentNullException(nameof(resolver));
            Reader = reader;
            Writer = writer;
            Resolver = resolver;
            Options = options;
        }

        public MessageReader Reader { get; }

        public MessageWriter Writer { get; }

        public IRpcMethodResolver Resolver { get; }

        public ISession Session { get; set; }

        public JsonRpcServiceHostOptions Options { get; }

        /// <summary>
        /// Asynchronously starts the JSON RPC service host.
        /// </summary>
        /// <param name="cancellationToken">The token used to shut down the service host.</param>
        /// <returns>A task that indicates the host state.</returns>
        public async Task RunAsync(CancellationToken cancellationToken)
        {
            var localRunning = Interlocked.Exchange(ref isRunning, 1);
            if (localRunning != 0) throw new InvalidOperationException("The service host is already running.");
            using (responseQueueSemaphore = new SemaphoreSlim(0))
            using (cts = new CancellationTokenSource())
            using (var linked = CancellationTokenSource.CreateLinkedTokenSource(cts.Token, cancellationToken))
            {
                var cancellationTcs = new TaskCompletionSource<bool>();
                var readerTask = Task.Factory.StartNew(state => ReaderEntryPoint((CancellationToken) state),
                    linked.Token, linked.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
                var writerTask = Task.Factory.StartNew(state => WriterEntryPoint((CancellationToken) state),
                    linked.Token, linked.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
                // Wait for Stop() or user cancellation.
                using (cancellationToken.Register(state => ((TaskCompletionSource<bool>) state).SetResult(true),
                    cancellationTcs))
                {
                    var finished = await Task.WhenAny(readerTask, cancellationTcs.Task);
                    if (finished == readerTask)
                    {
                        // Reader has reached EOF. Stop the server.
                        this.Stop();
                    }
                }
                // Wait some time for the writer task to finish.
                // await Task.WhenAny(writerTask.ContinueWith(_ => { }), Task.Delay(2000));
                // Cleanup.
                lock (responseQueue) responseQueue.Clear();
            }
            Interlocked.Exchange(ref isRunning, 0);
        }

        /// <inheritdoc />
        public void Stop()
        {<commit_msg>JsonRpcServiceHost waits for writerTask when readerTask has completed/been canceled.<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace JsonRpc.Standard.Server
{
    /// <summary>
    /// Used to control the lifecycle of a JSON RPC service host.
    /// </summary>
    public interface IJsonRpcServiceHost
    {
        /// <summary>
        /// Asynchronously starts the JSON RPC service host.
        /// </summary>
        /// <param name="cancellationToken">The token used to shut down the service host.</param>
        /// <returns>A task that indicates the host state.</returns>
        /// <exception cref="InvalidOperationException">The service host is already running.</exception>
        Task RunAsync(CancellationToken cancellationToken);

        /// <summary>
        /// Requests to stop the JSON RPC service host.
        /// </summary>
        /// <remarks>This method will do nothing if the service is not started.</remarks>
        void Stop();
    }

    /// <summary>
    /// Provides options for <see cref="JsonRpcServiceHost"/>.
    /// </summary>
    [Flags]
    public enum JsonRpcServiceHostOptions
    {
        None = 0,
        /// <summary>
        /// Makes the response sequence consistent with the request order.
        /// </summary>
        ConsistentResponseSequence
    }

    /// <summary>
    /// Used to host a JSON RPC services.
    /// </summary>
    public class JsonRpcServiceHost : IJsonRpcServiceHost
    {
        private volatile CancellationTokenSource cts = null;
        private int isRunning = 0;
        private volatile SemaphoreSlim responseQueueSemaphore = null;
        private readonly LinkedList<ResponseSlot> responseQueue = new LinkedList<ResponseSlot>();

        public JsonRpcServiceHost(MessageReader reader, MessageWriter writer,
            IRpcMethodResolver resolver, JsonRpcServiceHostOptions options)
        {
            if (reader == null) throw new ArgumentNullException(nameof(reader));
            if (writer == null) throw new ArgumentNullException(nameof(writer));
            if (resolver == null) throw new ArgumentNullException(nameof(resolver));
            Reader = reader;
            Writer = writer;
            Resolver = resolver;
            Options = options;
        }

        public MessageReader Reader { get; }

        public MessageWriter Writer { get; }

        public IRpcMethodResolver Resolver { get; }

        public ISession Session { get; set; }

        public JsonRpcServiceHostOptions Options { get; }

        /// <summary>
        /// Asynchronously starts the JSON RPC service host.
        /// </summary>
        /// <param name="cancellationToken">The token used to shut down the service host.</param>
        /// <returns>A task that indicates the host state.</returns>
        public async Task RunAsync(CancellationToken cancellationToken)
        {
            var localRunning = Interlocked.Exchange(ref isRunning, 1);
            if (localRunning != 0) throw new InvalidOperationException("The service host is already running.");
            using (responseQueueSemaphore = new SemaphoreSlim(0))
            using (cts = new CancellationTokenSource())
            using (var linked = CancellationTokenSource.CreateLinkedTokenSource(cts.Token, cancellationToken))
            {
                var cancellationTcs = new TaskCompletionSource<bool>();
                var readerTask = Task.Factory.StartNew(state => ReaderEntryPoint((CancellationToken) state),
                    linked.Token, linked.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
                var writerTask = Task.Factory.StartNew(state => WriterEntryPoint((CancellationToken) state),
                    linked.Token, linked.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
                // Tasks that we can safely await
                var readerTask1 = readerTask.ContinueWith(_ => { });
                var writerTask1 = readerTask.ContinueWith(_ => { });
                // Wait for Stop() or user cancellation.
                using (cancellationToken.Register(state => ((TaskCompletionSource<bool>) state).SetResult(true),
                    cancellationTcs))
                {
                    var finished = await Task.WhenAny(readerTask1, cancellationTcs.Task);
                    if (finished == readerTask1 && readerTask.IsCompleted)
                    {
                        // Reader has reached EOF. Wait for the writer to finish, then stop the server.
                        while (true)
                        {
                            int queueSize;
                            lock (responseQueue) queueSize = responseQueue.Count;
                            if (queueSize == 0) break;
                            await Task.WhenAny(writerTask1, Task.Delay(1000, linked.Token));
                        }
                        this.Stop();
                    }
                }
                // Cleanup.
                lock (responseQueue) responseQueue.Clear();
            }
            Interlocked.Exchange(ref isRunning, 0);
        }

        /// <inheritdoc />
        public void Stop()
        {
{'fraction_non_alphanumeric': 0.09378721714966257, 'fraction_numerical': 0.0025803890432711395, 'mean_word_length': 2.017969451931716, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 76, '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': '6637315', 'n_tokens_mistral': 2415, 'n_tokens_neox': 2363, 'n_words': 702}