Prevent web socket consumer code from being executed twice

Question

Imagine we consume a web socket with really high fast finance data. In peak times the web socket method is called hundreds to thousands times per second.

We have a condition in our web socket method which goes from time to time to true. And in this case another method should be called. But only once. Due to the execution speed of the method it is really hard to prevent double execution. The code looks like this:

private readonly ConcurrentDictionary<string, bool> _inExecution = new ConcurrentDictionary<string, bool>();

private void SubscribeToSocket()
{
    _socket.Connect();

    var subscription = SocketSubscriptions.CreateSubsription(data =>
    {
        Task.Run(async () =>
        {
            // read data

            if (condition)
            {
                // call method only once
                await Execute(key);

                condition = false;
            }
        }
    }
}

private async Task Execute(string key)
{
    // Even with this statement the code calls are too fast and sometimes gets executed twice
    if (!_inExecution[key])
    {
        _inExecution[key] = true;

        // do something..
    }
}

I already tried to prevent a double execution with a random wait before the Execute() method. Like this:

if (condition)
{
    var rnd = new Random();
    await Task.Delay(rnd.Next(15, 115));

    // call method only once
    await Execute(key);

    condition = false;
}

But even this got executed twice in some special cases. Is there a better way to prevent that?

Use a Mutex? or other concurrency management tool: https://learn.microsoft.com/en-us/dotnet/api/system.threading.mutex?view=netframework-4.8 — Milney, Jun 05 '19 at 20:28
It looks as though, as Milney suggests, you need some form of locking. Try `SemaphoreSlim` https://stackoverflow.com/a/45769160/1481699 — Prime, Jun 05 '19 at 20:43
note: SemaphoreSlim has some **nasty glitches** in edge cases when used from both sync and async code... I have a MutexSlim that was designed specifically to avoid them, if it might help... but yeah, this is a scenario that screams some kind of semaphore/mutex setup — Marc Gravell, Jun 05 '19 at 21:05
@MarcGravell sure, everything would help. Do you have a link to additional infos or a github repo with sample code? — t2t, Jun 05 '19 at 21:23

score 1 · Accepted Answer · answered Jun 06 '19 at 07:23

The key race condition here appears to be the race between checking _inExecution[key] and *updating_inExecution[key] = true'; multiple callers can get through there. There are multiple ways of making this robust, but in your case, after consideration, I'm pretty sure that the simplest approach is simply to synchronize over the collection, i.e.

    private readonly HashSet<string> _inExecution = new HashSet<string>();
    private async Task Execute(string key)
    {
        // Even with this statement the code calls are too fast and sometimes gets executed twice
        bool haveLock = false;
        try
        {
            lock(_inExecution) { haveLock = _inExecution.Add(key); }
            if (haveLock)
            {
                // ... your code here
            }
        }
        finally
        {
            if (haveLock)
            {
                lock (_inExecution) _inExecution.Remove(key);
            }
        }
    }

You could also use a Dictionary<string, bool>, but a HashSet<string> works fine here. A Dictionary<string, bool> may avoid some keyspace overhead, though - just manipulating the value instead - something like:

    private readonly Dictionary<string, bool> _inExecution = new Dictionary<string, bool>();
    private async Task Execute(string key)
    {
        // Even with this statement the code calls are too fast and sometimes gets executed twice
        bool haveLock = false;
        try
        {
            lock(_inExecution)
            {
                if (!_inExecution.TryGetValue(key, out var state) || !state)
                {   // if missing entirely, or not currently held: take it
                    haveLock = _inExecution[key] = true;
                }
            }
            if (haveLock)
            {
                // ... your code here
            }
        }
        finally
        {
            if (haveLock)
            {
                lock (_inExecution) _inExecution[key] = false;
            }
        }
    }

The important thing to note is that you don't keep the lock over the actual // ... your code here bit - that would prevent all concurrent executions, which isn't what you want.

If you want to tidy it up, there are ways to structure this with custom disposables etc, but try/finally works fine.

Works great with the HashSet. Thanks, Marc! – t2t Jun 12 '19 at 20:15 — t2t, Jun 12 '19 at 20:15

Prevent web socket consumer code from being executed twice

1 Answers1