sequential command processing with an async io cloud upsert

Question

We are new to c# and still trying to grok the async idioms.

We have a windows service that requires us to iterate a list of results queried from a PC database to feed the parse cloud server (which only supports async reads and writes). The results often have to be overridden with new results.

In a legacy java sdk, we simply did:

For each result:

do synchronous find
if not found
 create a new object/document
else
 Update the existing object/document
Save synchronously

and had no problems.

BTW - our server needs to feed Parse in seconds, not microseconds, though sub second performance would not hurt!

Our current production system used lock() to ensure the read/write stayed synchronized however we suspect deadlocks are occurring. A telling fact is that restarting the service processes the command queue just fine!!

Our proposed fix looks like this:

  private readonly SemaphoreSlim _mutex = new SemaphoreSlim(1);

  public async Task ProcessPostedCommandsAsync()
    {
        using (var showContext = new ShowContext())
        {
            var retryCnt = 0;
            var maxRetries = 8;


            var commands = FindCommands(showContext);

            foreach (var command in commands)
            {

                await _mutex.WaitAsync();
                try
                {
                     // read data from the cloud, 
                     // based on data from the cloud, write back to the cloud
                     // mark the command as processed in the database
                     await ProcessCommandAsync(showContext, command);
                }
                catch (Exception e)
                {
                    GlobalLogger.Logger.Error(e,
                        $"assinging executing task, Exception in Process Commands pkid = {command.Pkid} ");
                    GlobalLogger.Logger.Error($"Retry count = {retryCnt}");
                    retryCnt++;
                    await Task.Delay(2000);
                    if (retryCnt == maxRetries)
                    {
                        break; // from for
                    }
                }
                finally
                {
                    _mutex.Release();

                }
            }
        }
    }

We have since modified our code to hopefully address the deadlock problem using the SemaphoreSlim.

With only one command in the database, let's call it Command-A, duplication seems to occur when the command reading loop re-reads Command-A even though Command-A was has already kicked off its cloud read/write.

Everywhere we look says, “DO NOT BLOCK or you will get deadlocks” and says go async/await all the way. But if we don’t block in some way we will get race conditions and duplicates.

So we added locks and we get what seem like deadlocks after hours of processing commands just fine.

This seems like a pretty common use case.

1) Is there an c# idiom we are missing here that loops through a command list to write asynchronously to the cloud but in sequential order without sending duplicates and not creating deadlocks? We really just want a way to wait for each synchronized command to finish.

2) How can we test this in such a way to prove it works in varying environments.

We really feel like in a Catch 22 situation. We need to block to wait for the command to fully process while yet we should not block async code.

Note: We've looked at the following resources and are looking for direction so far.

Concurancy in C# Cookbook by Stephan Cleary

Async/Await Best Practices

Don't Block on Async Code

Answer 1

Basically, asyncronous tasks behave not much differently than threads, and most the thread intuition should apply to them as well. But you should not use threading primitives (such as locking) which may lock it. If you need to synchronize asyncronous tasks, you can acquire locks asynchnonously: https://blog.cdemi.io/async-waiting-inside-c-sharp-locks/

Actually, if I understand it correctly, as a start, you could run all your code as a single task:

public async Task StartProcessing()
{
    DisableStartProcessingUI();
    try {
        foreach (result in results) {
            if(TryFind(result, out Document document)) {
                await Update(document);
            } else {
                await CreateDocumentFromResulit(result);
            }
        }
    } finally {
        EnableStartProcessingUI();
    }
}

This runs basically synchronously, and blocks UI which is used to start processing, so that you don't run it again while first request is still executing. So you don't even have to lock it in this case. But if you want to use concurrent requests to Update() and CreateDocumentFromResulit() then you'd need to use some semaphores as above to avoid accessing same documents.

Edit: In your example, should probably be "await ProcessCommandAsync(showContext, command);"