<commit_before>using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;

namespace Lokad.Cqrs.TapeStorage
{
    /// <summary>
    /// Simple in-memory thread-safe cache
    /// </summary>
    public sealed class LockingInMemoryCache
    {
        readonly ReaderWriterLockSlim _thread = new ReaderWriterLockSlim();
        ConcurrentDictionary<string, DataWithVersion[]> _cacheByKey = new ConcurrentDictionary<string, DataWithVersion[]>();
        DataWithKey[] _cacheFull = new DataWithKey[0];

        public void ReloadEverything(IEnumerable<StorageFrameDecoded> sfd)
        {
            _thread.EnterWriteLock();
            try
            {
                _cacheFull = new DataWithKey[0];

                // [abdullin]: known performance problem identified by <NAME>
                // creating new immutable array on each line will kill performance
                // We need to at least do some batching here

                var cacheFullBuilder = new List<DataWithKey>();
                var streamPointerBuilder = new Dictionary<string, List<DataWithVersion>>();

                long storeVersion = 0;
                foreach (var record in sfd)
                {
                    storeVersion += 1;
                    cacheFullBuilder.Add(new DataWithKey(record.Name, record.Bytes, record.Stamp, storeVersion));
                    List<DataWithVersion> list;
                    if (!streamPointerBuilder.TryGetValue(record.Name, out list))
                    {
                        streamPointerBuilder.Add(record.Name, list = new List<DataWithVersion>());
                    }
                    list.Add(new DataWithVersion(record.Stamp, record.Bytes, storeVersion));
                }

                _cacheFull = cacheFullBuilder.ToArray();
                _cacheByKey = new ConcurrentDictionary<string, DataWithVersion[]>(streamPointerBuilder.Select(p => new KeyValuePair<string, DataWithVersion[]>(p.Key, p.Value.ToArray())));
                _storeVersion = storeVersion;
            }
            finally
            {
                _thread.ExitWriteLock();
            }
        }

        static T[] ImmutableAdd<T>(T[] source, T item)
        {
            var copy = new T[source.Length + 1];

            Array.Copy(source, copy, source.Length);
            copy[source.Length] = item;


            return copy;
        }

        long _storeVersion;

        public long StoreVersion { get { return _storeVersion; } }

        public delegate void OnCommit(long streamVersion, long storeVersion);

        public void ConcurrentAppend(string streamName, byte[] data, OnCommit commit, long expectedStreamVersion = -1)
        {
            _thread.EnterWriteLock();

            try
            {
                var list = _cacheByKey.GetOrAdd(streamName, s => new DataWithVersion[0]);
                if (expectedStreamVersion >= 0)
                {
                    if (list.Length != expectedStreamVersion)
                        throw new AppendOnlyStoreConcurrencyException(expectedStreamVersion, list.Length, streamName);
                }
                long newStreamVersion = list.Length + 1;
                long newStoreVersion = _storeVersion + 1;

                commit(expectedStreamVersion, newStoreVersion);

                // update in-memory cache only after real commit completed

                var record = new DataWithVersion(newStreamVersion, data, newStoreVersion);
                _cacheFull = ImmutableAdd(_cacheFull, new DataWithKey(streamName, data, newStreamVersion, newStoreVersion));
                _cacheByKey.AddOrUpdate(streamName, s => new[] { record }, (s, records) => ImmutableAdd(records, record));
                _storeVersion = newStoreVersion;
            }
            finally
            {
                _thread.ExitWriteLock();
            }

        }

