// using FASTER.core;
// 独自ストレージを使用する場合、FASTER.core.IDeviceを実装したインスタンスを代わりに生成する
// ファイルパスのみ必須。
IDevice logDevice = Devices.CreateLogDevice("[格納するファイルのパス]");

// using FASTER.core;
// LogDeviceのみ必須
var fls = new FasterLogSettings()
{
    LogDevice = logDevice
};
using(var fl = new FasterLog(fls))
{
    // 処理
    // flインスタンスはプロセス内で使い回すこと
}

// FasterLog fl;
// byte[] data;
// ReadOnlySpan<byte>でも可
// Enqueueの時点ではまだ永続化はされない
long recordAddress = fl.Enqueue(data);
// コミットデータの永続化
fl.Commit(true);

// FasterLog fl;
// fl.Scan([論理開始アドレス], [論理終端アドレス])という風にして指定する
using(FastLogScanIterator iter = fl.Scan(fl.CommittedBeginAddress, fl.CommittedUntilAddress))
{
    // enumeratorで走査
}

// FasterLog fl;
// fl.Scan([論理開始アドレス], [論理終端アドレス])という風にして指定する
using(FastLogScanIterator iter = fl.Scan(fl.CommittedBeginAddress, fl.CommittedUntilAddress))
{
    // dataの型はbyte[]
    // lenはデータ長(bytes)
    // 2019.11.18では二つだけだが、最新ソースでは更に long currentAddress も追加になる模様
    // https://github.com/microsoft/FASTER/commit/bf657635374873958d96b31db1299b58ef9a17b1
    await foreach(var (data, len) in iter.GetAsyncEnumerable())
    {
        // データの参照
    }
}

// FasterLog fl;
using(FastLogScanIterator iter = fl.Scan(fl.CommittedBeginAddress, fl.CommittedUntilAddress))
{
    while(iter.NextAddress >= [終端アドレス])
    {
        // データを取り出すまでループと待機を行う
        // 引数は、データ本体、データ長、現在のアドレスの三つ
        while(!iter.GetNext(out var entry, out var length, out var currentAddress))
        {
            if(iter.NextAddress >= [終端アドレス])
            {
                break;
            }
            await iter.WaitAsync();
        }
        // データの参照
    }
}

// FasterLog fl;
long untilAddress = 0;
using(var iter = fl.Scan(fl.BeginAddress, [終端]))
{
    await foreach(var x in iter.GetAsyncEnumerable())
    {
        // 処理
        untilAddress = iter.NextAddress;
    }
}
fl.TruncateUntil(untilAddress);
// 最後にCommitすると変更が反映される
fl.Commit(true);

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Channels;
using FASTER.core;
using System.IO;
using System.Linq;

namespace fasterlabs
{
    static class FasterLogCommitTest
    {
        static long EnqueueData(long value, FasterLog fl)
        {
            Span<long> data = stackalloc long[1];

            if (fl.TryEnqueue(System.Runtime.InteropServices.MemoryMarshal.AsBytes(data), out var logicalAddress))
            {
                return logicalAddress;
            }
            else
            {
                return -1;
            }
        }
        public static async ValueTask DoTest(int TaskNum)
        {
            const long TotalCount = 100000;
            // ensure using cleared data 
            if (File.Exists("logcommittest.log.0"))
            {
                File.Delete("logcommittest.log.0");
            }
            if (File.Exists("logcommittest.log.commit"))
            {
                File.Delete("logcommittest.log.commit");
            }
            var log = Devices.CreateLogDevice("logcommittest.log");
            var channel = Channel.CreateUnbounded<long>();
            using (var fl = new FasterLog(new FasterLogSettings() { LogDevice = log }))
            {
                var sw = new System.Diagnostics.Stopwatch();
                sw.Start();
                using (var csrc = new CancellationTokenSource(1000 * 240))
                {
                    await Task.WhenAll(
                        Task.WhenAll(Enumerable.Range(0, TaskNum).Select(async idx =>
                        {
                            long logicalAddress = 0;
                            try
                            {
                                for (int i = 0; i < TotalCount / TaskNum; i++)
                                {
                                    logicalAddress = EnqueueData(i + idx * TotalCount, fl);
                                    await channel.Writer.WriteAsync(logicalAddress, csrc.Token).ConfigureAwait(false);
                                    await fl.WaitForCommitAsync(logicalAddress, csrc.Token).ConfigureAwait(false);
                                    // Console.WriteLine($"{idx}, {i}, {logicalAddress}");
                                }
                            }
                            catch (Exception e)
                            {
                                Console.WriteLine($"producer error({idx}, {logicalAddress}, {fl.CommittedUntilAddress}, {fl.TailAddress}): {e}");
                            }
                            // Console.WriteLine($"exit producer({idx}, {sw.Elapsed})");
                        })).ContinueWith(t => channel.Writer.Complete()),
                        Task.Run(async () =>
                        {
                            int commitCount = 0;
                            try
                            {
                                while (true)
                                {
                                    if (!await channel.Reader.WaitToReadAsync(csrc.Token).ConfigureAwait(false))
                                    {
                                        break;
                                    }
                                    while (channel.Reader.TryRead(out var untiladdr))
                                    {
                                        if (fl.CommittedUntilAddress <= untiladdr && fl.CommittedUntilAddress != fl.TailAddress)
                                        {
                                            fl.Commit(true);
                                            // await fl.CommitAsync(csrc.Token).ConfigureAwait(false);
                                            commitCount++;
                                        }
                                    }
                                }
                            }
                            catch (Exception e)
                            {
                                Console.WriteLine($"consumer error:{e}");
                            }
                            Console.WriteLine($"exit consumer({commitCount})");
                        }).ContinueWith(t =>
                        {
                            if(fl.CommittedUntilAddress != fl.TailAddress)
                            {
                                Console.WriteLine($"last commit");
                                fl.Commit(true);
                            }
                        })
                    ).ConfigureAwait(false);
                    sw.Stop();
                    Console.WriteLine($"Multi({TotalCount}, {TaskNum}): {sw.Elapsed}, iops = {(TotalCount * 1000) / sw.Elapsed.TotalMilliseconds}");
                }
            }
            log.Close();
        }
    }
}