Concurrent non-blocking update of cached list of on-line users

Question

I have following problem: My server (ASP.MVC WebAPI) is tracking, when client application ("Agent") is on-line. It's storing this inf on following table:

+-------+-------------------------------+-----+
|AgentId|              Date             |State|
+-------+-------------------------------+-----+
| 1     | 30 may 2016 г. 3:02:20 +03:00 |True |
| 1     | 30 may 2016 г. 0:25:26 +02:00 |True |
| 1     |29 may 2016 г. 23:05:59 +02:00 |False|
| 1     |29 may 2016 г. 23:05:01 +02:00 |True |
+-------+-------------------------------+-----+

"Agent" software is sending some data to the server with some frequecy, each HTTP request is threated as ping/keep alive, and tels server, that agent is Up and running.

Algorithm is following:

• Ping received, if previous ping time isn't stored in RAM cache:
  • Write ping time into cache
  • Add DB record, that "Agent" become on-line
• Ping received, if previous ping time is stored in RAM cache:
  • update ping time in cache
• Server haven'r received anything from "Agent" for 10 minutes (AgentTimeout)
  • remove this agent's record from cache
  • add record to DB that agent wen't offline

Following implementation is using non blocking algorithm. I'd like you to review correctness of this algorithm.

namespace X
{
    [SingletonScope]
    public class AgentPingReceiver : IAgentPingReceiver
    {
        private ILogger Log = Serilog.Log.Logger.ForContext<AgentPingReceiver>();

        private static readonly TimeSpan AgentTimeout = TimeSpan.FromMinutes(10);
        private static readonly TimeSpan StateRefreshInterval = TimeSpan.FromMinutes(1);


        private CancellationTokenSource _cancelToken;
        ConcurrentDictionary<int,DateTimeOffset> _lastPingReceived;

        public AgentPingReceiver()
        {
            _lastPingReceived = new ConcurrentDictionary<int, DateTimeOffset>();
            _cancelToken = new CancellationTokenSource();

            Task.Run((Func<Task>)BackgroundChecker,_cancelToken.Token);            
        }

        private async Task BackgroundChecker()
        {
            do
            {
                if (_cancelToken.IsCancellationRequested)
                    break;

                try
                {
                    await Task.Delay(StateRefreshInterval, _cancelToken.Token);
                }
                catch (TaskCanceledException)
                {
                    break;
                }

                if (_lastPingReceived.Any())
                {
                    var lastValidDateTime = DateTimeOffset.Now - AgentTimeout;


                    var itemsToRemove = _lastPingReceived
                        .Where(p => p.Value <= lastValidDateTime)
                        .ToList();

                    if(itemsToRemove.Any())
                    using (IocMannager.BeginScope())
                    {
                        IAgentStateChangeHandler handler =
                            IocMannager.GetServiceContainer<IAgentStateChangeHandler>().Service;

                        foreach (var kv in itemsToRemove)
                        {
                            await handler.ChangeAgentState(kv.Key, false, kv.Value);
                            if (!((ICollection<KeyValuePair<int, DateTimeOffset>>) _lastPingReceived).Remove(kv))
                            {
                                DateTimeOffset newValue;
                                if (_lastPingReceived.TryGetValue(kv.Key, out newValue))
                                {
                                    //Item was resurected, but PingReceived() didn't knew that we are deleting it
                                    //                                                 so updating DB accordingly
                                    //
                                    //P.S. I thought to put handler.ChangeAgentState(kv.Key, false, kv.Value) if Remove() return true
                                    //     but then I realized, that if item was offline for 10 minutes, than it really was gone!
                                    //     Just need to keep DB in sync.
                                    try
                                    {
                                        await handler.ChangeAgentState(kv.Key, true, newValue);
                                    }
                                    catch (Exception e)
                                    {
                                        // we must not let this method to fail
                                        Log.Error(e, "Error in background checker");
                                    }
                                }
                            }
                        }
                    }
                }
            } while (true);
        }

        public async Task PingReceived(int agentId, DateTimeOffset when)
        {
            bool isNew = true;
            _lastPingReceived.AddOrUpdate(agentId, when, (_, __) =>
                {
                    isNew = false;
                    return when;
                }
            );

            if (isNew)
            {
                using (var serviceContainer = IocMannager.GetServiceContainer<IAgentStateChangeHandler>())
                {
                    var handler = serviceContainer.Service;
                    await handler.ChangeAgentState(agentId, true, when);
                }
            }
        }

        public void Dispose()
        {
            if (_cancelToken != null)
            {
                _cancelToken.Cancel();
                _cancelToken.Dispose();
            }
        }
    }
}

P.S. I foresee, that

 if (!((ICollection<KeyValuePair<int, DateTimeOffset>>) _lastPingReceived).Remove(kv))
     ...  
     await handler.ChangeAgentState(kv.Key, true, newValue);

code can cause duplicates in some rare cases, so inside method ChangeAgentState I'm chcking for KeyDuplicate exception, and ignoring it.