        public IEnumerable<DataWithVersion> ReadRecords(string streamName, long afterVersion, int maxCount)
        {
            if (afterVersion < 0)
                throw new ArgumentOutOfRangeException("afterVersion", "Must be zero or greater.");

            if (maxCount <= 0)
                throw new ArgumentOutOfRangeException("maxCount", "Must be more than zero.");

            // no lock is needed.
            DataWithVersion[] list;<commit_msg>Fix version locking in cache<commit_after>using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;

namespace Lokad.Cqrs.TapeStorage
{
    /// <summary>
    /// Simple in-memory thread-safe cache
    /// </summary>
    public sealed class LockingInMemoryCache
    {
        readonly ReaderWriterLockSlim _thread = new ReaderWriterLockSlim();
        ConcurrentDictionary<string, DataWithVersion[]> _cacheByKey = new ConcurrentDictionary<string, DataWithVersion[]>();
        DataWithKey[] _cacheFull = new DataWithKey[0];

        public void ReloadEverything(IEnumerable<StorageFrameDecoded> sfd)
        {
            _thread.EnterWriteLock();
            try
            {
                _cacheFull = new DataWithKey[0];

                // [abdullin]: known performance problem identified by <NAME>
                // creating new immutable array on each line will kill performance
                // We need to at least do some batching here

                var cacheFullBuilder = new List<DataWithKey>();
                var streamPointerBuilder = new Dictionary<string, List<DataWithVersion>>();

                long storeVersion = 0;
                foreach (var record in sfd)
                {
                    storeVersion += 1;
                    cacheFullBuilder.Add(new DataWithKey(record.Name, record.Bytes, record.Stamp, storeVersion));
                    List<DataWithVersion> list;
                    if (!streamPointerBuilder.TryGetValue(record.Name, out list))
                    {
                        streamPointerBuilder.Add(record.Name, list = new List<DataWithVersion>());
                    }
                    list.Add(new DataWithVersion(record.Stamp, record.Bytes, storeVersion));
                }

                _cacheFull = cacheFullBuilder.ToArray();
                _cacheByKey = new ConcurrentDictionary<string, DataWithVersion[]>(streamPointerBuilder.Select(p => new KeyValuePair<string, DataWithVersion[]>(p.Key, p.Value.ToArray())));
                _storeVersion = storeVersion;
            }
            finally
            {
                _thread.ExitWriteLock();
            }
        }

        static T[] ImmutableAdd<T>(T[] source, T item)
        {
            var copy = new T[source.Length + 1];

            Array.Copy(source, copy, source.Length);
            copy[source.Length] = item;


            return copy;
        }

        long _storeVersion;

        public long StoreVersion { get { return _storeVersion; } }

        public delegate void OnCommit(long streamVersion, long storeVersion);

        public void ConcurrentAppend(string streamName, byte[] data, OnCommit commit, long expectedStreamVersion = -1)
        {
            _thread.EnterWriteLock();

            try
            {
                var list = _cacheByKey.GetOrAdd(streamName, s => new DataWithVersion[0]);
                var actualStreamVersion = list.Length == 0 ? 0 : list.Max(d => d.StreamVersion);

                if (expectedStreamVersion >= 0)
                {
                    if (actualStreamVersion != expectedStreamVersion)
                        throw new AppendOnlyStoreConcurrencyException(expectedStreamVersion, actualStreamVersion, streamName);
                }
                long newStreamVersion = actualStreamVersion + 1;
                long newStoreVersion = _storeVersion + 1;

                commit(expectedStreamVersion, newStoreVersion);

                // update in-memory cache only after real commit completed

                var record = new DataWithVersion(newStreamVersion, data, newStoreVersion);
                _cacheFull = ImmutableAdd(_cacheFull, new DataWithKey(streamName, data, newStreamVersion, newStoreVersion));
                _cacheByKey.AddOrUpdate(streamName, s => new[] { record }, (s, records) => ImmutableAdd(records, record));
                _storeVersion = newStoreVersion;
            }
            finally
            {
                _thread.ExitWriteLock();
            }

        }

        public IEnumerable<DataWithVersion> ReadRecords(string streamName, long afterVersion, int maxCount)
        {
            if (afterVersion < 0)
                throw new ArgumentOutOfRangeException("afterVersion", "Must be zero or greater.");

            if (maxCount <= 0)
                throw new ArgumentOutOfRangeException("maxCount", "Must be more than zero.");

            // no lock is needed.
            DataWithVersion[] list;
{'fraction_non_alphanumeric': 0.08526870862784697, 'fraction_numerical': 0.0029170874004263436, 'mean_word_length': 1.9149771092217136, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 44, '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': '364647', 'n_tokens_mistral': 2224, 'n_tokens_neox': 2087, 'n_words': 542}