Active Object: The Retro-Concurrency Pattern That’s Still Relevant in C# .NET

Introduction

Sometimes, it’s handy to separate when a method is called from when it actually runs. That’s where the Active Object design pattern comes in. It lets us build more flexible and concurrent systems. One key feature is that each method call is wrapped up in an object and placed in a queue. The Active Object then processes this queue in a separate thread.

This may seem a bit cryptic, so let’s look at an example. In this example I will be using the SharpAndRusty nuget package, so if you want to type along make sure you add that nuget package. At the moment of writing it is available in version 1.24

Implementation in C#

Before we can start we need to import these packages:

using System.Threading.Channels;
using Esox.SharpAndRusty.Extensions;
using Esox.SharpAndRusty.Types;

Line by line:

1. We start by importing the System.Threading.Channels packages, which imports the modern Channel<T> API which replaced BlockingCollection<T> for better async/await support and non-blocking producer-consumer patterns.
2. Next we import the Essox.SharpAndRusty.Extensions This imports the custom extension methods from the SharpAndRusty library, providing utility methods like ErrorExtensions.Try().
3. Finally we import Esox.SharpAndRusty.Types. This imports the custom types from the SharpAndRusty library including Result<T,E>, Error and ErrorKind which will be used for functional error handling.

For this example we will be building a very simple logging system. Logmessages in this system, like in many logging systems, have levels, as expressed in this enum:

    public enum LogLevel
    {
        Info,
        Warn,
        Error
    }

Finally we define the LogMessage itself, which contains both a field level of type LogLevel and a message of type String.

Besides that, it contains a constructor and some utility methods:

    public class LogMessage
    {
        public LogLevel Level { get; }
        public string Message { get; }

        public LogMessage(LogLevel level, string message)
        {
            Level = level;
            Message = message;
        }

        public override string ToString() => $"[{Level}] {Message}";

        public override bool Equals(object? obj)
        {
            return obj is LogMessage other &&
                   Level == other.Level &&
                   Message == other.Message;
        }

        public override int GetHashCode()
        {
            return HashCode.Combine(Level, Message);
        }
    }

Because every logmessage could be handled differently, we need a way to wrap that functionality:

    internal abstract class QueueMessage
    {
        public class Run : QueueMessage
        {
            public Action<LogMessage> Handler { get; }
            public LogMessage Message { get; }

            public Run(Action<LogMessage> handler, LogMessage message)
            {
                Handler = handler;
                Message = message;
            }
        }

        public class Terminate : QueueMessage;      
    }

Line by line:

1. We start by creating an internal abstract which will serve as the parent type fir all possible message types in the queue.
2. Next we define the first subclass, the Run class. This represents a message that tells the worker thread to execute a logging operation. The class contain an Action<LogMessage> Handler to specify what action to perform. It also contains a LogMessage with the actual log data to process.The constructor initializes both properties.
3. The Terminate class represents a message that signals the worker thread to stop processing and shutdown gracefully.

This pattern enables type-safe message passing where the worker thread can use pattern matching (switch expression) to handle different command types similar to Rust’s enuma variants or F# discriminated unions.

Now we come to the heart of our pattern, the ActiveLogger itself. We will start with this:

    public class ActiveLogger : IDisposable
    {
        private readonly Channel<QueueMessage> _channel;
        private readonly Task _workerTask;
        private bool _disposed;
    }

Line by line:

1. We start by defining a channel. This the thread-safe communication channel which acts as the message queue between producers, i.e. callers of the Run() method and the single consumer, which is the worker task.
2. The worker task is running in the background, so we need a separate Task for that, the _workerTask.
3. Finally we need a flag to track whether the logger has been disposed or terminated, preventing operations on a closed logger and avoiding double-disposal.

These fields represent the core components of the Active Object pattern, the scheduler queue, the execution thread and lifecycle management.

Next add the constructor:

        public ActiveLogger()
        {
            _channel = Channel.CreateUnbounded<QueueMessage>(new UnboundedChannelOptions
            {
                SingleReader = true,
                SingleWriter = false
            });
            _workerTask = Task.Run(WorkerLoopAsync);
        }

