1
\$\begingroup\$

We currently have two legacy systems: the Consumer and the Worker. These systems are massively complex in ways that are not important to this review, but it is enough to say that a large-scale re-development of these systems is not currently possible.

The Consumer communicates with the Worker via http request/response. The request/response nature of this setup is critical and non-negotiable; this is really the whole purpose of this review. There is no opportunity to use (for example) notifications or events to solve this problem. The Worker's task of processing the request may be long (or longish) running, and returns a response when complete.

So the current state could not be simpler:

1. Consumer sends a request to the Worker via POST
2. Worker receives the request and does the necessary work
3. Worker returns a response to the Consumer

Now, the problem: the actual work that the Worker does has resource constraints, and allowing an unlimited number of "jobs" to be processed concurrently is problematic. What we need to do is introduce a way to queue up requests and feed them to the Worker in a measured, predictable way.

Our solution is to introduce a "Queue Service" in between the Consumer and the Worker. This will be a physical, singular service that acts as a pass-through, with its only other responsibility being to manage the Worker's load.

The actual implementation of the queuing mechanism is not important for this question, but a few important requirements:

1. The Queue Service will consist of custom code to queue, prioritize and feed the work to the Worker (out-of-box solutions like Gateways are not viable options).
2. The Queue Service will be asynchronous to avoid thread starvation when there are many http requests in flight.

So the desired state would be:

1. Consumer sends a request to the Queue Service via POST
2. Queue Service receives the request
3. Queue Service sends a request to the Worker via POST (immediately or delayed depending on the state of the queue) 
4. Worker receives the request and does the necessary work
5. Worker returns a response to the Queue Service
6. Queue Service returns a response to the Consumer

To be clear, the idea is that the request remains active even if the work gets put into queue and has to wait.

In the code below, I am trying to prove that this is possible by signaling using the TaskCompletionSource construct. There are all kinds of shortcuts/simplifications in this code--the main question is whether this technique is viable to keep the request alive until a separate process signals its completion (and provides a result).

[TestFixture]
public class ValidateAsyncContinuation
{
    [Test]
    public async Task RunAsync()
    {
        //start this as a background task (fire and forget)
        PretendQueueReader.StartProcessing();

        var sw = Stopwatch.StartNew();

        const int feedDelayInMs = 1;
        const int numToProcess = 1000;

        var tasks = new List<Task>();

        //this simulates x number of requests sent via http and waiting for a response, spaced out by a 1ms delay
        for (var x = 0; x < numToProcess; x++)
        {
            var x1 = x; //to avoid "access to modified closure" warning
            var task = Task.Run(async () =>
            {
                await PretendController.Post(new Request($"request {x1}"), x1);
                Debug.WriteLine($"Completed {x1}.");
            });
            tasks.Add(task);
            Thread.Sleep(feedDelayInMs);
        }

        await Task.WhenAll(tasks);
        Debug.WriteLine($"took {sw.ElapsedMilliseconds}ms");
        PretendQueueReader.Stop();
    }
}

//background process that periodically tells the PretendQueue to check to see if there is work to be done
public static class PretendQueueReader
{
    private static bool _stop;

    public static async Task StartProcessing()
    {
        while (!_stop)
        {
            await PretendQueue.ProcessQueue();
            await Task.Delay(1); //run every 1ms.  Not having any delay spikes the CPU.
        }
    }

    public static void Stop()
    {
        _stop = true;
    }
}

public static class PretendController
{
    //the controller passes the request along to the queue and waits for a response
    public static async Task<Response> Post(Request request, int index)
    {
        return await PretendQueue.QueueRequest(request, index);
    }
}

public static class PretendQueue
{
    private static readonly Queue<RequestCompletion> Queue = new Queue<RequestCompletion>();
    private static int _inProcess;
    private const int MaxConcurrent = 100;
    private static readonly SemaphoreSlim SemaphoreSlim = new SemaphoreSlim(1,1);

    public static async Task<Response> QueueRequest(Request request, int index)
    {
        var responseSignal = new TaskCompletionSource<Response>();

        //lock the queue when accessing and release when complete.
        await SemaphoreSlim.WaitAsync();
        try
        {
            Queue.Enqueue(new RequestCompletion(request, responseSignal));
            Debug.WriteLine($"Queuing Request #{index}:  {_inProcess} requests are in process and {Queue.Count} requests are in queue.");
        }
        finally
        {
            SemaphoreSlim.Release();
        }

        return await responseSignal.Task;
    }

    public static async Task ProcessQueue()
    {
        RequestCompletion requestCompletion;
        //lock the queue when accessing and release when complete.
        await SemaphoreSlim.WaitAsync();
        try
        {
            if (Queue.Count == 0 || _inProcess >= MaxConcurrent) return;
            Debug.WriteLine($"DE-Queuing a Request:  {_inProcess} requests are in process and {Queue.Count} requests are in queue.");
            requestCompletion = Queue.Dequeue();
        }
        finally
        {
            SemaphoreSlim.Release();
        }

        //fire and forget
        Run(requestCompletion.Request, requestCompletion.ResponseCompletionSource);
    }

    private static async Task<Response> Run(Request request, TaskCompletionSource<Response> responseSignal)
    {
        await SemaphoreSlim.WaitAsync();
        _inProcess++;
        SemaphoreSlim.Release();

        //Fire and forget.
        PretendWorker.ProcessAndSignal(request, responseSignal);

        var response = await responseSignal.Task;

        await SemaphoreSlim.WaitAsync();
        _inProcess--;
        SemaphoreSlim.Release();

        return response;
    }
}

public static class PretendWorker
{
    public static async Task ProcessAndSignal(Request request, TaskCompletionSource<Response> responseCompletionSource)
    {
        var start = DateTime.Now;

        //this simulates the call to the worker (taking 2 seconds)
        await Task.Delay(TimeSpan.FromSeconds(2));

        responseCompletionSource.SetResult(new Response(start, DateTime.Now));
    }
}

public class RequestCompletion
{
    public RequestCompletion(Request request, TaskCompletionSource<Response> responseCompletionSource)
    {
        Request = request;
        ResponseCompletionSource = responseCompletionSource;
    }

    public Request Request { get; }
    public TaskCompletionSource<Response> ResponseCompletionSource { get; }
}

public class Request
{
    public Request(string name)
    {
        Name = name;
    }

    public string Name { get; }
}

public class Response
{
    public Response(DateTime startedTime, DateTime completedTime)
    {
        StartedTime = startedTime;
        CompletedTime = completedTime;
    }

    public DateTime StartedTime { get; }
    public DateTime CompletedTime { get; }
}
\$\endgroup\$
  • 1
    \$\begingroup\$ I would be concerned over timeouts. Http servers usually have timeouts/request have timeouts/clients have timeouts. They are all usually customizable so you would need to make sure they are all accounted for. Just having a quest pending doesn't stop say iis from timing out your request. Packages like SignalR are designed to keep a long running connection with fall backs. Might be worth checking if you can use that possibly. \$\endgroup\$ – CharlesNRice Apr 25 at 13:24
  • \$\begingroup\$ @CharlesNRice timeouts are indeed a consideration that we will need mitigate carefully. \$\endgroup\$ – Phil Sandler Apr 25 at 13:30

Your Answer

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

Browse other questions tagged or ask your own question.