<commit_before>﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Isam.Esent;
using Microsoft.Isam.Esent.Interop;
using Microsoft.Isam.Esent.Interop.Vista;
using MsgPack.Serialization;

namespace EsentQueue
{
    public class PersistentQueue<T> : IDisposable
    {
        private readonly static MessagePackSerializer<T> _serializer = MessagePackSerializer.Get<T>();

        private readonly QueueCursorCache _cursorCache;
        private readonly string _databaseName;
        private readonly Instance _instance;

        public PersistentQueue(string path, StartOption startOption)
        {
            _instance = new Instance("EsentQueue.PersistentQueue");
            _instance.Parameters.CreatePathIfNotExist = true;
            _instance.Parameters.CircularLog = true;
            _instance.Parameters.MaxVerPages = 256;
            _instance.Init();
            _databaseName = path;
            _cursorCache = new QueueCursorCache(_instance, _databaseName);

            if (!File.Exists(_databaseName) || startOption == StartOption.CreateNew)
            {
                CreateDatabase();
            }
            else
            {
                CheckDatabase(startOption);
            }
        }

        public int Count
        {
            get
            {
                var cursor = _cursorCache.GetCursor();
                try
                {
                    int count = 0;
                    Api.MoveBeforeFirst(cursor.Session, cursor.DataTable);
                    while (Api.TryMoveNext(cursor.Session, cursor.DataTable))
                    {
                        count++;
                    }

                    return count;
                }
                finally
                {
                    _cursorCache.Return(cursor);
                }
            }
        }

        public void Enqueue(T item)
        {
            var cursor = _cursorCache.GetCursor();
            try
            {
                using (var transaction = cursor.BeginTransaction())
                {
                    using (var update = cursor.CreateDataTableUpdate())
                    {
                        using (var colStream = new ColumnStream(cursor.Session, cursor.DataTable, cursor.SerializedObjectColumn))
                        {
                            _serializer.Pack(colStream, item);
                        }

                        update.Save();
                    }

                    transaction.Commit(CommitTransactionGrbit.LazyFlush);
                }
            }
            finally
            {
                _cursorCache.Return(cursor);
            }
        }

        public QueueItem<T> Dequeue()
        {
            QueueItem<T> item;
            if (!TryDequeue(out item))
            {
                throw new InvalidOperationException("No items in the queue.");
            }

            return item;
        }

        public bool TryDequeue(out QueueItem<T> item)
        {
            var cursor = _cursorCache.GetCursor();
            try
            {
                while (Api.TryMoveFirst(cursor.Session, cursor.DataTable))
                {
                    try
                    {
                        int attempts = 1 + (Api.RetrieveColumnAsInt32(cursor.Session, cursor.DataTable, cursor.AttemptCountColumn) ?? 0);
                        using (var colStream = new ColumnStream(cursor.Session, cursor.DataTable, cursor.SerializedObjectColumn))
                        {
                            T obj = _serializer.Unpack(colStream);
                            item = new QueueItem<T>(obj, attempts);
                        }

                        using (var transaction = cursor.BeginTransaction())
                        {
                            Api.JetDelete(cursor.Session, cursor.DataTable);
                            transaction.Commit(CommitTransactionGrbit.LazyFlush);
                        }
                    }
                    catch (EsentException e) when
                        (e is EsentWriteConflictException || e is EsentRecordDeletedException)
                    {
                        continue;
                    }

                    return true;
                }

                item = default(QueueItem<T>);
                return false;
            }
            finally
            {
                _cursorCache.Return(cursor);
            }
        }

        public QueueItem<T> Peek()
        {
            QueueItem<T> item;
            if (!TryPeek(out item))
            {
                throw new InvalidOperationException("No items in the queue.");
            }

            return item;
        }

        public bool TryPeek(out QueueItem<T> item)
        {
            var cursor = _cursorCache.GetCursor();
            try
            {
                while (Api.TryMoveFirst(cursor.Session, cursor.DataTable))
                {
                    try
                    {
                        int attempts = 1 + Api.EscrowUpdate(cursor.Session, cursor.DataTable, cursor.AttemptCountColumn, 1);
                        using (var colStream = new ColumnStream(cursor.Session, cursor.DataTable, cursor.SerializedObjectColumn))
                        {
                            T obj = _serializer.Unpack(colStream);
                            item = new QueueItem<T>(obj, attempts);
                        }
                        
                        return true;
                    }
                    catch (EsentException e) when
                        (e is EsentWriteConflictException || e is EsentRecordDeletedException)
                    {
                        continue;
                    }
                }

                item = default(QueueItem<T>);
                return false;
            }
            finally
            {
                _cursorCache.Return(cursor);
            }
        }

