ConcurrentDictionary equivalent

[uler3161]

Hi, just found FASTER. Looks interesting, but it's awfully complex.

I'm just trying to create a replacement for ConcurrentDictionary, but I can't seem to find out how. More specifically, there's a couple things I want:

• I utilize ConcurrentDictionary.GetOrAdd(TKey key, Func<TKey, TValue> valueFactory) a lot. Mainly because creating values takes some time. I can't seem to figure out how to do this with FASTER. Seems like all the examples show creating a value prior to calling upsert.
• I want to be able to run my application multiple times and reload pre-cached data. I think I have this working with ResumeSession.

Honestly, the documentation for IFunctions has me confused. I don't know if I need to use those somehow or if they are unrelated to what I'm trying to do.

What I've tried so far:

1. Just do upsert all the time. Problem is it seems to increase the cache size every time, so I assume it's creating duplicates. I assume that's what is meant by "does not check for prior values" in the docs. And even if it didn't create duplicates, it doesn't give me the option to build the value only if the key doesn't exist like the valueFactory in ConcurrentDictionary allows.
2. Try to do a read before upsert and only upsert if read results in NotFound status. I'm not sure if this is the most efficient way to accomplish what I want, but it doesn't seem to work anyway. The read call results in "An item with the same key has already been added" exceptions. I don't see how a simple read would cause that, but it is. Very strange that the upsert without read worked fine even though I think there were duplicate keys.

The following code is where I'm at. It's based on various examples I've found on github. It'd be nice to know what's wrong or, perhaps easier, if someone has a drop-in replacement for ConcurrentDictionary based on FASTER.

using System.Text;
using FASTER.core;

namespace FasterKVTest
{
    internal class Program
    {
        private static void AddItem(
            int index,
            ClientSession<CacheKey, CacheValue, CacheValue, CacheValue, Empty,
                IFunctions<CacheKey, CacheValue, CacheValue, CacheValue, Empty>>? session)
        {
            var cacheKey = new CacheKey($"Key{index}");
            var read = session.Read(cacheKey, out var output);
            if (read.NotFound)
            {
                session.Upsert(cacheKey, new CacheValue($"Value{index}"));
            }
        }

        static void Main(
            string[] args)
        {
            var log = Devices.CreateLogDevice("E:/DigitalLogicCache/hlog.log", deleteOnClose: false);
            var objectLog = Devices.CreateLogDevice("E:/DigitalLogicCache/hlog.obj.log", deleteOnClose: false);

            var logSettings = new LogSettings
            {
                LogDevice = log,
                ObjectLogDevice = objectLog,
                ReadCacheSettings = null,
                PageSizeBits = 12,
                MemorySizeBits = 20
            };

            var serializerSettings = new SerializerSettings<CacheKey, CacheValue>
            {
                keySerializer = () => new CacheKeySerializer(),
                valueSerializer = () => new CacheValueSerializer()
            };

            var store = new FasterKV<CacheKey, CacheValue>(size: 1L << 20, logSettings: logSettings,
                checkpointSettings: new CheckpointSettings
                {
                    CheckpointDir = "E:/DigitalLogicCache"
                }, serializerSettings: serializerSettings, comparer: new CacheKey(""));

            var recovered = true;
            try
            {
                store.Recover();
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                recovered = false;
            }

            Console.WriteLine($"Recovered session? {recovered}");

            var functions = new SimpleFunctions<CacheKey, CacheValue>();

            using var session = recovered
                ? store.ResumeSession(functions, "S", out _)
                : store.NewSession(functions, "S");

            Enumerable.Range(0, 1000000)
                .AsParallel()
                .WithDegreeOfParallelism(8)
                .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                .ForAll(index =>
                {
                    AddItem(index, session);
                });

            session.CompletePending(true);
            store.TakeFullCheckpointAsync(CheckpointType.FoldOver)
                .GetAwaiter()
                .GetResult();

            Console.WriteLine(store.EntryCount);
        }
    }

    public struct CacheKey : IFasterEqualityComparer<CacheKey>
    {
        public string _key;

        public CacheKey(
            string key)
        {
            _key = key;
        }

        public long GetHashCode64(
            ref CacheKey key)
        {
            return key._key.GetHashCode();
        }

        public bool Equals(
            ref CacheKey k1,
            ref CacheKey k2)
        {
            return k1._key == k2._key;
        }
    }

    public class CacheKeySerializer : BinaryObjectSerializer<CacheKey>
    {
        public override void Deserialize(
            out CacheKey obj)
        {
            var byteCounts = reader.ReadInt32();
            obj = new CacheKey(Encoding.Default.GetString(reader.ReadBytes(byteCounts)));
        }

        public override void Serialize(
            ref CacheKey obj)
        {
            writer.Write(obj._key.Length);
            writer.Write(Encoding.Default.GetBytes(obj._key));
        }
    }