Answer 1

OK, it is going to be a long answer :)

UPDATE

Actually, marking for deletion is not necessary, see Little-known gems: Atomic conditional removals from ConcurrentDictionary.

I would still consider the rest of the answer as an approach to structure the task.

END OF UPDATE

ConcurrentDictionary - when you about to remove from it - first mark for deletion to get rid of racing conditions. I defined this helper class to assist with this task:

class AsyncDictionary<TKey, TValue>
    where TValue : class
{
    ConcurrentDictionary<TKey, TValue> Values { get; } =
        new ConcurrentDictionary<TKey, TValue>();

    public delegate bool TryUpdate(ref TValue value);
    
    /// <summary>
    /// It makes multiple attempts to execute if item under the question is marked for deletion.
    /// </summary>
    /// <param name="key">Well, the key.</param>
    /// <param name="add">Value factory for add opeartion.</param>
    /// <param name="tryUpdate">It will try to update, but might reject to do so.</param>
    /// <returns>Reports actual operation executed and the value assigned.</returns>
    public async Task<Result> AddOrUpdateAsync(TKey key, Func<TValue> add, TryUpdate tryUpdate)
    {
        while (true)
        {
            var result = Result.Cancelled;
            Values.AddOrUpdate(key,
                (k) =>
                {
                    result = new Result(ValueAction.Added, add());
                    return result.Value;
                },
                (k, v) =>
                {
                    if (v == null)
                        result = Result.Delayed;
                    else                        
                        result = tryUpdate(ref v) ?
                            new Result(ValueAction.Updated, v) :
                            Result.Cancelled;                        

                    return v;
                });

            if (result.Action == ValueAction.Delayed)
                await Task.Delay(10);
            else
                return result;
        }
    }

    /// <summary>
    /// Concurrent remove first marks the entry for deletion by assigning null 
    /// to the value. 
    /// </summary>
    /// <param name="key">Well, the key.</param>
    /// <param name="guard">Checks if we really want to kill it.</param>
    /// <returns></returns>
    public Result Remove(TKey key, Predicate<TValue> guard)
    {
        TValue value;
        if (!Values.TryGetValue(key, out value))
            return Result.Cancelled;

        if (value == null)
            return Result.Cancelled;

        if (!guard(value))
            return Result.Cancelled;

        if (!Values.TryUpdate(key, null, value))
            return Result.Cancelled;

        TValue nullValue;
        Values.TryRemove(key, out nullValue);
        if (nullValue != null)
            throw new NotImplementedException();

        return new Result(ValueAction.Removed, value);
    }

    /// <summary>
    /// Removes everything matching the predicate.
    /// </summary>
    /// <param name="predicate">To be or not to be.</param>
    /// <returns>The corpses.</returns>
    public IEnumerable<Result> Remove(Predicate<TValue> predicate) =>
        from kvp in Values
        where predicate(kvp.Value)
        select Remove(kvp.Key, predicate) into r
        where r.Action == ValueAction.Removed
        select r;        

    public struct Result
    {
        public static readonly Result Cancelled = new Result(ValueAction.Cancelled);
        public static readonly Result Removed = new Result(ValueAction.Removed);
        public static readonly Result Delayed = new Result(ValueAction.Delayed);

        public Result(ValueAction action)
            : this(action, null)
        {
        }

        public Result(ValueAction action, TValue value)
        {
            Action = action;
            Value = value;
        }

        public ValueAction Action { get; }
        public TValue Value { get; }
    }
}

enum ValueAction
{
    Added,
    Updated,
    Removed,
    Cancelled,
    Delayed
}

Now let’s define abstractions for proper dependency management:

interface IClock
{
    DateTimeOffset GetTime();
}

interface ISessionMonitor : IDisposable
{
    Task StartAsync(Session session);
    Task EndAsync(Session session);
}

interface IPingMonitor : IDisposable
{
    Task PingAsync(AgentId agentId);
}

A little bit of "explicit language" - not necessary at all :)

struct AgentId
{
    public static implicit operator AgentId(int value) => new AgentId { Value = value };
    public static implicit operator int(AgentId agentId) => agentId.Value;
    int Value { get; set; }
    public override string ToString() => $"Agent #{Value}";
}

We are defining one business object, it knows how to calculate expiration:

class Session : ValueObject<Session>
{
    private static readonly TimeSpan Ttl = TimeSpan.FromMinutes(10);

    public static Session Start(AgentId agentId, DateTimeOffset at) =>
        new Session(agentId, at, at);

    public Session(AgentId agentId, DateTimeOffset started, DateTimeOffset lastActivity)
    {
        AgentId = agentId;
        Started = started;
        LastActivity = lastActivity;
    }

    public AgentId AgentId { get; }
    public DateTimeOffset Started { get; }
    public DateTimeOffset LastActivity { get; }
    public bool IsExpired(DateTimeOffset at) => LastActivity + Ttl < at;