        public void Dispose()
        {<commit_msg>try removing write conflict handling<commit_after>﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Isam.Esent;
using Microsoft.Isam.Esent.Interop;
using Microsoft.Isam.Esent.Interop.Vista;
using MsgPack.Serialization;

namespace EsentQueue
{
    public class PersistentQueue<T> : IDisposable
    {
        private readonly static MessagePackSerializer<T> _serializer = MessagePackSerializer.Get<T>();

        private readonly QueueCursorCache _cursorCache;
        private readonly string _databaseName;
        private readonly Instance _instance;

        public PersistentQueue(string path, StartOption startOption)
        {
            _instance = new Instance("EsentQueue.PersistentQueue");
            _instance.Parameters.CreatePathIfNotExist = true;
            _instance.Parameters.CircularLog = true;
            _instance.Parameters.MaxVerPages = 256;
            _instance.Init();
            _databaseName = path;
            _cursorCache = new QueueCursorCache(_instance, _databaseName);

            if (!File.Exists(_databaseName) || startOption == StartOption.CreateNew)
            {
                CreateDatabase();
            }
            else
            {
                CheckDatabase(startOption);
            }
        }

        public int Count
        {
            get
            {
                var cursor = _cursorCache.GetCursor();
                try
                {
                    int count = 0;
                    Api.MoveBeforeFirst(cursor.Session, cursor.DataTable);
                    while (Api.TryMoveNext(cursor.Session, cursor.DataTable))
                    {
                        count++;
                    }

                    return count;
                }
                finally
                {
                    _cursorCache.Return(cursor);
                }
            }
        }

        public void Enqueue(T item)
        {
            var cursor = _cursorCache.GetCursor();
            try
            {
                using (var transaction = cursor.BeginTransaction())
                {
                    using (var update = cursor.CreateDataTableUpdate())
                    {
                        using (var colStream = new ColumnStream(cursor.Session, cursor.DataTable, cursor.SerializedObjectColumn))
                        {
                            _serializer.Pack(colStream, item);
                        }

                        update.Save();
                    }

                    transaction.Commit(CommitTransactionGrbit.LazyFlush);
                }
            }
            finally
            {
                _cursorCache.Return(cursor);
            }
        }

        public QueueItem<T> Dequeue()
        {
            QueueItem<T> item;
            if (!TryDequeue(out item))
            {
                throw new InvalidOperationException("No items in the queue.");
            }

            return item;
        }

        public bool TryDequeue(out QueueItem<T> item)
        {
            var cursor = _cursorCache.GetCursor();
            try
            {
                using (var transaction = cursor.BeginTransaction())
                {
                    Api.JetSetCurrentIndex(cursor.Session, cursor.DataTable, null);
                    Api.MakeKey(cursor.Session, cursor.DataTable, 0L, MakeKeyGrbit.NewKey);

                    if (Api.TrySeek(cursor.Session, cursor.DataTable, SeekGrbit.SeekGE))
                    {
                        int attempts = 1 + (Api.RetrieveColumnAsInt32(cursor.Session, cursor.DataTable, cursor.AttemptCountColumn) ?? 0);
                        using (var colStream = new ColumnStream(cursor.Session, cursor.DataTable, cursor.SerializedObjectColumn))
                        {
                            T obj = _serializer.Unpack(colStream);
                            item = new QueueItem<T>(obj, attempts);
                        }

                        Api.JetDelete(cursor.Session, cursor.DataTable);
                        transaction.Commit(CommitTransactionGrbit.LazyFlush);

                        return true;
                    }
                }

                item = default(QueueItem<T>);
                return false;
            }
            finally
            {
                _cursorCache.Return(cursor);
            }
        }

        public QueueItem<T> Peek()
        {
            QueueItem<T> item;
            if (!TryPeek(out item))
            {
                throw new InvalidOperationException("No items in the queue.");
            }

            return item;
        }

        public bool TryPeek(out QueueItem<T> item)
        {
            var cursor = _cursorCache.GetCursor();
            try
            {
                using (var transaction = cursor.BeginTransaction())
                {
                    if (Api.TryMoveFirst(cursor.Session, cursor.DataTable))
                    {
                        int attempts = 1 + Api.EscrowUpdate(cursor.Session, cursor.DataTable, cursor.AttemptCountColumn, 1);
                        using (var colStream = new ColumnStream(cursor.Session, cursor.DataTable, cursor.SerializedObjectColumn))
                        {
                            T obj = _serializer.Unpack(colStream);
                            item = new QueueItem<T>(obj, attempts);
                        }

                        return true;
                    }
                }

                item = default(QueueItem<T>);
                return false;
            }
            finally
            {
                _cursorCache.Return(cursor);
            }
        }

        public void Dispose()
        {
{'fraction_non_alphanumeric': 0.07515376190074985, 'fraction_numerical': 0.0017693150223270706, 'mean_word_length': 1.1503623188405796, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4413376', 'n_tokens_mistral': 2835, 'n_tokens_neox': 2640, 'n_words': 536}