<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace PostSimulationAnalysisOldRuntime
{


    public static class Program
    {

        private static Trial[] LoadDataParallel(string path)
        {
            BlockingCollection<string> inputLines = new BlockingCollection<string>();
            ConcurrentBag<Trial> trials = new ConcurrentBag<Trial>();
            int itemsAdded = 0;
            int itemsProcessed = 0;

            Parallel.Invoke(
                () =>
                {
                    int counter = 0;
                    using (StreamReader file = new StreamReader(path))
                    {
                        // Skip header
                        file.ReadLine();
                        string line;
                        while ((line = file.ReadLine()) != null)
                        {
                            inputLines.Add(line);
                            ++itemsAdded;
                            if (++counter >= 100000)
                            {
                                counter = 0;
                                Console.Out.WriteLine("Trial lines Read: {0}", itemsAdded);
                            }
                        }
                        inputLines.CompleteAdding();
                    }

                }, () => Parallel.ForEach(inputLines.GetConsumingPartitioner(), (line, state) =>
                {
                    string[] fields = line.Trim().Split(new char[] {','});
                    if (fields.Length >= 11)
                    {
                        Trial trial = new Trial(fields);
                        trials.Add(trial);
                    }
                    int numProcessed = Interlocked.Increment(ref itemsProcessed);
                    if (numProcessed % 100000 == 0)
                        Console.Out.WriteLine("Trial lines processed: {0}", numProcessed);
                })

                    )
                    ;
            return trials.ToArray();
        }

        private static List<Trial> LoadData(string path)
        {
            List<Trial> trials = new List<Trial>();
            int counter = 0;
            using (StreamReader file = new StreamReader(path))
            {
                // Skip header
                file.ReadLine();

                string line;
                while ((line = file.ReadLine()) != null)
                {
                    string[] fields = line.Trim().Split(new char[] { ',' });
                    if (fields.Length >= 11)
                        trials.Add(new Trial(fields));
                    if (++counter >= 10000)
                    {
                        counter = 0;
                        Console.Out.WriteLine("Loaded: {0}", trials.Count);
                    }
                }
            }
            return trials;
        }
<commit_msg>Improving parallel load speed<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace PostSimulationAnalysisOldRuntime
{


    public static class Program
    {

        private static Trial[] LoadDataParallel(string path)
        {
            BlockingCollection<string> inputLines = new BlockingCollection<string>();
            ConcurrentBag<Trial> trials = new ConcurrentBag<Trial>();
            int itemsAdded = 0;
            int itemsProcessed = 0;

            List<Task> tasks = new List<Task>();
            tasks.Add(Task.Run(() =>
            {
                int counter = 0;
                using (StreamReader file = new StreamReader(path))
                {
                    // Skip header
                    file.ReadLine();
                    string line;
                    while ((line = file.ReadLine()) != null)
                    {
                        inputLines.Add(line);
                        ++itemsAdded;
                        if (++counter >= 100000)
                        {
                            counter = 0;
                            Console.Out.WriteLine("Trial lines Read: {0}", itemsAdded);
                        }
                    }
                    inputLines.CompleteAdding();
                }
            }));

            for (int i = 0; i < Environment.ProcessorCount; i++)
            {
                tasks.Add(Task.Run(() =>
                {
                    string line;
                    while (inputLines.TryTake(out line, -1))
                    {
                        string[] fields = line.Trim().Split(new char[] {','});
                        if (fields.Length >= 11)
                        {
                            Trial trial = new Trial(fields);
                            trials.Add(trial);
                        }
                        int numProcessed = Interlocked.Increment(ref itemsProcessed);
                        if (numProcessed%100000 == 0)
                            Console.Out.WriteLine("Trial lines processed: {0}", numProcessed);
                    }
                }));
            }

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

        private static List<Trial> LoadData(string path)
        {
            List<Trial> trials = new List<Trial>();
            int counter = 0;
            using (StreamReader file = new StreamReader(path))
            {
                // Skip header
                file.ReadLine();

                string line;
                while ((line = file.ReadLine()) != null)
                {
                    string[] fields = line.Trim().Split(new char[] { ',' });
                    if (fields.Length >= 11)
                        trials.Add(new Trial(fields));
                    if (++counter >= 10000)
                    {
                        counter = 0;
                        Console.Out.WriteLine("Loaded: {0}", trials.Count);
                    }
                }
            }
            return trials;
        }

{'fraction_non_alphanumeric': 0.0978226064681396, 'fraction_numerical': 0.010246557796990073, 'mean_word_length': 1.0401698236446766, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 32, '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': '2482918', 'n_tokens_mistral': 1506, 'n_tokens_neox': 1330, 'n_words': 312}