<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Codex.ObjectModel;
using System.Collections.Concurrent;
using Codex.Sdk.Utilities;
using System.IO;
using Codex.Utilities;
using Codex.Serialization;
using System.Diagnostics.Contracts;
using System.Runtime.CompilerServices;
using Newtonsoft.Json;
using Codex.Storage.DataModel;
using Codex.ElasticSearch.Utilities;
using Codex.Logging;
using System.Threading;

namespace Codex.ElasticSearch.Store
{
    public class DirectoryCodexStore : ICodexStore, ICodexRepositoryStore
    {
        private readonly ConcurrentQueue<ValueTask<None>> backgroundTasks = new ConcurrentQueue<ValueTask<None>>();
        private readonly ConcurrentQueue<CommitFileLink> commitFiles = new ConcurrentQueue<CommitFileLink>();

        public readonly string DirectoryPath;
        private const string EntityFileExtension = ".cdx.json";

        // These two files should contain the same content after finalization
        private const string RepositoryInfoFileName = "repo" + EntityFileExtension;
        private const string RepositoryInitializationFileName = "initrepo" + EntityFileExtension;

        private RepositoryStoreInfo m_storeInfo;

        private Logger logger;

        /// <summary>
        /// Disables optimized serialization for use when testing
        /// </summary>
        public bool DisableOptimization { get; set; }
        public bool Clean { get; set; }

        public DirectoryCodexStore(string directory, Logger logger = null)
        {
            DirectoryPath = directory;
            this.logger = logger ?? Logger.Null;
        }

        public static IEnumerable<string> GetEntityFiles(string directory)
        {
            if (Directory.Exists(directory))
            {
                return Directory.GetFiles(directory, "*" + EntityFileExtension, SearchOption.AllDirectories);
            }

            return CollectionUtilities.Empty<string>.Array;
        }

