Receive concurrent asynchronous requests and process them one at a time

Question

Background

We have a service operation that can receive concurrent asynchronous requests and must process those requests one at a time.

In the following example, the UploadAndImport(...) method receives concurrent requests on multiple threads, but its calls to the ImportFile(...) method must happen one at a time.

Layperson Description

Imagine a warehouse with many workers (multiple threads). People (clients) can send the warehouse many packages (requests) at the same time (concurrently). When a package comes in a worker takes responsibility for it from start to finish, and the person who dropped off the package can leave (fire-and-forget). The workers' job is to put each package down a small chute, and only one worker can put a package down a chute at a time, otherwise chaos ensues. If the person who dropped off the package checks in later (polling endpoint), the warehouse should be able to report on whether the package went down the chute or not.

Question

The question then is how to write a service operation that...

1. can receive concurrent client requests,
2. receives and processes those requests on multiple threads,
3. processes requests on the same thread that received the request,
4. processes requests one at a time,
5. is a one way fire-and-forget operation, and
6. has a separate polling endpoint that reports on request completion.

We've tried the following and are wondering two things:

1. Are there any race conditions that we have not considered?
2. Is there a more canonical way to code this scenario in C#.NET with a service oriented architecture (we happen to be using WCF)?

Example: What We Have Tried?

This is the service code that we have tried. It works though it feels like somewhat of a hack or kludge.

static ImportFileInfo _inProgressRequest = null;

static readonly ConcurrentDictionary<Guid, ImportFileInfo> WaitingRequests = 
    new ConcurrentDictionary<Guid, ImportFileInfo>();

public void UploadAndImport(ImportFileInfo request)
{
    // Receive the incoming request
    WaitingRequests.TryAdd(request.OperationId, request);

    while (null != Interlocked.CompareExchange(ref _inProgressRequest, request, null))
    {
        // Wait for any previous processing to complete
        Thread.Sleep(500);
    }

    // Process the incoming request
    ImportFile(request);

    Interlocked.Exchange(ref _inProgressRequest, null);
    WaitingRequests.TryRemove(request.OperationId, out _);
}

public bool UploadAndImportIsComplete(Guid operationId) => 
    !WaitingRequests.ContainsKey(operationId);

This is example client code.

private static async Task UploadFile(FileInfo fileInfo, ImportFileInfo importFileInfo)
{
    using (var proxy = new Proxy())
    using (var stream = new FileStream(fileInfo.FullName, FileMode.Open, FileAccess.Read))
    {
        importFileInfo.FileByteStream = stream;
        proxy.UploadAndImport(importFileInfo);
    }

    await Task.Run(() => Poller.Poll(timeoutSeconds: 90, intervalSeconds: 1, func: () =>
    {
        using (var proxy = new Proxy())
        {
            return proxy.UploadAndImportIsComplete(importFileInfo.OperationId);
        }
    }));
}

It's hard to write a minimum viable example of this in a Fiddle, but here is a start that give a sense and that compiles.

As before, the above seems like a hack/kludge, and we are asking both about potential pitfalls in its approach and for alternative patterns that are more appropriate/canonical.

Answer 1

Simple solution using Producer-Consumer pattern to pipe requests in case of thread count restrictions.

You still have to implement a simple progress reporter or event. I suggest to replace the expensive polling approach with an asynchronous communication which is offered by Microsoft's SignalR library. It uses WebSocket to enable async behavior. The client and server can register their callbacks on a hub. Using RPC the client can now invoke server side methods and vice versa. You would post progress to the client by using the hub (client side). In my experience SignalR is very simple to use and very good documented. It has a library for all famous server side languages (e.g. Java).

Polling in my understanding is the totally opposite of fire-and-forget. You can't forget, because you have to check something based on an interval. Event based communication, like SignalR, is fire-and-forget since you fire and will get a reminder (cause you forgot). The "event side" will invoke your callback instead of you waiting to do it yourself!

Requirement 5 is ignored since I didn't get any reason. Waiting for a thread to complete would eliminate the fire and forget character.

private BlockingCollection<ImportFileInfo> requestQueue = new BlockingCollection<ImportFileInfo>();
private bool isServiceEnabled;
private readonly int maxNumberOfThreads = 8;
private Semaphore semaphore = new Semaphore(numberOfThreads);
private readonly object syncLock = new object();

