using System;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
using System.Numerics;
using System.Text;
using FASTER.core;
namespace PerfectNumbers.Core.Persistence;
///
/// Stores BitTree frontier batches using Microsoft.FASTER.Core.
/// Each batch is a byte[] payload containing a sequence of serialized states.
/// Maintains a simple head/tail cursor stored inside the KV store.
///
public sealed class BitTreeFrontierRepository : IDisposable
{
private readonly FasterKV _store;
private readonly IDevice _logDevice;
private readonly IDevice _objectLogDevice;
private long _headIndex;
private long _tailIndex;
private const string HeadKey = "__head__";
private const string TailKey = "__tail__";
public string DatabasePath { get; }
public static BitTreeFrontierRepository Open(string directory, string repositoryFileName = "bittree.frontier.faster")
{
Directory.CreateDirectory(directory);
string dbPath = Path.Combine(directory, repositoryFileName);
return new BitTreeFrontierRepository(dbPath);
}
private BitTreeFrontierRepository(string path)
{
DatabasePath = path;
_logDevice = Devices.CreateLogDevice(path + ".log");
_objectLogDevice = Devices.CreateLogDevice(path + ".obj.log");
_store = new FasterKV(
1L << 20,
new LogSettings
{
LogDevice = _logDevice,
ObjectLogDevice = _objectLogDevice,
});
LoadCursors();
}
public void AppendBatch(ReadOnlySpan batch)
{
using var session = _store.For(new SimpleFunctions()).NewSession>();
string key = BuildChunkKey(_tailIndex);
session.Upsert(key, batch.ToArray());
_tailIndex++;
PersistCursors(session);
session.CompletePending(true);
_store.Log.FlushAndEvict(true);
}
public bool TryDequeueBatch(int maxCount, out List batches)
{
batches = new List();
using var session = _store.For(new SimpleFunctions()).NewSession>();
while (_headIndex < _tailIndex && batches.Count < maxCount)
{
string key = BuildChunkKey(_headIndex);
if (session.Read(key, out byte[]? stored).Found && stored is not null)
{
batches.Add(stored);
session.Delete(key);
}
_headIndex++;
}
if (batches.Count > 0)
{
PersistCursors(session);
session.CompletePending(true);
_store.Log.FlushAndEvict(true);
return true;
}
session.CompletePending(true);
_store.Log.FlushAndEvict(true);
return false;
}
public void Dispose()
{
_store.Dispose();
_logDevice.Dispose();
_objectLogDevice.Dispose();
}
private void LoadCursors()
{
using var session = _store.For(new SimpleFunctions()).NewSession>();
_headIndex = ReadCursor(session, HeadKey);
_tailIndex = ReadCursor(session, TailKey);
session.CompletePending(true);
}
private static long ReadCursor(ClientSession> session, string key)
{
if (session.Read(key, out byte[]? stored).Found && stored is not null)
{
string text = Encoding.UTF8.GetString(stored);
if (long.TryParse(text, NumberStyles.Integer, CultureInfo.InvariantCulture, out long value))
{
return value;
}
}
return 0;
}
private void PersistCursors(ClientSession> session)
{
session.Upsert(HeadKey, Encoding.UTF8.GetBytes(_headIndex.ToString(CultureInfo.InvariantCulture)));
session.Upsert(TailKey, Encoding.UTF8.GetBytes(_tailIndex.ToString(CultureInfo.InvariantCulture)));
}
private static string BuildChunkKey(long index) => $"chunk-{index}";
}