[C#] Trying out Faster as a highly concurrent object cache

[JeremyTCD]

Hi! I'm trying out faster as a highly concurrent object-cache that spans memory and disk. It's working great!

It feels like there are a lot of knobs to turn and nuances to grasp though, so I'm worried I might've done things wrongly. Would greatly appreciate comments on my wrapper class below. Some use case details:

• I'm not using persistency features
• I'm only doing point reads and writes (no deletes, no batch reads/writes)
• Some objects are cached almost entirely on disk, others will have more in-memory space
• Minimizing disk space usage is important for me

Faster wrapper class:

using FASTER.core;
using MessagePack;
using Microsoft.Extensions.Options;
using System.Collections.Concurrent;
using System.IO;
using System.Threading.Tasks;

namespace <my namespace>
{
    public class ObjectCacheService<T> : IObjectCacheService<T> where T : class
    {
        // Use MessagePack to serialize and compress objects (see Serializer below)
        private static readonly MessagePackSerializerOptions _messagePackOptions = MessagePackSerializerOptions.
            Standard.
            WithCompression(MessagePackCompression.Lz4BlockArray);
        // Doesn't seem to have state, should be reusable
        private static readonly SimpleFunctions<int, T> _simpleFunctions = new();
        // Sessions pool
        private static readonly ConcurrentBag<ClientSession<int, T, T, T, Empty, SimpleFunctions<int, T>>> _sessions = new();
        private readonly FasterKV<int, T> _store;

        public ObjectCacheService(IOptions<ObjectCacheServiceOptions> objectCacheServiceOptionsAccessor) // Asp.Net dependency injection options system
        {
            var serializerSettings = new SerializerSettings<int, T>
            {
                valueSerializer = () => new Serializer()
            };

            string path = objectCacheServiceOptionsAccessor.Value.CachePath;
            string typeName = typeof(T).Name;
            IDevice log = Devices.CreateLogDevice(Path.Combine(path, $"{typeName}.log"));
            IDevice objlog = Devices.CreateLogDevice(Path.Combine(path, $"{typeName}.obj.log"));

            var logSettings = new LogSettings
            {
                LogDevice = log,
                ObjectLogDevice = objlog,
                // TODO configurable per cache
                PageSizeBits = 12, // 4K pages
                MemorySizeBits = 13 // 2 pages
            };

            _store = new(size: 1L << 20,
                logSettings: logSettings,
                serializerSettings: serializerSettings);
        }

        public void Upsert(int key, T obj)
        {
            ClientSession<int, T, T, T, Empty, SimpleFunctions<int, T>> session = GetSession();

            session.Upsert(key, obj);

            _sessions.Add(session);
        }

        public async ValueTask<(Status, T)> ReadAsync(int key)
        {
            ClientSession<int, T, T, T, Empty, SimpleFunctions<int, T>> session = GetSession();

            (Status, T) result = (await session.ReadAsync(key).ConfigureAwait(false)).Complete();

            _sessions.Add(session);

            return result;
        }

        private ClientSession<int, T, T, T, Empty, SimpleFunctions<int, T>> GetSession()
        {
            if (_sessions.TryTake(out ClientSession<int, T, T, T, Empty, SimpleFunctions<int, T>>? result))
            {
                return result;
            }

            // Thread-safe?
            return _store.
                For(_simpleFunctions). // ClientSessionbuilder reusable?
                NewSession<SimpleFunctions<int, T>>();
        }

        private class Serializer : BinaryObjectSerializer<T>
        {
            public override void Deserialize(out T obj)
            {
                obj = MessagePackSerializer.Deserialize<T>(reader.BaseStream, _messagePackOptions);
            }

            public override void Serialize(ref T obj)
            {
                MessagePackSerializer.Serialize(writer.BaseStream, obj, _messagePackOptions);
            }
        }
    }
}

Questions:

• In the async store example WaitForCommitAsync is called after Upsert:

  FASTER/cs/samples/StoreAsyncApi/Program.cs

  Lines 85 to 86 in b1d7937

  	session.Upsert(ref key, ref value, context);
  	await session.WaitForCommitAsync();

  Is this necessary if I'm not concerned about persisting data on disk?

• Am I right to say callback functions are primarily for when you want to read/write just part of an object (partial updates) or RMW? If I'm not doing either do I need to worry about IFunction's other members, ConcurrentReader, ConcurrentWriter?

• In my simple use-case, do I need to worry about log compaction and checkpoints?

• Is it safe to call session.Compact(compactUntil, shiftBeginAddress: true) while concurrently reading/writing using other session instances?

Edit:

• I think I initially misunderstood log compaction. My understanding now: the log contains entries for different versions of a record. So long as you're updating records, you'll benefit from compacting the log occasionally. There are two compaction options: "true compaction" removes entries for outdated record versions, e.g. if record x has versions v1 and v2 in the log, v1 is dropped. "Truncation" drops older entries. This may result in data loss, e.g. if both v1 and v2 of x exist in the set of entries dropped, the next read for x will result in Status.NotFound. I've added periodic session.Compact(compactUntil, shiftBeginAddress: true) calls to my wrapper.

[badrishc]

Is this necessary if I'm not concerned about persisting data on disk?

No, this is not needed if you are not concerned with persistence.

Am I right to say callback functions are primarily for when you want to read/write just part of an object (partial updates) or RMW

Generally, yes. Our default SimpleFunctions handle the common cases.

In my simple use-case, do I need to worry about checkpoints?

No

Is it safe to call session.Compact(compactUntil, shiftBeginAddress: true) while concurrently reading/writing using other session instances?

Yes, it should be thread-safe.

[badrishc]

Also, I see you are trying to pool sessions, which is reasonable if you don't want to associate sessions with caller instances/threads a priori.

It might be worth comparing performance with a ConcurrentQueue as an alternative data structure for this.

There might be contention on the single queue with large number of threads, so you could partition the queue by creating an array of say 8 queues, and choosing a random queue to dequeue from (enqueue back into the same queue).

The other option is to keep the session as a [ThreadStatic] (thread-local variable), while also keeping a global list of sessions for the purpose of disposing at the end. So you only allocate a session when the thread-local value is null. This will avoid accessing a concurrent data structure on the critical path.

[badrishc]

I think I initially misunderstood log compaction. My understanding now:

That is indeed an accurate summary!

[JeremyTCD]

Thank you for the detailed response! The session/per thread suggestion makes a lot of sense, I've incorporated it and further generalized my wrapper.

Unfortunately I've run into an intermittent IndexOutOfRangeException when using ThreadStatic. I've created a minimal repro (usage: clone, build in vs, ctrl + f5 run).

For quick reference, the wrapper:

using FASTER.core;
using MessagePack;
using System;
using System.Collections.Concurrent;
using System.IO;
using System.Threading.Tasks;

namespace Repro.Faster.IndexOutOfRangeException
{
    public class MixedStorageDictionary<TKey, TValue> : IMixedStorageDictionary<TKey, TValue>
    {
        // Serialization
        private static readonly MessagePackSerializerOptions _messagePackOptions = MessagePackSerializerOptions.
            Standard.
            WithCompression(MessagePackCompression.Lz4BlockArray);
        private static readonly SimpleFunctions<TKey, TValue> _simpleFunctions = new();

        // Sessions (1 per thread)
        [ThreadStatic]
        private static ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>>? _session;

        // Disposal
        private static readonly ConcurrentBag<ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>>> _sessions = new();
        private readonly FasterKV<TKey, TValue> _store;
        private bool _disposed;

        public MixedStorageDictionary(string? logFileDirectory = null,
            string? logFileName = null,
            int pageSizeBits = 12, // 4 KB pages
            int inMemorySpaceBits = 13, // 2 pages
            long indexNumBuckets = 1L << 20) // 64 MB index
        {
            var serializerSettings = new SerializerSettings<TKey, TValue>
            {
                valueSerializer = () => new Serializer()
            };

            logFileDirectory ??= Path.Combine(Path.GetTempPath(), "FasterLogs");
            logFileName ??= Guid.NewGuid().ToString();
            IDevice log = Devices.CreateLogDevice(Path.Combine(logFileDirectory, $"{logFileName}.log"));
            IDevice objlog = Devices.CreateLogDevice(Path.Combine(logFileDirectory, $"{logFileName}.obj.log"));

            var logSettings = new LogSettings
            {
                LogDevice = log,
                ObjectLogDevice = objlog,
                PageSizeBits = pageSizeBits,
                MemorySizeBits = inMemorySpaceBits
            };

            _store = new(size: indexNumBuckets,
                logSettings: logSettings,
                serializerSettings: serializerSettings);
        }

        public void Upsert(TKey key, TValue obj)
        {
            (_session ?? CreateSession()).Upsert(key, obj);
        }

        public async ValueTask<(Status, TValue)> ReadAsync(TKey key)
        {
            return (await (_session ?? CreateSession()).ReadAsync(key).ConfigureAwait(false)).Complete();
        }

        private ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>> CreateSession()
        {
            _session = _store.For(_simpleFunctions).NewSession<SimpleFunctions<TKey, TValue>>();
            _sessions.Add(_session);

            return _session;
        } 

        private class Serializer : BinaryObjectSerializer<TValue>
        {
            public override void Deserialize(out TValue obj)
            {
                obj = MessagePackSerializer.Deserialize<TValue>(reader.BaseStream, _messagePackOptions);
            }

            public override void Serialize(ref TValue obj)
            {
                MessagePackSerializer.Serialize(writer.BaseStream, obj, _messagePackOptions);
            }
        }

        protected virtual void Dispose(bool disposing)
        {
            if (!_disposed)
            {
                if (disposing)
                {
                    _store.Dispose();

                    foreach (ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>> session in _sessions)
                    {
                        session.Dispose();
                    }
                }

                _disposed = true;
            }
        }

        public void Dispose()
        {
            Dispose(disposing: true);
            GC.SuppressFinalize(this);
        }
    }
}

Repro program:

public static async Task Main()
{
    var mixedStorageDictionary = new MixedStorageDictionary<int, string>();
    int numRecords = 1000;
    string dummyRecordValue = "dummyString";

    // Concurrent upserts
    var upsertTasks = new Task[numRecords];
    for (int i = 0; i < numRecords; i++)
    {
        int key = i;
        upsertTasks[i] = Task.Run(() => mixedStorageDictionary.Upsert(key, dummyRecordValue));
    }
    await Task.WhenAll(upsertTasks).ConfigureAwait(false);

    // Concurrent reads
    var readTasks = new Task<(Status, string)>[numRecords];
    for (int i = 0; i < numRecords; i++)
    {
        readTasks[i] = mixedStorageDictionary.ReadAsync(i).AsTask();
    }
    await Task.WhenAll(readTasks).ConfigureAwait(false);

    // Assert
    for (int i = 0; i < numRecords; i++)
    {
        (Status status, string result) = readTasks[i].Result;
        Assert.Equal(Status.OK, status);
        Assert.Equal(dummyRecordValue, result);
    }

    Console.WriteLine("Success!");
}

Stack trace:

Unhandled exception. System.IndexOutOfRangeException: Index was outside the bounds of the array.
   at System.Collections.Generic.Dictionary`2.TryInsert(TKey key, TValue value, InsertionBehavior behavior)
   at FASTER.core.FasterKV`2.HandleOperationStatus[Input,Output,Context,FasterSession](FasterExecutionContext`3 opCtx, FasterExecutionContext`3 currentCtx, PendingContext`3& pendingContext, FasterSession fasterSession, OperationStatus status, Boolean asyncOp, AsyncIOContext`2& request)
   at Repro.Faster.IndexOutOfRangeException.MixedStorageDictionary`2.ReadAsync(TKey key) in C:\Projects\Repro.Faster.IndexOutOfRangeException\Repro.Faster.IndexOutOfRangeException\MixedStorageDictionary.cs:line 63
   at Repro.Faster.IndexOutOfRangeException.Program.Main() in C:\Projects\Repro.Faster.IndexOutOfRangeException\Repro.Faster.IndexOutOfRangeException\Program.cs:line 31
   at Repro.Faster.IndexOutOfRangeException.Program.<Main>()

The exception occurs at random. Increasing numRecords to 10000 almost guarantees that it'll be thrown on my machine. Upserts work fine, the exception is only thrown by ReadAsync calls.

Edit: The earlier, naïve "ConcurrentBag for sessions" design handled 1,000,000 records without throwing so it looks like this may be a thread affinity issue.

[badrishc]

At a quick glance, I think the thread local idea doesn't work because the continuation of async will continue to use the same session (on the continuation thread). At the same time a different task assigned to the original thread will start using the session from the thread local.

This will break the invariant of "one session is used at most once at the same time by some thread" ("mono-threadedness").

[badrishc]

Do you expect most Reads to happen synchronously, or will it go to disk most of the time?

Nice, clean sample by the way, would be nice to convert it eventually into an official sample to show best way to manage sessions internally to a non-session-based KV API.

[badrishc]

Here is a version that combines the idea of thread-local with shared pool used only when ops go async (I found queue to be slightly faster than bag in my experiments, for the shared pool):

    public class MixedStorageDictionary<TKey, TValue> : IMixedStorageDictionary<TKey, TValue>
    {
        // Serialization
        private static readonly MessagePackSerializerOptions _messagePackOptions = MessagePackSerializerOptions.
            Standard.
            WithCompression(MessagePackCompression.Lz4BlockArray);
        private static readonly SimpleFunctions<TKey, TValue> _simpleFunctions = new();

        // Sessions (1 per thread)
        [ThreadStatic]
        private static ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>>? _session;

        // Shared pool
        private static readonly ConcurrentQueue<ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>>> _sessions = new();

        // Track all created sessions for dispose
        private static readonly ConcurrentBag<ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>>> _allSessions = new();

        private readonly FasterKV<TKey, TValue> _store;
        private bool _disposed;

        public MixedStorageDictionary(string? logFileDirectory = null,
            string? logFileName = null,
            int pageSizeBits = 12, // 4 KB pages
            int inMemorySpaceBits = 13, // 2 pages
            long indexNumBuckets = 1L << 20) // 64 MB index
        {
            var serializerSettings = new SerializerSettings<TKey, TValue>
            {
                valueSerializer = () => new Serializer()
            };

            logFileDirectory ??= Path.Combine(Path.GetTempPath(), "FasterLogs");
            logFileName ??= Guid.NewGuid().ToString();
            IDevice log = Devices.CreateLogDevice(Path.Combine(logFileDirectory, $"{logFileName}.log"));
            IDevice objlog = Devices.CreateLogDevice(Path.Combine(logFileDirectory, $"{logFileName}.obj.log"));

            var logSettings = new LogSettings
            {
                LogDevice = log,
                ObjectLogDevice = objlog,
                PageSizeBits = pageSizeBits,
                MemorySizeBits = inMemorySpaceBits
            };

            _store = new(size: indexNumBuckets,
                logSettings: logSettings,
                serializerSettings: serializerSettings);
        }

        public void Upsert(TKey key, TValue obj)
        {
            (_session ?? GetSession()).Upsert(key, obj);
        }

        public async ValueTask<(Status, TValue)> ReadAsync(TKey key)
        {
            var session = _session ?? GetSession();
            var t = session.ReadAsync(key);

            // Retain thread-local in sync path
            if (t.IsCompleted)
                return t.Result.Complete();

            // Going async - remove session from thread-local
            _session = null;
            var result = (await t.ConfigureAwait(false)).Complete();

            // Return session to shared pool on async thread
            _sessions.Enqueue(session);
            return result;
        }

        private ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>> GetSession()
        {
            if (_sessions.TryDequeue(out ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>>? result))
            {
                _session = result;
                return result;
            }
            var session = _store.For(_simpleFunctions).NewSession<SimpleFunctions<TKey, TValue>>();
            _allSessions.Add(session);
            _session = session;
            return session;
        }


        private class Serializer : BinaryObjectSerializer<TValue>
        {
            public override void Deserialize(out TValue obj)
            {
                obj = MessagePackSerializer.Deserialize<TValue>(reader.BaseStream, _messagePackOptions);
            }

            public override void Serialize(ref TValue obj)
            {
                MessagePackSerializer.Serialize(writer.BaseStream, obj, _messagePackOptions);
            }
        }

        protected virtual void Dispose(bool disposing)
        {
            if (!_disposed)
            {
                if (disposing)
                {
                    _store.Dispose();

                    foreach (ClientSession<TKey, TValue, TValue, TValue, Empty, SimpleFunctions<TKey, TValue>> session in _allSessions)
                    {
                        session.Dispose();
                    }
                }

                _disposed = true;
            }
        }

        public void Dispose()
        {
            Dispose(disposing: true);
            GC.SuppressFinalize(this);
        }
    }

[badrishc]

Note: above will only help if reads tend to return synchronously (data is in memory) most of the time. Upserts could also benefit as they return synchronously.

Otherwise, your earlier solution is fine, and even preferable for its cleanliness.

[badrishc]

The other thing with your example is the read concurrency is limited as you are not issuing requests in parallel (only the disk I/O completion gets parallelized). You can get better throughput and disk utilization by parallelizing the issuing of requests, as the following example shows (vary issueParallel to increase read-issue parallelism):

    public class Program
    {
        public static void Main()
        {
            var mixedStorageDictionary = new MixedStorageDictionary<int, int>(indexNumBuckets: 1L << 22);
            int numRecords = 20_000_000;

            for (int i = 0; i < numRecords; i++)
                mixedStorageDictionary.Upsert(i, 45);

            Console.WriteLine("upsert done");

            int issueParallel = 4; // vary this to increase read issue parallelism
            Task[] tasks = new Task[issueParallel];
            Random[] r = new Random[issueParallel];
            for (int i = 0; i < issueParallel; i++)
                r[i] = new Random(i);

            int cnt = 0;
            Stopwatch sw = new Stopwatch();
            sw.Start();
            while (true)
            {
                for (int i = 0; i < issueParallel; i++)
                    tasks[i] = ReadIssuer(r[i]);
                Task.WaitAll(tasks);
            }

            async Task ReadIssuer(Random r)
            {
                await Task.Yield();

                int batchSize = 1000;
                var readTasks = new ValueTask<(Status, int)>[batchSize];
                for (int i = 0; i < batchSize; i++)
                {
                    readTasks[i] = mixedStorageDictionary.ReadAsync(r.Next(numRecords));
                }
                for (int i = 0; i < batchSize; i++)
                    await readTasks[i].ConfigureAwait(false);
                
                if (Interlocked.Increment(ref cnt) % 1000 == 0)
                {
                    sw.Stop();
                    Console.WriteLine($"Time for {1000 * batchSize} ops = {sw.ElapsedMilliseconds}ms");
                    sw.Restart();
                }
            }
        }
    }

[JeremyTCD]

At a quick glance, I think the thread local idea doesn't work because the continuation of async will continue to use the same session (on the continuation thread).
...
Do you expect most Reads to happen synchronously, or will it go to disk most of the time?

Diagnosis makes sense.

As an aside, I looked up ConcurrentBag internals. They use ThreadLocal queues to avoid locking most of the time. Locking only occurs when a single thread requests a large number of objects, then they lock and steal objects from other threads' queues.

It felt wasteful placing a ConcurrentBag in front of Faster, after all of you guys' effort to avoid locking within Faster. But it seems framework concurrency datastructures avoid locking most of the time, so using them to pool sessions shouldn't significantly blunt Faster's edge.

I've gone with your suggested solution of a fast-path for sync and a ConcurrentQueue, works well!

Nice, clean sample by the way, would be nice to convert it eventually into an official sample to show best way to manage sessions internally to a non-session-based KV API.

Thanks. I'm finding Faster to be really useful. Makes it possible to use SSDs as "extended RAM" which allows for fewer round trips to the DB etc. I suspect simplified APIs could widen the user-base significantly.

I've cleaned up the repro with your suggestions and a couple of syntactic changes for consistency with my other code. Will paste the full thing at the end of this thread for reference.

The other thing with your example is the read concurrency is limited as you are not issuing requests in parallel

Yeah you're right, logic up till waiting for IO is sequential in the example program.

---

All that said, I've run into an issue with log compaction. I've updated the repro to reproduce the issue. Some snippets for quick reference:

public class MixedStorageDictionary<TKey, TValue> : IMixedStorageDictionary<TKey, TValue>
{
    ...

    // Log compaction
    private readonly LogAccessor<TKey, TValue> _logAccessor;
    private readonly int _operationsBetweenLogCompactions;
    private int _operationsSinceLastLogCompaction = 0; // We're using interlocked so this doesn't need to be volatile

    ...

    public MixedStorageDictionary(string? logFileDirectory = null,
        string? logFileName = null,
        int pageSizeBits = 12, // 4 KB pages
        int inMemorySpaceBits = 13, // 2 pages
        long indexNumBuckets = 1L << 20, // 64 MB index
        int operationsBetweenLogCompactions = 1000) // Number of upserts and deletes between compaction attempts
    {
        ...
        _operationsBetweenLogCompactions = operationsBetweenLogCompactions;
    }

    public void Upsert(TKey key, TValue obj)
    {
        if (_disposed)
        {
            throw new ObjectDisposedException(nameof(MixedStorageDictionary<TKey, TValue>));
        }

        GetSession().Upsert(key, obj);
        CompactLogIfRequired();
    }

    public Status Delete(TKey key)
    {
        if (_disposed)
        {
            throw new ObjectDisposedException(nameof(MixedStorageDictionary<TKey, TValue>));
        }

        Status result = GetSession().Delete(key);
        CompactLogIfRequired();

        return result;
    }

    ...

    private void CompactLogIfRequired()
    {
        // Faster log addresses:
        //
        // (oldest entries here) BeginAddress <= HeadAddress (where the in-memory region begins) <= SafeReadOnlyAddress (entries between here and tail updated in-place) < TailAddress (entries added here)
        //
        // If _operationsSinceLastLogCompaction < _operationsBetweenLogCompactions, it's not yet time to compact.
        // If _operationsSinceLastLogCompaction > _operationsBetweenLogCompactions, we're attempting to compact, do nothing.
        if (Increment(ref _operationsSinceLastLogCompaction) != _operationsBetweenLogCompactions)
        {
            return;
        }

        if (_logAccessor.BeginAddress == _logAccessor.SafeReadOnlyAddress) // All records fit within update-in-place region, nothing to compact
        {
            Exchange(ref _operationsSinceLastLogCompaction, 0);
            return;
        }

        // TOOD throws
        CompactLog();

        // Run asynchronously
        // Task.Run(CompactLog);
    }

    private void CompactLog()
    {
        long compactUntilAddress = (long)(_logAccessor.BeginAddress + 0.2 * (_logAccessor.SafeReadOnlyAddress - _logAccessor.BeginAddress));
        Console.WriteLine("compacting");
        // TODO throws
        _store.For(_simpleFunctions).NewSession<SimpleFunctions<TKey, TValue>>().Compact(compactUntilAddress, shiftBeginAddress: true);

        // GetSession().Compact(compactUntilAddress, shiftBeginAddress: true);
        Exchange(ref _operationsSinceLastLogCompaction, 0);
    }

    ...
}

I've also slightly tweaked Program.cs in the repro. The exception is thrown when upserts run in parallel and seems to occur more when the dictionary's TValue is string:

Unhandled exception. System.AggregateException: One or more errors occurred. (Object reference not set to an instance of an object.)
 ---> System.NullReferenceException: Object reference not set to an instance of an object.
   at FASTER.core.GenericScanIterator`2.BufferAndLoad(Int64 currentAddress, Int64 currentPage, Int64 currentFrame)
   at FASTER.core.GenericScanIterator`2.GetNext(RecordInfo& recordInfo)
   at FASTER.core.LogAccessor`2.LogScanForValidity[Input,Output,Context,Functions](Int64& untilAddress, Int64& scanUntil, ClientSession`6 tempKvSession)
   at FASTER.core.LogAccessor`2.Compact[Input,Output,Context,Functions,CompactionFunctions](ClientSession`6 fhtSession, Functions functions, CompactionFunctions cf, Int64 untilAddress, VariableLengthStructSettings`2 variableLengthStructSettings, Boolean shiftBeginAddress)
   at FASTER.core.ClientSession`6.Compact[CompactionFunctions](Int64 untilAddress, Boolean shiftBeginAddress, CompactionFunctions compactionFunctions)
   at FASTER.core.ClientSession`6.Compact(Int64 untilAddress, Boolean shiftBeginAddress)
   at Repro.Faster.IndexOutOfRangeException.MixedStorageDictionary`2.CompactLog() in C:\Projects\Repro.Faster.IndexOutOfRangeException\Repro.Faster.IndexOutOfRangeException\MixedStorageDictionary.cs:line 152
   at Repro.Faster.IndexOutOfRangeException.MixedStorageDictionary`2.CompactLogIfRequired() in C:\Projects\Repro.Faster.IndexOutOfRangeException\Repro.Faster.IndexOutOfRangeException\MixedStorageDictionary.cs:line 141
   at Repro.Faster.IndexOutOfRangeException.MixedStorageDictionary`2.Upsert(TKey key, TValue obj) in C:\Projects\Repro.Faster.IndexOutOfRangeException\Repro.Faster.IndexOutOfRangeException\MixedStorageDictionary.cs:line 76
   at Repro.Faster.IndexOutOfRangeException.Program.<>c__DisplayClass0_0.<Main>b__0(Int32 key) in C:\Projects\Repro.Faster.IndexOutOfRangeException\Repro.Faster.IndexOutOfRangeException\Program.cs:line 16
   at System.Threading.Tasks.Parallel.<>c__DisplayClass19_0`1.<ForWorker>b__1(RangeWorker& currentWorker, Int32 timeout, Boolean& replicationDelegateYieldedBeforeCompletion)
--- End of stack trace from previous location ---
   at System.Threading.Tasks.Parallel.<>c__DisplayClass19_0`1.<ForWorker>b__1(RangeWorker& currentWorker, Int32 timeout, Boolean& replicationDelegateYieldedBeforeCompletion)
   at System.Threading.Tasks.TaskReplicator.Replica`1.ExecuteAction(Boolean& yieldedBeforeCompletion)
   at System.Threading.Tasks.TaskReplicator.Replica.Execute()
   --- End of inner exception stack trace ---
   at System.Threading.Tasks.TaskReplicator.Run[TState](ReplicatableUserAction`1 action, ParallelOptions options, Boolean stopOnFirstFailure)
   at System.Threading.Tasks.Parallel.ForWorker[TLocal](Int32 fromInclusive, Int32 toExclusive, ParallelOptions parallelOptions, Action`1 body, Action`2 bodyWithState, Func`4 bodyWithLocal, Func`1 localInit, Action`1 localFinally)
--- End of stack trace from previous location ---
   at System.Threading.Tasks.Parallel.ThrowSingleCancellationExceptionOrOtherException(ICollection exceptions, CancellationToken cancelToken, Exception otherException)
   at System.Threading.Tasks.Parallel.ForWorker[TLocal](Int32 fromInclusive, Int32 toExclusive, ParallelOptions parallelOptions, Action`1 body, Action`2 bodyWithState, Func`4 bodyWithLocal, Func`1 localInit, Action`1 localFinally)
   at System.Threading.Tasks.Parallel.For(Int32 fromInclusive, Int32 toExclusive, Action`1 body)
   at Repro.Faster.IndexOutOfRangeException.Program.Main() in C:\Projects\Repro.Faster.IndexOutOfRangeException\Repro.Faster.IndexOutOfRangeException\Program.cs:line 16
   at Repro.Faster.IndexOutOfRangeException.Program.<Main>()

[badrishc]

• Fix to log compaction is here: [C#] Log compaction fix in presence of concurrent updates #406
• Your code does log compaction too frequently, and log compaction does not help during insert phase because all records are unique so it just rolls over all records to the tail. This makes the process rather slow as there is a lot of insertion in parallel with existing insert threads.
• Log compaction should not be invoked in parallel by multiple threads. One simple fix is:

        int _compact = 0;

        private void CompactLog()
        {
            if (CompareExchange(ref _compact, 1, 0) == 0)
            {
                long compactUntilAddress = (long)(_logAccessor.BeginAddress + 0.2 * (_logAccessor.SafeReadOnlyAddress - _logAccessor.BeginAddress));
                Console.WriteLine("compacting");
                // TODO throws
                _store.For(_simpleFunctions).NewSession<SimpleFunctions<TKey, TValue>>().Compact(compactUntilAddress, shiftBeginAddress: true);

                // GetSession().Compact(compactUntilAddress, shiftBeginAddress: true);
                Exchange(ref _operationsSinceLastLogCompaction, 0);
                _compact = 0;
            }
        }

[badrishc]

Also, it would be better would be to have a separate compaction task that wakes periodically, checks if the log on disk has grown too much (by comparing Log.BeginAddress and Log.SafeReadOnlyAddress), and if it is larger than a threshold, then does the compaction and goes to sleep. Checking for this on every Upsert/Delete is rather expensive.

The workload should benefit from log compaction, i.e., during compaction there should be sufficient dead records that compaction is useful. Otherwise, if all records are still live, then compaction will simply waste cycles and write the records back to the tail of the log.

[JeremyTCD]

Fix to log compaction is here: #406

Thanks for the quick fix!

I considered a threshold initially - compact when (SafeReadOnly - Begin)/RecordSize > threshold. The thing that bothered me about it was, what happens when
the log is compact but still exceeds the threshold? Once that point is reached, every check would trigger a redundant compaction.

I figured updates and deletes (once pushed outside the readonly region) pretty much guarantee gains from compaction, hence the counter approach. The insert thing bothered me too, I understand why its pointless to "compact" a sequence of inserts, but upsert couldn't be replaced with separate update and insert operations and most of my inserts occur close to application startup.

That said, points taken on overhead of checking every upsert/delete and how bad the counter approach is when it's all inserts. I ended up using a while loop with Task.Delay. Am keeping track of the number of sequential checks that result in compaction and adjusting the threshold based on that, e.g. n sequential checks all result in compaction --> raise threshold.

Extracted my wrapper to a standalone repo for reuse across projects. Will share here after I've neatened it up.

[badrishc]

Great. I think some version of a session-free API should make it into this repo as well, as a sample at first but eventually supported natively. Feel free to send a pull request!

[JeremyTCD]

Jering.KeyValueStore. Working Faster wrapper that handles:

• Sessions
• Support for variable length types using SpanByte and MessagePack for binary serialization
• Periodic log compaction
• Configuration for general use

Doesn't support persistence.

Usage:

var mixedStorageKVStore = new MixedStorageKVStore<int, string>();

// Insert
mixedStorageKVStore.Upsert(0, "dummyString1");

// Read
(Status status, string? result) = await mixedStorageKVStore.ReadAsync(0).ConfigureAwait(false);
Assert.Equal(Status.OK, status);
Assert.Equal("dummyString1", result);

// Update
mixedStorageKVStore.Upsert(0, "dummyString2");

// Verify updated
(status, result) = await mixedStorageKVStore.ReadAsync(0).ConfigureAwait(false);
Assert.Equal(Status.OK, status);
Assert.Equal("dummyString2", result);

// Delete
mixedStorageKVStore.Delete(0);

// Verify deleted
(status, result) = await mixedStorageKVStore.ReadAsync(0).ConfigureAwait(false);
Assert.Equal(Status.NOTFOUND, status);
Assert.Null(result);

[nhaberl]

Great man, as an application developer this is a point to start, way easier to use than the core lib.