    public bool TryExtend(DateTimeOffset till, out Session extended)
    {
        extended = this;
        if (IsExpired(till))
            return false;

        extended = new Session(AgentId, Started, till);
        return true;
    }

    protected override IEnumerable<object> EqualityCheckAttributes =>
        new object[] { AgentId, Started };
}

The most interesting thing:

class PingMonitor : IPingMonitor
{
    private static readonly TimeSpan Interval = TimeSpan.FromMinutes(1);
    AsyncDictionary<AgentId, Session> Sessions { get; } =
        new AsyncDictionary<AgentId, Session>();

    public PingMonitor(IClock clock, ISessionMonitor monitor)
    {
        Clock = clock;
        Monitor = monitor;
        CancellationTokenSource = new CancellationTokenSource();
        Task.Run(RunAsync);
    }

    public void Dispose()
    {
        CancellationTokenSource.Cancel();
        Monitor.Dispose();
    }

    IClock Clock { get; }
    ISessionMonitor Monitor { get; }
    CancellationTokenSource CancellationTokenSource { get; }

    public Task PingAsync(AgentId agentId) =>
        StartOrExtendAsync(agentId);

    async Task StartOrExtendAsync(AgentId agentId)
    {
        var result = await Sessions.AddOrUpdateAsync(
            agentId,
            () => Session.Start(agentId, Clock.GetTime()),
            (ref Session s) => s.TryExtend(Clock.GetTime(), out s));

        switch (result.Action)
        {
            case ValueAction.Added:
                await Monitor.StartAsync(result.Value);
                return;

            case ValueAction.Updated:
                return;

            case ValueAction.Cancelled:
                await EndAsync(agentId);
                await StartOrExtendAsync(agentId);
                return;

            default:
                throw new NotImplementedException();
        }
    }

    async Task EndAsync(AgentId agentId)
    {
        var result = Sessions.Remove(agentId, s => s.IsExpired(Clock.GetTime()));
        switch (result.Action)
        {
            case ValueAction.Removed:
                await Monitor.EndAsync(result.Value);
                return;

            case ValueAction.Cancelled:
                return;

            default:
                throw new NotImplementedException();
        }
    }

    async Task RunAsync()
    {
        while (!CancellationTokenSource.Token.IsCancellationRequested)
        {
            await Task.Delay(Interval, CancellationTokenSource.Token);
            await Task.WhenAll(from r in Sessions.Remove(s => s.IsExpired(Clock.GetTime()))
                               select Monitor.EndAsync(r.Value));
        }
    }
}

I used this helper classes:

class Clock : IClock
{
    public DateTimeOffset GetTime() => DateTimeOffset.Now;
}

abstract class ValueObject<T> : IEquatable<ValueObject<T>>
    where T : ValueObject<T>
{
    protected abstract IEnumerable<object> EqualityCheckAttributes { get; }

    public override int GetHashCode() =>
        EqualityCheckAttributes
            .Aggregate(0, (hash, a) => hash = hash * 31 + (a?.GetHashCode() ?? 0));

    public override bool Equals(object obj) =>
        Equals(obj as ValueObject<T>);

    public bool Equals(ValueObject<T> other) =>
        other != null &&
            EqualityCheckAttributes.SequenceEqual(other.EqualityCheckAttributes);

    public static bool operator ==(ValueObject<T> left, ValueObject<T> right) =>
        Equals(left, right);

    public static bool operator !=(ValueObject<T> left, ValueObject<T> right) =>
        !Equals(left, right);
}

Let’s test:

class Program
{
    static void Main(string[] args)
    {
        using (IPingMonitor m = new PingMonitor(new Clock(), new SessionLogWriter()))
        {
            Task.Run(async () =>
            {
                while (true)
                {
                    await m.PingAsync(1);
                    await Task.Delay(1000);
                }
            });

            Task.Run(async () =>
            {
                while (true)
                {
                    await m.PingAsync(10);
                    await Task.Delay(4000);
                }
            });

            Console.ReadLine();
        }
    }
}

Where:

class SessionLogWriter : ISessionMonitor
{
    public SessionLogWriter()
        : this(Console.Out)
    {
    }

    public SessionLogWriter(TextWriter writer)
    {
        _writer = writer;
    }

    readonly TextWriter _writer;

    public void Dispose()
    {
        _writer.Dispose();
    }

    public Task StartAsync(Session session) =>
        _writer.WriteLineAsync(
            $"{session.AgentId}\t" +
            $"{session.Started:hh:mm:ss}\t" +
            $"{session.LastActivity:hh:mm:ss}\tTrue");

    public Task EndAsync(Session session) =>
        _writer.WriteLineAsync(
            $"{session.AgentId}\t" +
            $"{session.Started:hh:mm:ss}\t" +
            $"{session.LastActivity:hh:mm:ss}\tFalse");
}

I would recommend to create composite ISessionMonitor to broadcast your changes between multiple ISessionMonitor implementations if needed.