Line by line:

1. We start by creating an unbounded channel, which is a channel with unlimited queue capacity to hold the incoming messages.
2. Set SingleReader to true, an optimization hint, indicating only one task, the worker, will read from the channel, allowing internal optimizations.
3. Setting SingleWriter to false indicates multiple threads or tasks may write to the channel simultaneously, so proper synchronization needs to be mainitained.
4. Now we start the worker tasks immediately on the thread pool which begins waiting and processing messages from the channel.

Now it is time to implement the message handling loop:

        private async Task WorkerLoopAsync()
        {
            try
            {
                await foreach (var message in _channel.Reader.ReadAllAsync())
                {
                    switch (message)
                    {
                        case QueueMessage.Run run:
                            try
                            {
                                run.Handler(run.Message);
                            }
                            catch (Exception ex)
                            {
                                Console.Error.WriteLine($"Error executing log handler: {ex.Message}");
                            }

                            break;
                        case QueueMessage.Terminate:
                            Console.WriteLine("Terminating worker");
                            return;
                    }
                }
            }
            catch (ChannelClosedException)
            {
                Console.WriteLine("Channel closed");
            }
        }

Line by line:

1. The foreach loop iterates asynchronously over messages from the channel, awaiting when no messages are available. It is non-blocking.
2. Next we do some pattern matching on the message type.
3. If type is QueueMessage.Run we try and run the handler with the provided message.
4. If we encounter QueueMessage.Terminate we print a message to the console, and return from this method, effectively ending the loop.

We also need a method to add messages to the queue:

        public async Task<Result<Unit,Error>> Run(Action<LogMessage> handler, LogMessage message)
        {
            if (_disposed)
            {
                return Result<Unit, Error>.Err(
                    Error.New("Logger has been disposed").WithKind(ErrorKind.Other)
                );
            }

            return (await ErrorExtensions.TryAsync(async () =>
            {
                var queueMessage = new QueueMessage.Run(handler, message);
                await _channel.Writer.WriteAsync(queueMessage);
                return Unit.Value;
            })).Context("Failed to queue message");
        }

Line by line:

1. The method is async, returning a result indicating success Unit or an Error.
2. We first make sure the logger has not been disposed, and it has we return an error result.
3. The TryAsync method wraps an entire operation, catches exceptions and converts them to Result<T,E>.
4. In the body of the TryAsync we create a new QueueMessage.Run.
5. Now by using the WriteAsync() method we asynchronously write the message the message to the channel, non-blocking, making it available for the worker task to process.
6. If everything went well, we return a Unit.Value indicating success.
7. Finally the call to Context() adds contextual information to any error that occurs.

We also need method to indicate termination:

        public async Task<Result<Unit, Error>> Terminate()
        {
            if (_disposed)
            {
                return Result<Unit, Error>.Err(
                    Error.New("Logger has already been terminated")
                        .WithKind(ErrorKind.Other)
                );
            }

            Console.WriteLine("Terminating...");

            return (await ErrorExtensions.TryAsync(async () =>
            {
                await _channel.Writer.WriteAsync(new QueueMessage.Terminate());
                _channel.Writer.Complete();
                await _workerTask;
                _disposed = true;
                return Unit.Value;
            })).Context("Failed to terminate logger");
        }

Line by line:

1. Like in the Run() method we check if the logger has been disposed, and return an error it is.
2. Next we print a termination message to the console.
3. Here we also use the TryAsync() method, which catches exceptions and turns them into Result<T,E> objects.
4. In the body we start by writing a QueueMessage.Terminate to the queue, which will end processing of messages.
5. The call to Complete() marks the channel as complete, preventing any new messages from being written.
6. Since there may still be messages in the queue we need to wait for the worker task to finish by calling .await
7. Since everything is disposed, we can set _disposed to true
8. Now that all is done, we can return Unit.Value to indicate success.
9. The call to Context() adds context to any errors during shutdown.

You may have noticed that ActiveLogger implements the IDisposable interface, so need to implement a Dispose() method:

        public void Dispose()
        {
            if (!_disposed)
            {
                Terminate().GetAwaiter().GetResult();
            }
        }

