<commit_before>﻿using System;
using System.Security.AccessControl;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;

namespace AppendLog
{
    /// <summary>
    /// A streaming file-based implementation of <see cref="IAppendLog"/>.
    /// </summary>
    public sealed class FileLog : IAppendLog
    {
        //FIXME: use 128 bit transaction ids, and store the "base" offset at the beginning of the db file.

        //FIXME: implement file rollover for compaction purposes, ie. can delete old entries by starting to
        //write to a different file than the one we're reading from, initialized with a new base offset. We
        //can then either copy the entries starting at the new base into the new file, or change the read
        //logic to also open the new file when it's done with the read file. The latter is preferable.

        //FIXME: I should place a db version number at the beginning so I can perform future upgrades if needed.
        //Also, this means the atomic update to the nextId doesn't take place at Pos=0, so no need to adjust it
        //before flush.
        string path;

        const long VERSION = 0x01;

        public FileLog(string path)
        {
            if (path == null) throw new ArgumentNullException("path");
            this.path = Path.GetFullPath(path);
            // check the file's version number if file exists, else write it out
            using (var fs = File.Open(path, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None))
            {
                var buf = new byte[sizeof(ulong)];
                if (buf.Length == 0)
                {
                    VERSION.WriteId(buf);
                    fs.Write(buf, 0, buf.Length);
                }
                else
                {
                    fs.Read(buf, 0, buf.Length);
                    var vers = buf.FillNextId();
                    if (vers != VERSION)
                    {
                        throw new NotSupportedException(string.Format("File log expects version {0}.{1}.{2} but found version {3}.{4}.{5}",
                            Major(VERSION), Minor(VERSION), Revision(VERSION), Major(vers), Minor(vers), Revision(vers)));
                    }
                }
            }
        }

        public TransactionId First
        {
            get { return new TransactionId { Id = sizeof(ulong) }; }
        }

        public IEventEnumerator Replay(TransactionId lastEvent)
        {
            return new EventEnumerator(this, lastEvent);
        }

        public Stream Append()
        {
            return new TransactedStream(this);
        }

        public void Dispose()
        {
            // should track outstanding write stream and dispose of it?
        }

        #region Internals
        static long Major(long x)    { return 0x1FFFFF & (x >> (64 - 21)); }
        static long Minor(long x)    { return 0x1FFFFF & (x >> (64 - 42)); }
        static long Revision(long x) { return 0x1FFFFF & x; }

        sealed class EventEnumerator : IEventEnumerator
        {
            FileLog log;
            FileStream file;
            TransactionId next;
            public TransactionId Transaction { get; internal set; }
            public Stream Stream { get; internal set; }

            ~EventEnumerator()
            {
                Dispose();
            }

            public EventEnumerator(FileLog slog, TransactionId lastEvent)
            {
                file = log.Open();
                log = slog;
                // skip header if default TransactionId
                next = lastEvent.Id == 0 ? log.First : lastEvent;
                file.Position = next.Id;
            }

            public async Task<bool> MoveNext()
            {
                if (file == null) throw new ObjectDisposedException("IEventEnumerator");
                var tmp = new byte[sizeof(long)];
                await file.ReadAsync(tmp, 0, sizeof(long));
                var end = tmp.FillNextId();
                if (next.Id >= end)
                    return false;<commit_msg>small cleanups<commit_after>﻿using System;
using System.Security.AccessControl;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;

namespace AppendLog
{
    /// <summary>
    /// A streaming file-based implementation of <see cref="IAppendLog"/>.
    /// </summary>
    public sealed class FileLog : IAppendLog
    {
        //FIXME: use 128 bit transaction ids, and store the "base" offset at the beginning of the db file.

        //FIXME: implement file rollover for compaction purposes, ie. can delete old entries by starting to
        //write to a different file than the one we're reading from, initialized with a new base offset. We
        //can then either copy the entries starting at the new base into the new file, or change the read
        //logic to also open the new file when it's done with the read file. The latter is preferable.
        string path;

        const long VERSION = 0x01;

        public FileLog(string path)
        {
            if (path == null) throw new ArgumentNullException("path");
            this.path = Path.GetFullPath(path);
            // check the file's version number if file exists, else write it out
            using (var fs = File.Open(path, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None))
            {
                var buf = new byte[sizeof(ulong)];
                if (buf.Length == 0)
                {
                    VERSION.WriteId(buf);
                    fs.Write(buf, 0, buf.Length);
                }
                else
                {
                    fs.Read(buf, 0, buf.Length);
                    var vers = buf.FillNextId();
                    if (vers != VERSION)
                    {
                        throw new NotSupportedException(string.Format("File log expects version {0}.{1}.{2} but found version {3}.{4}.{5}",
                            Major(VERSION), Minor(VERSION), Revision(VERSION), Major(vers), Minor(vers), Revision(vers)));
                    }
                }
            }
        }

        public TransactionId First
        {
            get { return new TransactionId { Id = sizeof(ulong) }; }
        }

        public IEventEnumerator Replay(TransactionId lastEvent)
        {
            return new EventEnumerator(this, lastEvent);
        }

        public Stream Append()
        {
            return new TransactedStream(this);
        }

        public void Dispose()
        {
            // should track outstanding write stream and dispose of it?
        }

        #region Internals
        static long Major(long x)    { return 0x1FFFFF & (x >> (64 - 21)); }
        static long Minor(long x)    { return 0x1FFFFF & (x >> (64 - 42)); }
        static long Revision(long x) { return 0x1FFFFF & x; }

        sealed class EventEnumerator : IEventEnumerator
        {
            FileLog log;
            FileStream file;
            TransactionId next;
            public TransactionId Transaction { get; internal set; }
            public Stream Stream { get; internal set; }

            ~EventEnumerator()
            {
                Dispose();
            }

            public EventEnumerator(FileLog slog, TransactionId lastEvent)
            {
                file = log.Open();
                log = slog;
                // skip version header if default TransactionId
                next = lastEvent.Id == 0 ? log.First : lastEvent;
                file.Position = next.Id;
            }

            public async Task<bool> MoveNext()
            {
                if (file == null) throw new ObjectDisposedException("IEventEnumerator");
                var tmp = new byte[sizeof(long)];
                await file.ReadAsync(tmp, 0, sizeof(long));
                var end = tmp.FillNextId();
                if (next.Id >= end)
                    return false;
{'fraction_non_alphanumeric': 0.08807134894091416, 'fraction_numerical': 0.007803790412486065, 'mean_word_length': 1.6036762334730732, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 21, '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': '4473918', 'n_tokens_mistral': 2102, 'n_tokens_neox': 1988, 'n_words': 696}