<commit_before>﻿using Sir.Core;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.IO.MemoryMappedFiles;
using System.Threading;
using System.Threading.Tasks;

namespace Sir.Store
{
    /// <summary>
    /// Dispatcher of sessions.
    /// </summary>
    public class SessionFactory : IDisposable, ILogger
    {
        private readonly ITokenizer _tokenizer;
        private readonly IConfigurationProvider _config;
        private readonly ConcurrentDictionary<ulong, ConcurrentDictionary<ulong, long>> _keys;
        private readonly ConcurrentDictionary<ulong, ConcurrentDictionary<long, NodeReader>> _indexReaders;
        private readonly ConcurrentDictionary<string, object> _collectionLocks;
        private readonly Semaphore _writeSync;
        private readonly ConcurrentBag<MemoryMappedFile> _mmfs;

        public string Dir { get; }
        public IConfigurationProvider Config { get { return _config; } }

        public SessionFactory(string dir, ITokenizer tokenizer, IConfigurationProvider config)
        {
            Dir = dir;
            _keys = LoadKeys();
            _tokenizer = tokenizer;
            _config = config;
            _indexReaders = new ConcurrentDictionary<ulong, ConcurrentDictionary<long, NodeReader>>();
            _collectionLocks = new ConcurrentDictionary<string, object>();
            _mmfs = new ConcurrentBag<MemoryMappedFile>();

            bool createdSystemWideSem;

            _writeSync = new Semaphore(1, 2, "Sir.Store.SessionFactory", out createdSystemWideSem);

            if (!createdSystemWideSem)
            {
                _writeSync.Dispose();
                _writeSync = Semaphore.OpenExisting("Sir.Store.SessionFactory");
            }
        }

        public async Task Write(Job job)
        {
            _writeSync.WaitOne();

            var timer = Stopwatch.StartNew();

            var colId = job.Collection.ToHash();

            using (var writeSession = CreateWriteSession(job.Collection, colId))
            using (var indexSession = CreateIndexSession(job.Collection, colId))
            {
                foreach (var doc in job.Documents)
                {
                    await writeSession.Write(doc);
                    indexSession.Index(doc);
                }
            }
<commit_msg>prohibit writing while in readonly mode<commit_after>﻿using Sir.Core;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.IO.MemoryMappedFiles;
using System.Threading;
using System.Threading.Tasks;

namespace Sir.Store
{
    /// <summary>
    /// Dispatcher of sessions.
    /// </summary>
    public class SessionFactory : IDisposable, ILogger
    {
        private readonly ITokenizer _tokenizer;
        private readonly IConfigurationProvider _config;
        private readonly ConcurrentDictionary<ulong, ConcurrentDictionary<ulong, long>> _keys;
        private readonly ConcurrentDictionary<ulong, ConcurrentDictionary<long, NodeReader>> _indexReaders;
        private readonly ConcurrentDictionary<string, object> _collectionLocks;
        private readonly Semaphore _writeSync;
        private readonly ConcurrentBag<MemoryMappedFile> _mmfs;

        public string Dir { get; }
        public IConfigurationProvider Config { get { return _config; } }

        public SessionFactory(string dir, ITokenizer tokenizer, IConfigurationProvider config)
        {
            Dir = dir;
            _keys = LoadKeys();
            _tokenizer = tokenizer;
            _config = config;
            _indexReaders = new ConcurrentDictionary<ulong, ConcurrentDictionary<long, NodeReader>>();
            _collectionLocks = new ConcurrentDictionary<string, object>();
            _mmfs = new ConcurrentBag<MemoryMappedFile>();

            bool createdSystemWideSem;

            _writeSync = new Semaphore(1, 2, "Sir.Store.SessionFactory", out createdSystemWideSem);

            if (!createdSystemWideSem)
            {
                _writeSync.Dispose();
                _writeSync = Semaphore.OpenExisting("Sir.Store.SessionFactory");
            }
        }

        public async Task Write(Job job)
        {
            if (_mmfs.Count > 0)
            {
                throw new InvalidOperationException("wrinting while in readonly mode");
            }

            _writeSync.WaitOne();

            var timer = Stopwatch.StartNew();

            var colId = job.Collection.ToHash();

            using (var writeSession = CreateWriteSession(job.Collection, colId))
            using (var indexSession = CreateIndexSession(job.Collection, colId))
            {
                foreach (var doc in job.Documents)
                {
                    await writeSession.Write(doc);
                    indexSession.Index(doc);
                }
            }

{'fraction_non_alphanumeric': 0.08039692183070068, 'fraction_numerical': 0.001012555690562981, 'mean_word_length': 2.3214525891055815, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 28, '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': '2479209', 'n_tokens_mistral': 1244, 'n_tokens_neox': 1209, 'n_words': 303}