Line by line:

1. We start to check whether the logger needs disposing.
2. If we it does we call the Terminate() method and synchronously block and wait for this async method to complete by unwrapping the Task.

Using the IDisposable interface enables automatic resource cleanup when used in a using block ensuring the worker tasks is properly shut down even if the caller forgets to call Terminate() explicitly.

The full code of the ActiveLogger:

    public class ActiveLogger : IDisposable
    {
        private readonly Channel<QueueMessage> _channel;
        private readonly Task _workerTask;
        private bool _disposed;

        public ActiveLogger()
        {
            _channel = Channel.CreateUnbounded<QueueMessage>(new UnboundedChannelOptions
            {
                SingleReader = true,
                SingleWriter = false
            });
            _workerTask = Task.Run(WorkerLoopAsync);
        }

        private async Task WorkerLoopAsync()
        {
            try
            {
                await foreach (var message in _channel.Reader.ReadAllAsync())
                {
                    switch (message)
                    {
                        case QueueMessage.Run run:
                            try
                            {
                                run.Handler(run.Message);
                            }
                            catch (Exception ex)
                            {
                                Console.Error.WriteLine($"Error executing log handler: {ex.Message}");
                            }

                            break;
                        case QueueMessage.Terminate:
                            Console.WriteLine("Terminating worker");
                            return;
                    }
                }
            }
            catch (ChannelClosedException)
            {
                Console.WriteLine("Channel closed");
            }
        }

        public async Task<Result<Unit,Error>> Run(Action<LogMessage> handler, LogMessage message)
        {
            if (_disposed)
            {
                return Result<Unit, Error>.Err(
                    Error.New("Logger has been disposed").WithKind(ErrorKind.Other)
                );
            }

            return (await ErrorExtensions.TryAsync(async () =>
            {
                var queueMessage = new QueueMessage.Run(handler, message);
                await _channel.Writer.WriteAsync(queueMessage);
                return Unit.Value;
            })).Context("Failed to queue message");
        }

        public async Task<Result<Unit, Error>> Terminate()
        {
            if (_disposed)
            {
                return Result<Unit, Error>.Err(
                    Error.New("Logger has already been terminated")
                        .WithKind(ErrorKind.Other)
                );
            }

            Console.WriteLine("Terminating...");

            return (await ErrorExtensions.TryAsync(async () =>
            {
                await _channel.Writer.WriteAsync(new QueueMessage.Terminate());
                _channel.Writer.Complete();
                await _workerTask;
                _disposed = true;
                return Unit.Value;
            })).Context("Failed to terminate logger");
        }

        public void Dispose()
        {
            if (!_disposed)
            {
                Terminate().GetAwaiter().GetResult();
            }
        }
    }

Testing time

Now we can test this pattern in a real program:

class Program
    {
        static async Task Main(string[] args)
        {
            using var logger = new ActiveLogger();

            var warnMessage = new LogMessage(LogLevel.Warn, "Hello, world!");
            (await logger.Run(
                mes => Console.WriteLine($"{mes.Level}:{mes.Message}"),
                warnMessage
            )).Match<Unit>(
                success: _ => Unit.Value,
                failure: error => { Console.Error.WriteLine($"Failed to log message: {error.Message}"); return Unit.Value; }
            );

            var infoMessage = new LogMessage(LogLevel.Info, "Hello, world!");
            (await logger.Run(
                mes => Console.WriteLine($"{mes.Level}:{mes.Message}"),
                infoMessage
            )).Match<Unit>(
                success: _ => Unit.Value,
                failure: error => { Console.Error.WriteLine($"Failed to log message: {error.Message}"); return Unit.Value; }
            );

            var errorMessage = new LogMessage(LogLevel.Error, "Error!");
            (await logger.Run(
                mes => Console.WriteLine($"{mes.Level}:{mes.Message}"),
                errorMessage
            )).Match<Unit>(
                success: _ => Unit.Value,
                failure: error => { Console.Error.WriteLine($"Failed to log message: {error.Message}"); return Unit.Value; }
            );

            (await logger.Terminate()).Match<Unit>(
                success: _ => Unit.Value,
                failure: error => { Console.Error.WriteLine($"Failed to terminate logger: {error.Message}"); return Unit.Value; }
            );
        }
    }
}

