Add a GetOrAdd that takes a value?

[mburbea]

ConcurrentDictionary has an overload of GetOrAdd that takes a value instead of a valueFactory, can an overload be added for the Lru?

[bitfaster]

The common use case for a cache is that you know the key, and computing the associated value is expensive. If you already have the value there is no need to do a cache lookup. The only scenario where knowing the value is useful is seeding the cache (unless I'm missing something), which presumably is done much less frequently than looking up values by key (else cache overhead would be high).

Today you can simply write:

cache.GetOrAdd(key, k => value);

With your proposed method, this would become:

cache.GetOrAdd(key, value);

I benchmarked both, speed difference between the two is negligible and adding a method increases code size (which is bad). Given that there isn't really a perf advantage, its an uncommon scenario and the workaround only involves typing a couple of extra characters, I'm inclined to keep the interface of the class smaller and not add this.

If it's really too inconvenient to write the lambda over and over, you can write your own extension method like this:

    public static class CacheExtensions
    {
        public static V GetOrAdd<K, V>(this ICache<K, V> cache, K key, V value)
        {
            return cache.GetOrAdd(key, k => value);
        }
    }

I will put this extension method into the next release.

[mburbea]

Yes, I'm currently using it with a delegate and trying to basically implement a version of GetOrAddAsync that will only invoke the valueFactory once in a very likely race to limit an expensive api call.

while(true)
{
      if(_cache.TryGet(key, out V value))
      {
           return await value.ConfigureAwait(false);
      }
      var tcs = new TaskCompletionSource<V>()
      if (_cache.GetOrAdd(key, _ => tcs.Task) is { } res && object.ReferenceEquals(tcs.Task, res))
      {
            try
            {
                  V val = await valueFactory(key).ConfigureAwait(false);
                  tcs.SetResult(val);
                  return val;
            }
            catch(Exception e)
            {
                  tcs.SetException(e);
                  _cache.Remove(key);
                  throw;
            }
     }
}

My concern is that every single time this is called I'm allocating an additional delegate. And in the usual case, you are right that it probably won't matter too much.

[bitfaster]

My advice would be to measure - if the API call is expensive, the additional delegate is not likely to be a measurable overhead.

Regarding the race and invoking value factory exactly once per key, there are a couple of other options like caching an AsyncLazy value or double checked locking. Yours is an interesting approach, I hadn't considered using TaskCompletionSource like that.

I tend to use double checked locking because the async lazy approach can be problematic if you get an exception lazy initializing (which is what your code is handling) or if the value is IDisposable. In the simplest form, the pattern looks like this:

private class DoubleChecked
{
   private object instance;
   private object @lock = new object();

   public object Instance()
   {
      if (instance == null)
      {
         lock (@lock)
         {
             if (instance == null)
             {
                instance = new object();
             }
         }

        return instance;
      }
   }
}

The code inside LRU GetOrAdd (and ConcurrentDictionary for that matter) is roughly this:

public V GetOrAdd(K key, Func<K, V> valueFactory)
{
   if (dictionary.TryGet(key, out var value)
   {
       return value;
   }

   var newValue = valueFactory(key);

   if (dictionary.TryAdd(key, newValue))
   {
       // bookkeeping...
      return newValue;
   }

   return GetOrAdd(key,valueFactory);
}

To synchronize value creation and avoid a race, we need to insert a lock statement starting from the line var newValue = valueFactory(key); To do that, we can create a wrapper class for the value factory that performs the synchronization.

Below is an example using SingletonCache so that the locking is per key (no global lock). For GetOrAddAsync, you would need to do the same thing using Semaphore instead of object for the lock - code is basically the same but its a bit longer with try/finally etc so I am writing the shorter version here. In production code you may also need to consider things lock timeouts etc.

public class SynchronizedValueFactory<K, V>
{
    private readonly ICache<K, V> cache;
    private readonly Func<K, V> valueFactory;
    private SingletonCache<K, object> lockCache = new SingletonCache<K, object>();

    public SynchronizedValueFactory(ICache<K, V> cache, Func<K, V> valueFactory)
    { 
        this.cache = cache;
        this.valueFactory = valueFactory;
    }

    public V Create(K key)
    { 
        using (var @lock = this.lockCache.Acquire(key))
        { 
            lock (@lock.Value)
            {
                return this.cache.GetOrAdd(key, this.valueFactory);
            }
        }
    }
}

To use SynchronizedValueFactory, you would write some code like this:

public void Test()
{ 
    var lru = new ConcurrentLru<int, string>(123);
    Func<int, string> valueFactory = x => x.ToString();
    var synchronized = new SynchronizedValueFactory<int, string>(lru, valueFactory);

    var res = lru.GetOrAdd(1, k => synchronized.Create(k));
}

Now the valueFactory delegate is invoked only once per key, inside the lock. If you were to flatten all this code out as if it were written in a single class, it would look like this in LRU.GetOrAdd - with the double checked lock inserted:

public V GetOrAdd(K key, Func<K, V> valueFactory)
{
   if (dictionary.TryGet(key, out var value)
   {
       return value;
   }

  V newValue;
  // SynchronizedValueFactory ////////////////////////////////////////////////////////
  using (var @lock = this.lockCache.Acquire(key))
  { 
      lock (@lock.Value)
      {
            // Nested GetOrAdd ////////////////////////////////////////////////////////
            if (dictionary.TryGet(key, out var value)
            {
                return value;
            }

            newValue = valueFactory(key);

            if (dictionary.TryAdd(key, newValue))
            {
                // bookkeeping...
            }
            // Nested GetOrAdd ////////////////////////////////////////////////////////
       }
   }
   // SynchronizedValueFactory ////////////////////////////////////////////////////////

   if (dictionary.TryAdd(key, newValue))
   {
       // bookkeeping...
      return newValue;
   }

   return GetOrAdd(key,valueFactory);
}

This will give the strict create value once guarantee, but with some overhead:

• For cache hits, it is equal cost (single dictionary lookup)
• For a cache miss, there are 2x dictionary lookups for the cache, 1x dictionary lookup for creating lock object, the 1x lookup and 1x dictionary remove for cleaning up the lock afterwards, and 2x dictionary.TryAdds in the LRU. Also SingletonCache will allocate 1 tracking object internally. Total = approx 7x dictionary ops.
• To reduce overhead you could replace the SingletonCache with a single global lock object to omit 3x dictionary ops and the tracking object allocs, but scalability is reduced - all threads would contend the same lock in the event of concurrent misses to different keys (values would be loaded serially instead of in parallel). Total = approx 4x dictionary ops.

The overhead of this approach vs handwriting the double checked lock code is the redundant second attempt to add the value to the dictionary after exiting SynchronizedValueFactory. If your value factory delegate is doing something expensive like an RPC call to a database or deserializing a large json payload or something, while wasteful it is not a practical concern.

[mburbea]

Thanks, in this case my caches are all of the form Lru<Key, Task<Value>>. This also means that I can handle this as basically an extension method, or really the only entry point.

The TCS approach seems to work well, and I adapted this from a gist by davidfowl of the aspnet core team (LRUs does not expose a TryAdd so, I'm using the GetOrAdd and just checking if the added Task is the winning one.)
https://gist.github.com/davidfowl/3dac8f7b3d141ae87abf770d5781feed

In this case, when I said expensive I meant actual dollars amounts. Calling this api for various keys adds up quickly. Our license agreement also puts in a clause of one year of caching results. (My valueFactory actually adds a second level cache through the database.)

Edit:
Also, one thing it'd be nice to do although I think its a bit niche is to allow setting the explicit expiration for a TLRU entry or similarly a TLRU with a sliding expiration mechanism. I'm currently using a SlidingWindowPolicy in my stack, where the only real change I've made is that the Touch restores the timeout.

[bitfaster]

I have been working on some ideas for caching Lazy values that would behave similarly to use of ConcurrentDictionary.GetOrAdd with respect to exceptions, i.e. not to cache them so the cache lookup doesn't throw forever. This is the PR.

I combined your approach into the class AsyncAtomic, which lets you write code as below with the same GetOrAddAsync signature (via an extension method):

var asyncAtomicLru = new ConcurrentLru<int, AsyncAtomic<int, int>>(5);

int ar = await asyncAtomicLru.GetOrAddAsync(1, i => Task.FromResult(i));

In a 100% cache hit benchmark, the lookup performance of this wrapper is about 20% slower than ConcurrentLru.GetOrAddAsync with no memory allocs - so no delegate alloc (numbers below based on .NET 6). This is a relatively small penalty to synchronize creation - it's about half the overhead compared to your code (I copy pasted it into the benchmark, you can see below).

Method	Mean	Ratio	Code Size	Allocated
ConcurrentDictionary.GetOrAdd	7.096 ns	0.16	340 B	-
ConcurrentLru.GetOrAddAsync	45.457 ns	1.00	495 B	-
AtomicConcurrentLru.GetOrAddAsync	55.937 ns	1.23	495 B	-
mburbea.GetOrAddAsync	64.890 ns	1.43	495 B	24 B

This is the benchmark code:

    [DisassemblyDiagnoser(printSource: true)]
    [MemoryDiagnoser]
    public class AsyncAtomicBench
    {
        private static readonly ConcurrentDictionary<int, int> dictionary = new ConcurrentDictionary<int, int>(8, 9, EqualityComparer<int>.Default);

        private static readonly ConcurrentLru<int, int> concurrentLru = new ConcurrentLru<int, int>(8, 9, EqualityComparer<int>.Default);

        private static readonly ConcurrentLru<int, AsyncAtomic<int, int>> atomicConcurrentLru = new ConcurrentLru<int, AsyncAtomic<int, int>>(8, 9, EqualityComparer<int>.Default);

        private static readonly ConcurrentLru<int, Task<int>> mburbeaConcurrentLru = new ConcurrentLru<int, Task<int>>(8, 9, EqualityComparer<int>.Default);

        [Benchmark()]
        public void ConcurrentDictionary()
        {
            Func<int, int> func = x => x;
            dictionary.GetOrAdd(1, func);
        }

        [Benchmark(Baseline = true)]
        public async Task ConcurrentLruAsync()
        {
            Func<int, Task<int>> func = x => Task.FromResult(x);
            await concurrentLru.GetOrAddAsync(1, func).ConfigureAwait(false);
        }

        [Benchmark()]
        public async Task AtomicConcurrentLruAsync()
        {
           Func<int, Task<int>> func = x => Task.FromResult(x);
           await atomicConcurrentLru.GetOrAddAsync(1, func).ConfigureAwait(false);
        }

        [Benchmark()]
        public async Task mburbeaGetOrAddAsync()
        {
           Func<int, Task<int>> func = x => Task.FromResult(x);
           await mburbeaConcurrentLru.GetOrAddAsync2(1, func).ConfigureAwait(false);
        }
    }

    public static class Ext
    { 
        public static async Task<V> GetOrAddAsync2<K, V>(this ICache<K, Task<V>> _cache, K key, Func<K, Task<V>> valueFactory)
        { 
            while(true)
            {
                  if(_cache.TryGet(key, out Task<V> value))
                  {
                       return await value.ConfigureAwait(false);
                  }
                  var tcs = new TaskCompletionSource<V>();

                  if (_cache.GetOrAdd(key, _ => tcs.Task) is { } res && object.ReferenceEquals(tcs.Task, res))
                  {
                        try
                        {
                              V val = await valueFactory(key).ConfigureAwait(false);
                              tcs.SetResult(val);
                              return val;
                        }
                        catch(Exception e)
                        {
                              tcs.SetException(e);
                              _cache.TryRemove(key);
                              throw;
                        }
                 }
            }
        }
    }

I need to do some more work on the family of Atomic classes before I would add these to the library - I want to have a full suite to handle IDisposable, async and IDisposable async with consistent APIs. In the mean time, you can just replicate this code in your repo, it works over the top of ICache<K, V>. Below are all the relevant classes.

    [DebuggerDisplay("IsValueCreated={IsValueCreated}, Value={ValueIfCreated}")]
    public class AsyncAtomic<K, V>
    {
        private volatile Initializer initializer;

        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private V value;

        public AsyncAtomic()
        {
            this.initializer = new Initializer();
        }

        public AsyncAtomic(V value)
        {
            this.value = value;
        }

        public async Task<V> GetValueAsync(K key, Func<K, Task<V>> valueFactory)
        {
            if (this.initializer == null)
            {
                return this.value;
            }

            return await CreateValueAsync(key, valueFactory).ConfigureAwait(false);
        }

        public bool IsValueCreated => this.initializer == null;

        public V ValueIfCreated
        {
            get
            {
                if (!this.IsValueCreated)
                {
                    return default;
                }

                return this.value;
            }
        }

        private async Task<V> CreateValueAsync(K key, Func<K, Task<V>> valueFactory)
        {
            Initializer init = this.initializer;

            if (init != null)
            {
                this.value = await init.CreateValue(key, valueFactory).ConfigureAwait(false);
                this.initializer = null;
            }

            return this.value;
        }

        private class Initializer
        {
            private object syncLock = new object();
            private bool isInitialized;
            private Task<V> valueTask;

            public async Task<V> CreateValue(K key, Func<K, Task<V>> valueFactory)
            {
                var tcs = new TaskCompletionSource<V>(TaskCreationOptions.RunContinuationsAsynchronously);

                var synchronizedTask = Synchronized.Initialize(ref this.valueTask, ref isInitialized, ref syncLock, tcs.Task);

                if (ReferenceEquals(synchronizedTask, tcs.Task))
                {
                    try
                    {
                        var value = await valueFactory(key).ConfigureAwait(false);
                        tcs.SetResult(value);
                        return value;
                    }
                    catch (Exception ex)
                    {
                        Volatile.Write(ref isInitialized, false);
                        tcs.SetException(ex);
                        throw;
                    }
                }

                return await synchronizedTask.ConfigureAwait(false);
            }
        }
    }

    public static class AsyncAtomicCacheExtensions
    {
        public static Task<V> GetOrAddAsync<K, V>(this ICache<K, AsyncAtomic<K, V>> cache, K key, Func<K, Task<V>> valueFactory)
        {
            var synchronized = cache.GetOrAdd(key, _ => new AsyncAtomic<K, V>());
            return synchronized.GetValueAsync(key, valueFactory);
        }

        public static void AddOrUpdate<K, V>(this ICache<K, AsyncAtomic<K, V>> cache, K key, V value)
        {
            cache.AddOrUpdate(key, new AsyncAtomic<K, V>(value));
        }

        public static bool TryUpdate<K, V>(this ICache<K, AsyncAtomic<K, V>> cache, K key, V value)
        {
            return cache.TryUpdate(key, new AsyncAtomic<K, V>(value));
        }

        public static bool TryGet<K, V>(this ICache<K, AsyncAtomic<K, V>> cache, K key, out V value)
        {
            AsyncAtomic<K, V> output;
            bool ret = cache.TryGet(key, out output);

            if (ret)
            {
                value = output.ValueIfCreated;
            }
            else
            {
                value = default;
            }

            return ret;
        }
    }

    internal static class Synchronized
    {
        public static V Initialize<V>(ref V target, ref bool initialized, ref object syncLock, V value)
        {
            // Fast path
            if (Volatile.Read(ref initialized))
            {
                return target;
            }

            lock (syncLock)
            {
                if (!Volatile.Read(ref initialized))
                {
                    target = value;
                    Volatile.Write(ref initialized, true);
                }
            }

            return target;
        }
    }

[bitfaster]

Atomic value factory is now implemented in the following PRs: #155 + #153

Create an async cache like this:

IAsyncCache<int, int> cache = new LruBuilder<int, int>()
    .WithAtomicCreate()
    .AsAsyncCache()
    .Build();

Then call it as below, where func will be executed exactly once per key (per WithAtomicCreate() when the cache was created):

Func<int, Task<int>> func = x => Task.FromResult(x);
await cache.GetOrAddAsync(1, func);

This will be part of the next release.