        public async Task ReadAsync(ICodexStore store, bool finalize = true, string repositoryName = null)
        {
            FileSystem fileSystem;
            if (Directory.Exists(DirectoryPath))
            {
                fileSystem = new DirectoryFileSystem(DirectoryPath, "*" + EntityFileExtension);
            }
            else
            {
                fileSystem = new ZipFileSystem(DirectoryPath);
            }

            using (fileSystem)
            {
                m_storeInfo = Read<RepositoryStoreInfo>(fileSystem, RepositoryInitializationFileName);
                m_storeInfo.Repository.Name = repositoryName ?? m_storeInfo.Repository.Name ?? m_storeInfo.Commit.RepositoryName;

                ConcurrentQueue<Func<Task>> actionQueue = new ConcurrentQueue<Func<Task>>();

                logger.LogMessage("Reading repository information");
                var repositoryStore = await store.CreateRepositoryStore(m_storeInfo.Repository, m_storeInfo.Commit, m_storeInfo.Branch);

                logger.LogMessage($"Read repository information (repo name: {m_storeInfo.Repository.Name})");

                int nextIndex = 0;
                int count = 0;
                List<Task> tasks = new List<Task>();
                foreach (var kind in StoredEntityKind.Kinds)
                {
                    // TODO: Do we even need concurrency here?
                    tasks.Add(Task.Run(() =>
                    {
                        var kindDirectoryPath = Path.Combine(DirectoryPath, kind.Name);
                        logger.LogMessage($"Reading {kind} infos from {kindDirectoryPath}");
                        foreach (var file in fileSystem.GetFiles(kind.Name))
                        {
                            if (fileSystem.GetFileSize(file) > 1 << 20)
                            {
                                // Ignore files larger than 1 MB
                                continue;
                            }

                            actionQueue.Enqueue(async () =>
                            {
                                var i = Interlocked.Increment(ref nextIndex);
                                logger.LogMessage($"{i}/{count}: Reading {kind} info at {file}");
                                await kind.Add(this, fileSystem, file, repositoryStore);
                                logger.LogMessage($"{i}/{count}: Added {file} to store.");
                            });
                        }
                    }
                    ));
                }

                await Task.WhenAll(tasks);

                count = actionQueue.Count;
                tasks.Clear();
                for (int i = 0; i < Math.Min(Environment.ProcessorCount, 32); i++)
                {
                    tasks.Add(Task.Run(async () =>
                    {
                        while (actionQueue.TryDequeue(out var taskFactory))
                        {
                            await taskFactory();
                        }
                    }
                    ));
                }
<commit_msg>Put a try/catch around file Add so that a crash in there doesn't tear down the process.<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Codex.ObjectModel;
using System.Collections.Concurrent;
using Codex.Sdk.Utilities;
using System.IO;
using Codex.Utilities;
using Codex.Serialization;
using System.Diagnostics.Contracts;
using System.Runtime.CompilerServices;
using Newtonsoft.Json;
using Codex.Storage.DataModel;
using Codex.ElasticSearch.Utilities;
using Codex.Logging;
using System.Threading;

namespace Codex.ElasticSearch.Store
{
    public class DirectoryCodexStore : ICodexStore, ICodexRepositoryStore
    {
        private readonly ConcurrentQueue<ValueTask<None>> backgroundTasks = new ConcurrentQueue<ValueTask<None>>();
        private readonly ConcurrentQueue<CommitFileLink> commitFiles = new ConcurrentQueue<CommitFileLink>();

        public readonly string DirectoryPath;
        private const string EntityFileExtension = ".cdx.json";

        // These two files should contain the same content after finalization
        private const string RepositoryInfoFileName = "repo" + EntityFileExtension;
        private const string RepositoryInitializationFileName = "initrepo" + EntityFileExtension;

        private RepositoryStoreInfo m_storeInfo;

        private Logger logger;

        /// <summary>
        /// Disables optimized serialization for use when testing
        /// </summary>
        public bool DisableOptimization { get; set; }
        public bool Clean { get; set; }

        public DirectoryCodexStore(string directory, Logger logger = null)
        {
            DirectoryPath = directory;
            this.logger = logger ?? Logger.Null;
        }

        public static IEnumerable<string> GetEntityFiles(string directory)
        {
            if (Directory.Exists(directory))
            {
                return Directory.GetFiles(directory, "*" + EntityFileExtension, SearchOption.AllDirectories);
            }

            return CollectionUtilities.Empty<string>.Array;
        }

        public async Task ReadAsync(ICodexStore store, bool finalize = true, string repositoryName = null)
        {
            FileSystem fileSystem;
            if (Directory.Exists(DirectoryPath))
            {
                fileSystem = new DirectoryFileSystem(DirectoryPath, "*" + EntityFileExtension);
            }
            else
            {
                fileSystem = new ZipFileSystem(DirectoryPath);
            }

            using (fileSystem)
            {
                m_storeInfo = Read<RepositoryStoreInfo>(fileSystem, RepositoryInitializationFileName);
                m_storeInfo.Repository.Name = repositoryName ?? m_storeInfo.Repository.Name ?? m_storeInfo.Commit.RepositoryName;

                ConcurrentQueue<Func<Task>> actionQueue = new ConcurrentQueue<Func<Task>>();

                logger.LogMessage("Reading repository information");
                var repositoryStore = await store.CreateRepositoryStore(m_storeInfo.Repository, m_storeInfo.Commit, m_storeInfo.Branch);

                logger.LogMessage($"Read repository information (repo name: {m_storeInfo.Repository.Name})");

                int nextIndex = 0;
                int count = 0;
                List<Task> tasks = new List<Task>();
                foreach (var kind in StoredEntityKind.Kinds)
                {
                    // TODO: Do we even need concurrency here?
                    tasks.Add(Task.Run(() =>
                    {
                        var kindDirectoryPath = Path.Combine(DirectoryPath, kind.Name);
                        logger.LogMessage($"Reading {kind} infos from {kindDirectoryPath}");
                        foreach (var file in fileSystem.GetFiles(kind.Name))
                        {
                            if (fileSystem.GetFileSize(file) > 1 << 20)
                            {
                                // Ignore files larger than 1 MB
                                continue;
                            }

                            actionQueue.Enqueue(async () =>
                            {
                                var i = Interlocked.Increment(ref nextIndex);
                                logger.LogMessage($"{i}/{count}: Reading {kind} info at {file}");
                                try
                                {
                                    await kind.Add(this, fileSystem, file, repositoryStore);
                                }
                                catch (Exception ex)
                                {
                                    logger.LogExceptionError("AddFile", ex);
                                }

                                logger.LogMessage($"{i}/{count}: Added {file} to store.");
                            });
                        }
                    }
                    ));
                }

                await Task.WhenAll(tasks);

                count = actionQueue.Count;
                tasks.Clear();
                for (int i = 0; i < Math.Min(Environment.ProcessorCount, 32); i++)
                {
                    tasks.Add(Task.Run(async () =>
                    {
                        while (actionQueue.TryDequeue(out var taskFactory))
                        {
                            await taskFactory();
                        }
                    }
                    ));
                }

{'fraction_non_alphanumeric': 0.08247810940589398, 'fraction_numerical': 0.001694755672723849, 'mean_word_length': 1.578781257586793, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1371411', 'n_tokens_mistral': 2519, 'n_tokens_neox': 2375, 'n_words': 572}