Some notes:

• We start by creating a new ActiveLogger using a using keyword. This means that the Dispose() method will automatically be called once the variable goes out of scope.
• Next we repeat the same pattern 3 times but with different loglevels:
  • Create a new logmessage.
  • Call the Run() method asynchronously providing a lamba expression, which is the handler, and the newly constructed logmessage.
  • Handle errors if needed by calling the Match() method.
• The final call is to Terminate() to terminate the logger. This is technically not necessary since the Dispose() method will be called automtically at the end of the app.

Conclusion

The Active Object pattern, though originating from a time before modern async/await, has demonstrated its enduring relevance and effectiveness in tackling modern concurrency challenges in C# .NET. Our implementation of the ActiveLogger showcased the pattern’s core strength: achieving a robust separation between scheduling client work and the sequential execution of that work on a dedicated, single thread.

The Foundational Benefits of the Active Object

The core structure, composed of the Channel<QueueMessage> as the thread-safe scheduler and the _workerTask as the exclusive execution engine, provided two critical benefits:

1. Non-Blocking Responsiveness: Client threads calling logger.Run(...) are never blocked waiting for the I/O-bound logging operation to complete. They merely enqueue a command object and return immediately, ensuring the application remains highly responsive.
2. Inherently Thread-Safe Execution: The pattern eliminates the need for complex internal locks on the logger’s state. Since all operations (the handlers) are strictly executed one-by-one by a single worker thread via the WorkerLoopAsync, we achieve thread safety by design, making the code easier to reason about and less prone to concurrency bugs.
3. Graceful Resource Management: Leveraging modern C# constructs like System.Threading.Channels for the non-blocking queue and the IDisposable interface ensures the lifecycle of the Active Object is managed cleanly. The Terminate message provides a mechanism for graceful shutdown, allowing the worker to finish processing any remaining messages in the queue before the task completes.

The Force Multiplier: Functional Error Handling with SharpAndRusty

The reliability and structure of our ActiveLogger were significantly enhanced by integrating the SharpAndRusty library, adopting its Rust-inspired approach to explicit, type-safe error handling. This library moved us beyond relying on traditional exceptions for expected failures, resulting in a system that is more robust and self-documenting.

The use of Task<Result<Unit, Error>> as the return type for Run and Terminate provided these key advantages:

• Explicit Contract for Failure: The type signature itself acts as documentation, explicitly telling the caller that the operation might fail (return an Err variant) rather than silently throwing an exception.
• Safe Handling of Void Operations: For operations like logging, where success yields no meaningful return value, the Unit type serves as the perfect placeholder for the success value of the Result. This prevents having to use less descriptive types like Result<bool, Error> or Result<object, Error>.
• Structured Exception-to-Error Conversion: The ErrorExtensions.TryAsync utility seamlessly handled the conversion of low-level exceptions (e.g., trying to write to a disposed channel) into the rich Error type, ensuring that our functional error pipeline receives structured error objects, not raw exceptions.
• Rich, Debuggable Error Context: By chaining methods like .Context("...") onto the result, we were able to embed crucial contextual information directly into the error object as the failure propagated up the call stack, dramatically simplifying debugging and error tracing in production environments. The SharpAndRusty Error type also supports metadata attachment, allowing us to include arbitrary details like timestamps or user IDs to the failure record.
• Forced Error Handling: The pattern matching approach via .Match(...) on the returned Result forces the developer to explicitly handle both the success path and the failure path, preventing missed error scenarios that are common with try-catch blocks.

In conclusion, the Active Object is far from obsolete. It is a powerful structured concurrency model, perfectly complemented by modern C# features and functional programming principles. By combining the thread-safe sequential execution of the Active Object with the type-safe, explicit error handling of SharpAndRusty, developers can build concurrent systems that prioritize both performance and maintainability, yielding demonstrably higher quality production code.