<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Turbocharged.Beanstalk
{
    class PhysicalConnection : IDisposable
    {
        TcpClient _client;
        NetworkStream _stream;
        IDisposable _receiveTask;

        BlockingCollection<Request> _requestsAwaitingResponse =
            new BlockingCollection<Request>(new ConcurrentQueue<Request>());

        PhysicalConnection()
        {
            _client = new TcpClient();
        }

        public static async Task<PhysicalConnection> ConnectAsync(string hostname, int port)
        {
            var conn = new PhysicalConnection();
            await conn._client.ConnectAsync(hostname, port).ConfigureAwait(false);
            conn._stream = conn._client.GetStream();
            var cts = new CancellationTokenSource();

#pragma warning disable 4014
            conn.ReceiveAsync(cts.Token);
#pragma warning restore 4014

            conn._receiveTask = Disposable.Create(() =>
            {
                cts.Cancel();
                cts.Dispose();
            });

            return conn;
        }

        public void Dispose()
        {
            using (_client)
            using (_stream)
            {
                _receiveTask.Dispose();
                _client.Close();
            }
        }

        public Task SendAsync<T>(Request<T> request, CancellationToken cancellationToken)
        {
            // TODO: Need a locking mechanism here so requests don't
            //       get inserted into the queue in a different order than
            //       they were sent over the wire (and therefore the
            //       order the server will respond in)

            _requestsAwaitingResponse.Add(request);
            var data = request.ToByteArray();
            return _stream.WriteAsync(data, 0, data.Length, cancellationToken);
        }
<commit_msg>Add a lock around the insertion of messages<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Turbocharged.Beanstalk
{
    class PhysicalConnection : IDisposable
    {
        TcpClient _client;
        NetworkStream _stream;
        IDisposable _receiveTask;
        SemaphoreSlim _insertionLock = new SemaphoreSlim(1, 1);

        BlockingCollection<Request> _requestsAwaitingResponse =
            new BlockingCollection<Request>(new ConcurrentQueue<Request>());

        PhysicalConnection()
        {
            _client = new TcpClient();
        }

        public static async Task<PhysicalConnection> ConnectAsync(string hostname, int port)
        {
            var conn = new PhysicalConnection();
            await conn._client.ConnectAsync(hostname, port).ConfigureAwait(false);
            conn._stream = conn._client.GetStream();
            var cts = new CancellationTokenSource();

#pragma warning disable 4014
            conn.ReceiveAsync(cts.Token);
#pragma warning restore 4014

            conn._receiveTask = Disposable.Create(() =>
            {
                cts.Cancel();
                cts.Dispose();
            });

            return conn;
        }

        public void Dispose()
        {
            using (_client)
            using (_stream)
            {
                _receiveTask.Dispose();
                _client.Close();
            }
        }

        public async Task SendAsync<T>(Request<T> request, CancellationToken cancellationToken)
        {
            await _insertionLock.WaitAsync().ConfigureAwait(false);
            try
            {
                _requestsAwaitingResponse.Add(request);
                var data = request.ToByteArray();
                await _stream.WriteAsync(data, 0, data.Length, cancellationToken).ConfigureAwait(false);
            }
            finally
            {
                _insertionLock.Release();
            }
        }

{'fraction_non_alphanumeric': 0.08285302593659942, 'fraction_numerical': 0.004803073967339097, 'mean_word_length': 1.9434628975265018, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '247445', 'n_tokens_mistral': 1070, 'n_tokens_neox': 1025, 'n_words': 248}