4
\$\begingroup\$

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.

\$\endgroup\$
3
\$\begingroup\$

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.

\$\endgroup\$
14
  • \$\begingroup\$ Nice answer. I would remove the parameterless constructor in SessionLogWriter. In that way you can omit the Dispose and IDisposable because an object which is passed to the constructor which takes a TextWriter should be disposed by the object which created it. Otherwise +1 \$\endgroup\$ – Heslacher May 31 '16 at 8:33
  • \$\begingroup\$ Thanks. As for Dispose - Microsoft suggests it for all the streaming classes: CONSIDER implementing the Basic Dispose Pattern on classes that themselves don’t hold unmanaged resources or disposable objects but are likely to have subtypes that do. \$\endgroup\$ – Dmitry Nogin May 31 '16 at 20:31
  • \$\begingroup\$ Thanks for your time, Moitry. It's seems pretty cool. I really like concept of AgentId and ValueObject. I'll try to propagate AgentId like structs to EF using [ComplexType]. Can you explain why do we need to put null into dictionary before deletion? I'm using special overload of remove, that takes KeyValuePair, that checks both Key and value, if value!=stored it returns false. it utilises internal method TryRemoveInternal(keyValuePair.Key, out obj, true, keyValuePair.Value);. \$\endgroup\$ – Bogdan Mart May 31 '16 at 20:46
  • \$\begingroup\$ I'm not sure why MS didn't made that signature public, but if cast to ICollection, we can remove by KV pair. So it removes atmicaly at one call, why do we need to make 3 calls? Idea with session object is great too, we can store more useful info. I've learned many new, but please clarify with Remove. \$\endgroup\$ – Bogdan Mart May 31 '16 at 20:48
  • \$\begingroup\$ How slow is ValueObject is, compared to regular objects, with manual equals? it's seems to create some overhead, but looks really nice. \$\endgroup\$ – Bogdan Mart May 31 '16 at 20:49

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.