<commit_before>﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Threading.Tasks;
using Cowboy.Buffer;

namespace Cowboy.TcpLika
{
    internal class TcpLikaEngine
    {
        private TcpLikaCommandLineOptions _options;
        private Action<string> _logger = (s) => { };
        private IBufferManager _bufferManager;

        public TcpLikaEngine(TcpLikaCommandLineOptions options, Action<string> logger = null)
        {
            if (options == null)
                throw new ArgumentNullException("options");
            _options = options;

            if (logger != null)
                _logger = logger;

            if (_options.IsSetConnections)
                _bufferManager = new GrowingByteBufferManager(_options.Connections, 256);
            else
                _bufferManager = new GrowingByteBufferManager(10, 256);
        }

        public void Start()
        {
            int threads = 1;
            int connections = 1;
            int connectionsPerThread = 1;

            if (_options.IsSetThreads)
                threads = _options.Threads;
            if (threads > Environment.ProcessorCount)
                threads = Environment.ProcessorCount;

            if (_options.IsSetConnections)
                connections = _options.Connections;

            if (connections < threads)
                threads = connections;

            connectionsPerThread = connections / threads;

            var tasks = new List<Task>();
            for (int p = 0; p < threads; p++)
            {
                var task = Task.Run(async () =>
                {
                    foreach (var remoteEP in _options.RemoteEndPoints)
                    {
                        await PerformLoad(connectionsPerThread, remoteEP);
                    }
                });
                tasks.Add(task);
            }

            Task.WaitAll(tasks.ToArray());
        }

        private async Task PerformLoad(int connections, IPEndPoint remoteEP)
        {
            var channels = new List<TcpClient>();

            for (int c = 0; c < connections; c++)
            {
                var client = new TcpClient();
                Stream stream = null;
                try
                {
                    _logger(string.Format("Connecting to [{0}].", remoteEP));<commit_msg>resolved the connection remainer issue<commit_after>﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Threading.Tasks;
using Cowboy.Buffer;

namespace Cowboy.TcpLika
{
    internal class TcpLikaEngine
    {
        private TcpLikaCommandLineOptions _options;
        private Action<string> _logger = (s) => { };
        private IBufferManager _bufferManager;

        public TcpLikaEngine(TcpLikaCommandLineOptions options, Action<string> logger = null)
        {
            if (options == null)
                throw new ArgumentNullException("options");
            _options = options;

            if (logger != null)
                _logger = logger;

            if (_options.IsSetConnections)
                _bufferManager = new GrowingByteBufferManager(_options.Connections, 256);
            else
                _bufferManager = new GrowingByteBufferManager(10, 256);
        }

        public void Start()
        {
            int threads = 1;
            int connections = 1;

            if (_options.IsSetThreads)
                threads = _options.Threads;
            if (threads > Environment.ProcessorCount)
                threads = Environment.ProcessorCount;

            if (_options.IsSetConnections)
                connections = _options.Connections;

            if (connections < threads)
                threads = connections;

            int connectionsPerThread = connections / threads;
            int connectionsRemainder = connections % threads;

            var tasks = new List<Task>();
            for (int p = 0; p < threads; p++)
            {
                var task = Task.Run(async () =>
                {
                    foreach (var remoteEP in _options.RemoteEndPoints)
                    {
                        await PerformLoad(connectionsPerThread, remoteEP);
                    }
                });
                tasks.Add(task);
            }
            if (connectionsRemainder > 0)
            {
                var task = Task.Run(async () =>
                {
                    foreach (var remoteEP in _options.RemoteEndPoints)
                    {
                        await PerformLoad(connectionsRemainder, remoteEP);
                    }
                });
                tasks.Add(task);
            }

            Task.WaitAll(tasks.ToArray());
        }

        private async Task PerformLoad(int connections, IPEndPoint remoteEP)
        {
            var channels = new List<TcpClient>();

            for (int c = 0; c < connections; c++)
            {
                var client = new TcpClient();
                Stream stream = null;
                try
                {
                    _logger(string.Format("Connecting to [{0}].", remoteEP));
{'fraction_non_alphanumeric': 0.08214676889375684, 'fraction_numerical': 0.006206644760861628, 'mean_word_length': 1.5698874296435272, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 19, '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': '1760091', 'n_tokens_mistral': 1395, 'n_tokens_neox': 1289, 'n_words': 328}