public void UploadAndImport(ImportFileInfo request) 
{            
  // Start the request handler background loop
  if (!this.isServiceEnabled)
  {
    this.requestQueue?.Dispose();
    this.requestQueue = new BlockingCollection<ImportFileInfo>();

    // Fire and forget (requirement 4)
    Task.Run(() => HandleRequests());
    this.isServiceEnabled = true;
  }

  // Cache multiple incoming client requests (requirement 1) (and enable throttling)
  this.requestQueue.Add(request);
}

private void HandleRequests()
{
  while (!this.requestQueue.IsCompleted)
  {
    // Wait while thread limit is exceeded (some throttling)
    this.semaphore.WaitOne();

    // Process the incoming requests in a dedicated thread (requirement 2) until the BlockingCollection is marked completed.
    Task.Run(() => ProcessRequest());
  }

  // Reset the request handler after BlockingCollection was marked completed
  this.isServiceEnabled = false;
  this.requestQueue.Dispose();
}

private void ProcessRequest()
{
  ImportFileInfo request = this.requestQueue.Take();
  UploadFile(request);

  // You updated your question saying the method "ImportFile()" requires synchronization.
  // This a bottleneck and will significantly drop performance, when this method is long running. 
  lock (this.syncLock)
  {
    ImportFile(request);
   }

  this.semaphore.Release();
}

Remarks:

• BlockingCollection is a IDisposable
• TODO: You have to "close" the BlockingCollection by marking it completed: "BlockingCollection.CompleteAdding()" or it will loop indeterminately waiting for further requests. Maybe you introduce a additional request methods for the client to cancel and/ or to update the process and to mark adding to the BlockingCollection as completed. Or a timer that waits an idle time before marking it as completed. Or make your request handler thread block or spin.
• Replace Take() and Add(...) with TryTake(...) and TryAdd(...) if you want cancellation support
• Code is not tested
• Your "ImportFile()" method is a bottleneck in your multi threading environment. I suggest to make it thread safe. In case of I/O that requires synchronization, I would cache the data in a BlockingCollection and then write them to I/O one by one.

Answer 2

The problem is that your total bandwidth is very small-- only one job can run at a time-- and you want to handle parallel requests. That means that queue time could vary wildly. It may not be the best choice to implement your job queue in-memory, as it would make your system much more brittle, and more difficult to scale out when your business grows.

A traditional, scaleable way to architect this would be:

• An HTTP service to accept requests, load balanced/redundant, with no session state.
• A SQL Server database to persist the requests in a queue, returning a persistent unique job ID.
• A Windows service to process the queue, one job at a time, and mark jobs as complete. The worker process for the service would probably be single-threaded.

This solution requires you to choose a web server. A common choice is IIS running ASP.NET. On that platform, each request is guaranteed to be handled in a single-threaded manner (i.e. you don't need to worry about race conditions too much), but due to a feature called thread agility the request might end with a different thread, but in the original synchronization context, which means you will probably never notice unless you are debugging and inspecting thread IDs.

Answer 3

Given the constraints context of our system, this is the implementation we ended up using:

static ImportFileInfo _importInProgressItem = null;

static readonly ConcurrentQueue<ImportFileInfo> ImportQueue = 
    new ConcurrentQueue<ImportFileInfo>();

public void UploadAndImport(ImportFileInfo request) {
    UploadFile(request);
    ImportFileSynchronized(request);
}

// Synchronize the file import, 
// because the database allows a user to perform only one write at a time.
private void ImportFileSynchronized(ImportFileInfo request) {
    ImportQueue.Enqueue(request);
    do {
        ImportQueue.TryPeek(out var next);
        if (null != Interlocked.CompareExchange(ref _importInProgressItem, next, null)) {
            // Queue processing is already under way in another thread.
            return;
        }

        ImportFile(next);
        ImportQueue.TryDequeue(out _);
        Interlocked.Exchange(ref _importInProgressItem, null);
    }
    while (ImportQueue.Any());
}

public bool UploadAndImportIsComplete(Guid operationId) =>
    ImportQueue.All(waiting => waiting.OperationId != operationId);

This solution works well for the loads we are expecting. That load involves a maximum of about 15-20 concurrent PDF file uploads. The batch of up to 15-20 files tends to arrive all at once and then to go quiet for several hours until the next batch arrives.

Criticism and feedback is most welcome.