6
\$\begingroup\$

UPDATE: I found the namespace Channels, and this does exactly what I want to do, but in a much more stable way. Please don't use my code for anything important! :)

Use case

I have a site that (at peak time) will serve 1000 requests per second. Each request should trigger a record in an SQL Server database table.

Requirements

  1. The request must not fail, even though the writing to the database could fail.
  2. The user should not wait for the writing to the database
  3. Writing to the database should happen as fast as possible, still respecting 1. and 2. Writing should be allowed to happen in bulk, even though records are added individually to the queue.

My approach

My approach aims at using a background queue with a single thread, that will create the records in the database in bulk (using table valued parameters), which can perform 10.000+ writes per second. Using a single thread will prevent SQL connection pool starvation and local TCP port starvation. Furthermore SQL Server will perform much better compared to writing single lines individually.

Usage of the Memory Queue

I have made a general implementation that is not bound to the specific use case with a database writing of visits.

I am using ASP.net 5 but setting up the queue and injecting it in the controller will be left out here. I will just stick to the raw usage of the queue.

Creating a queue

var memoryQueue = new MemoryQueue<MyQueueElementType>(
                      handler: elementList => { */ elementList will contain 1....maxBulkSize elements /* },
                      maxBulkSize: 10);
memoryQueue.StartListening(); // This will start the internal thread

Actual case with writing to a database

var memoryQueue = new MemoryQueue<Visit>(
                      handler: elementList => visitRepo.WriteInBulk(elementList),
                      maxBulkSize: 10);
memoryQueue.StartListening(); // This will start the internal thread

Adding to the queue

memoryQueue.AddAsync(new Visit(ip: "1.2.3.4")); // The visit details have been left out to avoid noise.

The AddAsync call will return a task that completes, once the added item is part of a bulk handling. In the actual web calls, these tasks will not be awaited due to requirement 2, but it is still nice to be able to await the tasks when timing unit-test actions.

The implementation

And now to the thing that I would really like people to comment on; the implementation. I have using a BlockingCollection as my queue internally and a single thread reacting when at least 1 element is in this queue:

Interface

public interface IMemoryQueue<T> : IDisposable
{
    void StartListening();
    Task AddAsync(T element);
    Task ShutdownAsync();
}

Class

public class MemoryQueue<T> : IMemoryQueue<T>
{
    private readonly BlockingCollection<MemoryElementWrapper<T>> _queue;
    private readonly Action<IEnumerable<T>> _handler;
    private readonly int _maxBulkSize;
    private readonly ICommonLog _log;
    private readonly Type _messageType;
    public ThreadState InternalThreadState => _thread?.ThreadState ?? ThreadState.Unstarted;
    private Thread _thread = null;

    private TaskCompletionSource<object> _threadCompleteTaskSource = new TaskCompletionSource<object>();
    private CancellationTokenSource _disposeCancelTokenSource = new CancellationTokenSource();
    public MemoryQueue(Action<IEnumerable<T>> handler, int maxBulkSize, ICommonLogProvider logProvider)
    {
        _handler = handler ?? throw new ArgumentNullException(nameof(handler));
        _maxBulkSize = maxBulkSize;
        _queue = new BlockingCollection<MemoryElementWrapper<T>>();
        _log = logProvider.GetLog(GetType());
        _messageType = typeof(T);
    }

    public void StartListening()
    {
        if (_thread != null)
            throw new ArgumentException("Start listening may only happen once");

        // This call will block until the thread has actually started to ensure MemoryQueue.InternalThreadState to always return 'Running' immediately after calling this method.
        var threadStartEvent = new ManualResetEvent(initialState: false);

        _thread = new Thread(new ThreadStart(() =>
        {
            // Signal that thread has now started. This is used for the "StartListening" call to block until it is certain that thread thread has actually been started.
            threadStartEvent.Set();
            try
            {
                _log.Warn($"Thread processing messages of type {_messageType.FullName} has started");

                while (!_queue.IsCompleted)
                {
                    // FetchAtLeastOneBlocking will block until at least 1 element can be fetched. If more element are available, up to _maxBulkSize will be returned.
                    List<MemoryElementWrapper<T>> elementWrappers = _queue.TakeAtLeastOneBlocking(maxCount: _maxBulkSize, log: _log, cancellationToken: _disposeCancelTokenSource.Token);
                    
                    // Get elements
                    var elements = elementWrappers.Select(wrapper => wrapper.Element).ToList();

                    // process
                    try
                    {
                        _handler(elements);
                        CompleteTasks(elements: elementWrappers);
                    }
                    catch (Exception ex)
                    {
                        _log.Fatal($"Unexpected exception processing {elements.Count} of type {_messageType.FullName}. Entry data: {JsonSerializeSafe(elements)}", ex);
                        FaultTasks(elements: elementWrappers, exceptionToSetOnTasks: ex);
                    }
                }

                // Can reach this in case of queue ended
            }
            catch (InvalidOperationException ex)
            {
                // Queue has been completed

                // Sanity check and alarm if needed
                if(_queue.Count > 0)
                {
                    _log.Fatal($"InvalidOperationException thrown, but {_queue.Count} elements still in queue. This is unexpected behaviour.");
                    CompleteQueueAndFaultAllPendingTasks(exceptionToSetOnTasks: ex);
                }
            }
            catch (OperationCanceledException ex)
            {
                // Dispose has been called
                CompleteQueueAndFaultAllPendingTasks(exceptionToSetOnTasks: ex);
            }
            catch(Exception ex)
            {
                // Totally unexpected error happened in queue loop logic, unrelated to the handler.
                // This is a unrecoverable case and requires the memory queue to be restarted (by restarting the entire application)
                _log.Fatal($"Internal thread in MemoryQueue {GetType().FullName} experienced an unexpected exception. The thread is no longer running. This can only be recovered by restarting the entire application.", ex);
                CompleteQueueAndFaultAllPendingTasks(exceptionToSetOnTasks: ex);
            }
            finally
            {
                _threadCompleteTaskSource.SetResult(null);
                _queue.Dispose();
                _log.Warn($"Thread processing messages of type {_messageType.FullName} has stopped");
            }

            // Whenever the thread is done, if any elements are still on the queue, the corresponding tasks should be notified to not let any awaiters hang.
        }));

        _thread.Start();

        // Wait for thread to actually spin up
        threadStartEvent.WaitOne();
    }

    private void CompleteTasks(List<MemoryElementWrapper<T>> elements)
    {
        foreach (var element in elements)
        {
            element.TaskCompletionSource.SetResult(true);
        }
    }
    private void FaultTasks(List<MemoryElementWrapper<T>> elements, Exception exceptionToSetOnTasks)
    {
        foreach (var element in elements)
        {
            element.TaskCompletionSource.SetException(exceptionToSetOnTasks);
        }
    }

    private void CompleteQueueAndFaultAllPendingTasks(Exception exceptionToSetOnTasks)
    {
        if(!_queue.IsAddingCompleted)
            _queue.CompleteAdding();

        FaultTasks(elements: _queue.TakeAvailableNonBlocking(maxCount: int.MaxValue, log: _log), exceptionToSetOnTasks: exceptionToSetOnTasks);
    }

    private string JsonSerializeSafe(object obj)
    {
        try
        {
            return JsonConvert.SerializeObject(obj);
        }catch(Exception ex)
        {
            _log.Fatal("Unexpected error serializing object", ex);
            return $"(Unexpected error serializing object. Exception message: {ex.Message})";
        }
    }

    public Task AddAsync(T element)
    {
        // Due to race conditions, either the thread state will not be running or the queue has been completed but the thread is not yet done. In this case, this particular guard will pass but the Add call below will throw and InvalidOperationException
        if (InternalThreadState != ThreadState.Running)
            throw new AddNotAllowedException($"Cannot add element as the state of the queue is {InternalThreadState}");

        var wrapper = new MemoryElementWrapper<T>(item: element);
        
        try
        {
            _queue.Add(wrapper);
        }
        catch (InvalidOperationException)
        {
            throw new AddNotAllowedException($"Queue is no longer accepting new elements");
        }
        
        return wrapper.TaskCompletionSource.Task;
    }

    public Task ShutdownAsync()
    {
        _queue.CompleteAdding(); // Tells the thread that no more elements will be added. This will trigger the thread to stop when the queue is empty
        return _threadCompleteTaskSource.Task;
    }

    public void Dispose()
    {
        // Signal to the listening thread to stop everything. Calling dispose on the queue directly here proved to be subject to deadlocks where the waiting thread would just hang forever.
        _disposeCancelTokenSource.Cancel();
    }

    public Task AwaitThreadToStopRunning_ForUseInUnitTestsOnlyAsync()
    {
        return _threadCompleteTaskSource.Task;
    }
}

public class AddNotAllowedException : Exception
{
    public AddNotAllowedException(string message) : base(message)
    {
    }
}

public class MemoryElementWrapper<T>
{
    public TaskCompletionSource<object> TaskCompletionSource { get; set; }
    public T Element { get; set; }

    public MemoryElementWrapper(T item)
    {
        TaskCompletionSource = new TaskCompletionSource<object>();
        Element = item;
    }
}

Unit tests

    public class MemoryQueueTest : UnitTestBase
    {
        [Fact]
        public void StartListening_CallTwice_ExpectException_Test()
        {
            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(messages => { }, maxBulkSize: 5, LogProvider))
            {
                memoryQueue.StartListening();
                Assert.Throws<ArgumentException>(() => memoryQueue.StartListening());
                Assert.Throws<ArgumentException>(() => memoryQueue.StartListening());
            }
        }

        [Fact]
        public void AddAsync_WithoutListenerStarted_ExpectException_Test()
        {
            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(messages => { }, maxBulkSize: 5, LogProvider))
            {
                Action action = () => memoryQueue.AddAsync(new UnitTestMessage("some val"));
                Assert.Throws<AddNotAllowedException>(action);
            }
        }

        [Fact]
        public void AddAsync_AfterDisposed_ExpectException_Test()
        {
            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(messages => { }, maxBulkSize: 5, LogProvider))
            {
                Action action = () => memoryQueue.AddAsync(new UnitTestMessage("some val"));
                memoryQueue.Dispose();

                Assert.Throws<AddNotAllowedException>(action);
            }
        }

        [Fact]
        public void AddAsync_AfterShutdown_ExpectExceptionInSyncCall_Test()
        {
            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(elements => { }, maxBulkSize: 1, LogProvider))
            {
                memoryQueue.StartListening();
                var task = memoryQueue.ShutdownAsync();
                Assert.Throws<AddNotAllowedException>((Action)(() => memoryQueue.AddAsync(new UnitTestMessage("val1"))));
            }
        }

        [Fact]
        public async void Dispose_EnsureQueueIsStoppedAndThreadIsStopped_AndAllPendingTasksWillBeFaulted_Test()
        {
            var msg1 = new UnitTestMessage("1");
            var msg2 = new UnitTestMessage("2");
            var msg3 = new UnitTestMessage("3");
            var msg4 = new UnitTestMessage("4");
            var msg5 = new UnitTestMessage("5");

            var handlerStartEvent = new ManualResetEvent(initialState: false);
            var handlerEndEvent = new ManualResetEvent(initialState: false);

            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(messages => { handlerStartEvent.Set(); handlerEndEvent.WaitOne(); }, maxBulkSize: 2, LogProvider))
            {
                memoryQueue.StartListening();

                // Add 5 messages
                var task1 = memoryQueue.AddAsync(msg1);
                var task2 = memoryQueue.AddAsync(msg2);
                var task3 = memoryQueue.AddAsync(msg3);
                var task4 = memoryQueue.AddAsync(msg4);
                var task5 = memoryQueue.AddAsync(msg5);

                // Wait until first batch is executed
                handlerStartEvent.WaitOne();

                // Now we know that the handler is being called and the worker thread is being blocked.
                // At this point we call dispose and then ensure that current tasks are completed but following tasks are faulted
                memoryQueue.Dispose();

                // After disposing we allow the handler to proceed
                handlerEndEvent.Set();

                await task1.ThrowIfTimeout(timeout: TimeSpan.FromSeconds(5));
                await task2.ThrowIfTimeout(timeout: TimeSpan.FromSeconds(5));

                await Assert.ThrowsAsync<OperationCanceledException>(() => task3.ThrowIfTimeout(TimeSpan.FromSeconds(5)));
                await Assert.ThrowsAsync<OperationCanceledException>(() => task4.ThrowIfTimeout(TimeSpan.FromSeconds(5)));
                await Assert.ThrowsAsync<OperationCanceledException>(() => task5.ThrowIfTimeout(TimeSpan.FromSeconds(5)));

                // Now we will busy loop until the internal thread expectedly ends up in 'Stopped' state.
                var stopwatch = System.Diagnostics.Stopwatch.StartNew();
                while (ThreadState.Stopped != memoryQueue.InternalThreadState)
                {
                    await Task.Delay(TimeSpan.FromMilliseconds(100));
                    if(stopwatch.Elapsed > TimeSpan.FromSeconds(5))
                        throw new Exception("Thread blocking state: " + ThreadState.Stopped);
                }
            }
        }

        [Fact]
        public async void BeginShutdown_EnsureQueueIsEmptiedAndThenThreadIsStopped_Test()
        {
            var msg1 = new UnitTestMessage("val1");
            var msg2 = new UnitTestMessage("val2");
            var msg3 = new UnitTestMessage("val3");

            var handlerStartedEvent = new ManualResetEvent(initialState: false);
            var handlerCompleteEvent = new ManualResetEvent(initialState: false);
            var elementsProcessed = new List<UnitTestMessage>();

            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(elements => { elementsProcessed.AddRange(elements); handlerStartedEvent.Set(); handlerCompleteEvent.WaitOne(); }, maxBulkSize: 1, LogProvider))
            {
                memoryQueue.StartListening();

                // Add 3 elements
                var task1 = memoryQueue.AddAsync(msg1);
                var task2 = memoryQueue.AddAsync(msg2);
                var task3 = memoryQueue.AddAsync(msg3);

                // Now trigger the shutdown and wait a second to ensure that the thread is actually blocked in the
                handlerStartedEvent.WaitOne(); // Ensure thread is actually in the middle of calling the handler write now
                Assert.Single(elementsProcessed); // Sanity testing

                // Now stop the actual queue
                var shutdownTask = memoryQueue.ShutdownAsync();
                handlerCompleteEvent.Set(); // Unfreeze the thread

                // Ensure all elements in the queue are processed
                await shutdownTask;

                // Ensure awaiting all 3 tasks will complete
                await task1;
                await task2;
                await task3;

                // Ensure all processed
                Assert.Equal(3, elementsProcessed.Count);
                Assert.Contains(msg1, elementsProcessed);
                Assert.Contains(msg2, elementsProcessed);
                Assert.Contains(msg3, elementsProcessed);

                // Ensure thread is stopped
                Assert.Equal(ThreadState.Stopped, memoryQueue.InternalThreadState);
            }
        }

        [Fact]
        public async void AddAsync_EnsureExecuteInBackground_WithResult_ExpectTaskSuccess_Test()
        {
            var jobStartEvent = new ManualResetEvent(initialState: false);
            var jobContinueEvent = new ManualResetEvent(initialState: false);

            IEnumerable<UnitTestMessage> inputElementList = null;
            int callCount = 0;
            Action<IEnumerable<UnitTestMessage>> handler = (IEnumerable<UnitTestMessage> elementList) =>
            {
                callCount++;
                inputElementList = elementList;
                jobStartEvent.Set();
                jobContinueEvent.WaitOne();
            };

            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(handler, maxBulkSize: 10, LogProvider))
            {
                var element = new UnitTestMessage("some val");

                memoryQueue.StartListening();
                var task = memoryQueue.AddAsync(element);

                jobStartEvent.WaitOne();// Wait for the background task to execute the job
                Assert.NotEqual(TaskStatus.RanToCompletion, task.Status);
                Assert.NotEqual(TaskStatus.Faulted, task.Status);

                // Now allow the job running on the other thread to run to completion
                jobContinueEvent.Set();

                // An await the final task to complete
                await task;
                Assert.Equal(TaskStatus.RanToCompletion, task.Status);

                Assert.Equal(1, callCount);
                Assert.NotNull(inputElementList);
                Assert.Same(element, inputElementList.Single());
            }
        }

        [Fact]
        public async void AddAsync_EnsureExecuteInBackground_WithException_ExpectTaskException_Test()
        {
            var jobStartEvent = new ManualResetEvent(initialState: false);
            var jobContinueEvent = new ManualResetEvent(initialState: false);

            var targetException = new FieldAccessException("Some specific exception");

            IEnumerable<UnitTestMessage> inputElementList = null;
            int callCount = 0;
            Action<IEnumerable<UnitTestMessage>> handler = (IEnumerable<UnitTestMessage> elementList) =>
            {
                callCount++;
                inputElementList = elementList;
                jobStartEvent.Set();

                jobContinueEvent.WaitOne();

                throw targetException;
            };

            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(handler, maxBulkSize: 10, LogProvider))
            {
                var element = new UnitTestMessage("some val");

                memoryQueue.StartListening();
                var task = memoryQueue.AddAsync(element);

                jobStartEvent.WaitOne();// Wait for the background task to execute the job
                Assert.NotEqual(TaskStatus.RanToCompletion, task.Status);
                Assert.NotEqual(TaskStatus.Faulted, task.Status);

                // Now allow the job running on the other thread to run to completion
                jobContinueEvent.Set();

                // An await the final task to complete
                try
                {
                    await task;
                    throw new Exception("Expected a FieldAccessException exception here");
                }
                catch (FieldAccessException ex)
                {
                    Assert.Same(targetException, ex);
                }

                Assert.Equal(TaskStatus.Faulted, task.Status);

                Assert.Equal(1, callCount);
                Assert.NotNull(inputElementList);
                Assert.Same(element, inputElementList.Single());

                Log.Assert(Common.LogLevel.Fatal, message: @"Unexpected exception processing 1 of type Inmobile.Tools.MemoryQueues.Test.MemoryQueueTest+UnitTestMessage. Entry data: [{""Foo"":""some val""}]");
            }
        }

        [Fact]
        public async void BulkSize_EnsuringBulkSizeIsNotExceeded_Test()
        {
            var msg1 = new UnitTestMessage("val1");
            var msg2 = new UnitTestMessage("val2");
            var msg3 = new UnitTestMessage("val3");
            var msg4 = new UnitTestMessage("val4");
            var msg5 = new UnitTestMessage("val5");
            var msg6 = new UnitTestMessage("val6");
            var msg7 = new UnitTestMessage("val7");

            var handlerCallBulks = new List<List<UnitTestMessage>>();
            
            var handlerStartedEvent = new ManualResetEvent(initialState: false);

            Action<IEnumerable<UnitTestMessage>> handler = (IEnumerable<UnitTestMessage> elementList) =>
            {
                handlerStartedEvent.Set();
                handlerCallBulks.Add(elementList.ToList());
            };

            using (var memoryQueue = new MemoryQueue<UnitTestMessage>(handler, maxBulkSize: 4, LogProvider)) // Max bulk size: 4
            {
                var element = new UnitTestMessage("some val");

                memoryQueue.StartListening();

                // Adding the first message will trigger adding the rest inside the handler call
                var task = memoryQueue.AddAsync(msg1);

                // Wait until the handler is called the first time - this ensures that the handler is only called with a single element
                handlerStartedEvent.WaitOne();

                // At the first call to this handler, it is expected to only have a single element in the list
                // While this handler is being called, the rest of the elements are added, enabling full control of how many elements are in the queue when the worker thread is fetching new entries

#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
                memoryQueue.AddAsync(msg2);
                memoryQueue.AddAsync(msg3);
                memoryQueue.AddAsync(msg4);
                memoryQueue.AddAsync(msg5);
                memoryQueue.AddAsync(msg6);
                memoryQueue.AddAsync(msg7);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed

                // Now stop the memoryQueue and expect it to run until no more messages
                await memoryQueue.ShutdownAsync();

                // Start out ensuring all elements have actually been delivered to the handler in any call
                var allProcessedElements = handlerCallBulks.SelectMany(list => list).ToList();
                Assert.Contains(msg1, allProcessedElements);
                Assert.Contains(msg2, allProcessedElements);
                Assert.Contains(msg3, allProcessedElements);
                Assert.Contains(msg4, allProcessedElements);
                Assert.Contains(msg5, allProcessedElements);
                Assert.Contains(msg6, allProcessedElements);
                Assert.Contains(msg7, allProcessedElements);
                Assert.Equal(7, allProcessedElements.Count);

                // Now assert the expect bulk sizes with the maximum bulk size of 4. Expected calls:
                // msg1
                // msg2, msg3, msg4, msg5
                // msg6, msg7
                Assert.Equal(3, handlerCallBulks.Count);
                Assert.Single(handlerCallBulks[0]);
                Assert.Equal(4, handlerCallBulks[1].Count);
                Assert.Equal(2, handlerCallBulks[2].Count);

                // An await the final task to complete
                await task;
                Assert.Equal(TaskStatus.RanToCompletion, task.Status);
            }
        }

        public class UnitTestMessage
        {
            public string Foo { get; set; }

            public UnitTestMessage(string foo)
            {
                Foo = foo;
            }
        }
    }

Things I would like reviewed

Is the construction of the async interface correct?

I am using a TaskCompletionSource with an object as a generic type parameter (because I couldn't find a non-generic version). Is this use correct?

Risk of deadlocks

I have already fixed one deadlock: calling .Dispose() directly on the collection while another thread is listening will from time to time yield a deadlock with the waiting thread keeping blocking on the Take() call.

The solution is to use a cancellation token that is Cancel()ed in the MemoryQueue.Dispose() method. And then use this cancellation token inside the thread's logic and dispose the queue when the thread has run dry.

Are there any remaining risks of deadlocks?

Multithreaded unit tests

I have to this day never created a multithreaded unit test that I was proud of. How would other people go about the testing of all this? Feel free to pinpoint single test methods to show how I could improve on this.

Did I miss something?

Should I set up my post differently? Did I reinvent the wheel? Should I go about with a totally different approach?

\$\endgroup\$

0

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.