    public class CacheValueSerializer : BinaryObjectSerializer<CacheValue>
    {
        public override void Deserialize(
            out CacheValue obj)
        {
            var byteCounts = reader.ReadInt32();
            var bytes = reader.ReadBytes(byteCounts);

            obj = new CacheValue("")
            {
                _bytes = bytes
            };
        }

        public override void Serialize(
            ref CacheValue obj)
        {
            writer.Write(obj._bytes.Length);
            writer.Write(obj._bytes);
        }
    }

    public class CacheValue
    {
        public byte[] _bytes;

        public CacheValue(
            string key)
        {
            _bytes = Encoding.Default.GetBytes(key);
        }
    }
}

Edit 1:
I was looking at the docs and my code some more and I'm not really sure whether sessions are thread-safe?. That I think may throw a wrench into what I'm trying to do and might be the reason I'm getting the same key already added exceptions? I tried moving the code that creates the session into the ForAll block and it doesn't work on resuming because it can't find the named session.

Edit 2:
So I decided to use ThreadLocal to get a session per thread. I don't know yet whether that's necessary vs just using one session for all threads. Still don't know how to utilize ResumeSession unless I were to name the session after the thread ID. Don't think that really works anyway, because who knows whether the thread IDs are going to be the same the next time I run the application. Plus, I don't really understand how/whether one session has access to data stored in another session which seems like I'd need in order to do what I want.

Edit 3:
I've decided to try storing session IDs as suggested here: #73.

Changed my code that adds items to do this:

var threadSafeSession = new ThreadSafeSession<CacheKey, CacheValue>(recovered, store);

Enumerable.Range(0, 1000000)
    .AsParallel()
    .WithDegreeOfParallelism(8)
    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
    .ForAll(index =>
    {
        var session = threadSafeSession.Get();
        AddItem(index, session);
        session.CompletePending(true);
    });

Functions that I was hoping would implement GetOrAdd:

public class GetOrAddFunctions<TKey, TValue> : FunctionsBase<TKey, TValue, Func<TKey, TValue>, TValue, Empty>
{
    public override bool CopyUpdater(
        ref TKey key,
        ref Func<TKey, TValue> input,
        ref TValue oldValue,
        ref TValue newValue,
        ref TValue output,
        ref RMWInfo rmwInfo)
    {
        newValue = oldValue;
        return true;
    }

    public override bool InitialUpdater(
        ref TKey key,
        ref Func<TKey, TValue> input,
        ref TValue value,
        ref TValue output,
        ref RMWInfo rmwInfo)
    {
        var tValue = input(key);
        value = tValue;
        return true;
    }

    public override bool InPlaceUpdater(
        ref TKey key,
        ref Func<TKey, TValue> input,
        ref TValue value,
        ref TValue output,
        ref RMWInfo rmwInfo)
    {
        return true;
    }
}

And then the ThreadSafeSession class is:

public class ThreadSafeSession<TKey, TValue>
{
    private readonly ThreadLocal<ClientSession<TKey, TValue, Func<TKey, TValue>, TValue, Empty,
        IFunctions<TKey, TValue, Func<TKey, TValue>, TValue, Empty>>> _localSession;

    private int _sessionCount;

    public ThreadSafeSession(
        bool isRecovered,
        IFasterKV<TKey, TValue> store)
    {
        Func<ClientSession<TKey, TValue, Func<TKey, TValue>, TValue, Empty,
            IFunctions<TKey, TValue, Func<TKey, TValue>, TValue, Empty>>> builder = () =>
        {
            var functions = new GetOrAddFunctions<TKey, TValue>();
            var sessionName = GetSessionName();

            var session = isRecovered
                ? store.ResumeSession(functions, GetSessionID(sessionName), out _)
                : store.NewSession(functions, sessionName);

            SaveSessionID(sessionName, session.ID);

            return session;
        };

        _localSession =
            new ThreadLocal<ClientSession<TKey, TValue, Func<TKey, TValue>, TValue, Empty,
                IFunctions<TKey, TValue, Func<TKey, TValue>, TValue, Empty>>>(builder);
    }

    private void SaveSessionID(
        string name,
        int id)
    {
        File.WriteAllText($"{name}.id", id.ToString());
    }

    private int GetSessionID(
        string name)
    {
        return int.Parse(File.ReadAllText($"{name}.id"));
    }

    public ClientSession<TKey, TValue, Func<TKey, TValue>, TValue, Empty,
        IFunctions<TKey, TValue, Func<TKey, TValue>, TValue, Empty>> Get()
    {
        return _localSession.Value;
    }

    private string GetSessionName()
    {
        var id = Interlocked.Increment(ref _sessionCount);
        var name = $"Session{id}";
        return name;
    }
}

However, I end up with a bunch of these kind of exceptions:

Unable to find session 7 to recover

Edit 4:

