ConcurrentCache.cs

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace Akka.Util
{
    //This is a cache based on the ConcurrentQueue
    //The idea is to remove stale entries after a certain amount of operations
    //Once the operation limit is reached, start purging stale items in the background
    public class ConcurrentCache<T>
    {
        public class ConcurrentCacheElement<T>
        {
            public ConcurrentCacheElement(T value)
            {
                Value = value;
            }
            public T Value { get; private set; }
            public int UseCount = 0;
        }

        private readonly ConcurrentDictionary<string, ConcurrentCacheElement<T>> _cache = new ConcurrentDictionary<string, ConcurrentCacheElement<T>>();
        private int _useCount;
        private readonly int _upperLimit;
        private readonly int _lowerLimit;
        public ConcurrentCache(int upperLimit, int lowerLimit)
        {
            _upperLimit = upperLimit;
            _lowerLimit = lowerLimit;
        }
        public T GetOrAdd(string key, Func<string, T> valueFactory)
        {
            if (Interlocked.Increment(ref _useCount) == _upperLimit)
            {
                //once we have reached the upper limit, spawn a task that starts removing items that haven't been used enough
                //items that are kept, get their use count reset to 0
                Task.Factory.StartNew(() =>
                {
                    foreach (var value in _cache.ToArray())
                    {
                        if (value.Value.UseCount <= _lowerLimit)
                        {
                            ConcurrentCacheElement<T> tmp;
                            _cache.TryRemove(value.Key, out tmp);
                        }
                        else
                        {
                            //reset use count
                            value.Value.UseCount = 0;
                        }
                    }
                    Volatile.Write(ref _useCount, 0);
                });
            }
            Func<string, ConcurrentCacheElement<T>> wrapper = k => new ConcurrentCacheElement<T>(valueFactory(k));

            var res = _cache.GetOrAdd(key, wrapper);
            Interlocked.Increment(ref res.UseCount);
            return res.Value;
        }
    }
}