Well... after changing my CacheKey class to use an integer rather than a string AND making my test harness run on only a single thread, I was able to apparently get what I wanted. The log files aren't growing.

I don't know what is wrong with using a string in CacheKey. I changed the serializer to simply write/read the string value rather than the length and the bytes. Didn't help.

Changed the IFasterEqualityComparer to use a different hash code algorithm and I changed the equals to do a string.equals call. Didn't help.

And, oddly enough, the integer version seemed to save every single record (store.EntryCount matched the number of tests I ran). But, my original string CacheKey and CacheKeySerializer only seemed to save about 75% of the entries. And the changes I mentioned above resulted in only about 50% of the entries being saved. It's almost like there are hash collisions and equals isn't getting called?

And, as mentioned, I had to run this on a single thread. Still not possible to resume multiple named sessions. I found the store has a RecoverableSessions property but it's always empty (before and after the store.Recover() call).

Edit 5:

Getting closer. I don't understand why, but apparently the implementations of GetHashCode64 I've been trying don't work. I don't see why I can't just return the hash code of the CacheKey._key field, but it just plain doesn't work. What does mostly work is what I found here: https://github.com/microsoft/FASTER/blob/50a19c0ef386998b06230bf8e0c4d7880ccdb859/cs/playground/ClassRecoveryDurability/Types.cs

I say mostly works, because I was expecting every single entry to be in the cache when the application finishes (the Console.WriteLine(store.EntryCount) line). As mentioned in my previous edit, it does have the correct count when I use int in CacheKey rather than string. But when I use string, even with this new GetHashCode64 implementation, I'm only getting 99371 in a 100000 test and only 992297 in a 1000000 test. I guess at least I finally have a solution that uses a string in the key object and doesn't result in the log files continually growing on subsequent runs.

It also still doesn't recover in multi-threaded situation. Only way to recover a session is if there's only one session.

Edit 6:

I decided to remove the ResumeSession calls and always use NewSession. I think that might have fixed the multi-threaded issue. To be honest, I don't have the first clue what ResumeSession is used for. Doesn't seem to be useful for me.

That said, I found that all my values were empty. But, I found I needed to add this to my GetOrAddFunctions class to fix that issue:

    public override bool ConcurrentReader(
        ref TKey key,
        ref Func<TKey, TValue> input,
        ref TValue value,
        ref TValue dst,
        ref ReadInfo readInfo)
    {
        dst = value;
        return true;
    }

    public override bool SingleReader(
        ref TKey key,
        ref Func<TKey, TValue> input,
        ref TValue value,
        ref TValue dst,
        ref ReadInfo readInfo)
    {
        dst = value;
        return true;
    }

And, unfortunately, that's not the only problem. Perhaps this next problem is related to the same reason the EntryCount isn't matching the number of tests. I've updated the AddItem method to this:

    private static void AddItem(
        int index,
        ClientSession<CacheKey, CacheKey, Func<CacheKey, CacheKey>, CacheKey, Empty,
            IFunctions<CacheKey, CacheKey, Func<CacheKey, CacheKey>, CacheKey, Empty>> session)
    {
        var cacheKey = new CacheKey($"Key{index}");
        var results = session.Read(cacheKey);
        if (!results.status.NotFound && results.output._key != $"Item{index}")
        {
            Console.WriteLine($"Got {results.output._key} for {cacheKey._key} rather than Item{index}");
        }

        session.RMW(cacheKey, k => new CacheKey($"Item{index}"));
    }

So, the idea of this code is to only have console output if:

1. The key is in the cache AND
2. The value for the key doesn't match what it's supposed to. Specifically, "Key1" should have "Item1", "Key2" should have "Item2", etc.

If this was working correctly, I'd expect no output on the initial run (because all keys should result in NotFound) and I'd expect no output on subsequent runs (because the value found for that key matches what we expect).

On the initial run, if I use a small enough amount of tests it runs fine. And on subsequent runs on this amount of tests, it runs fine.

If, however, I use a large enough amount of tests (100000 it seems will do it), the initial run apparently finds matching records (which it shouldn't as all the keys are unique). And those records that it thinks it has matched all have empty values (which is also wrong because I've never upserted an empty value).

Edit 7:

It would appear my PageSizeBits and/or MemorySizeBits is causing these unexpected duplicate empty valued entries to be created. Removing those settings seems to fix the problem. Seems like a bug in FASTER, but not sure.

[badrishc]

FASTER is extremely efficient and high-performance but as you see there is a learning curve. It is definitely possible to create a simpler wrapper compatible with ConcurrentDictionary API on top, and this would be quite valuable to new users, but we have not done it ourselves. Feel free to contribute such a wrapper back to the repository.

[badrishc]

Also you should look at this sample in playground. It takes care of all the complexities of using sessions via a session pool. There is a pretty easy-to-use IFasterWrapper defined there as well:

https://github.com/microsoft/FASTER/tree/main/cs/playground/AsyncStress