Threading Model

Threading Model

Overview

FLEXON is designed to be thread-safe and efficient in multi-threaded environments. This guide covers threading patterns, synchronization, and best practices.

Thread Safety

1. Serializer Thread Safety

public class FlexonSerializer
{
    // Thread-safe instance
    private static readonly FlexonSerializer _instance = new();
    
    // Thread-local storage for buffers
    private readonly ThreadLocal<byte[]> _buffer = 
        new(() => new byte[8192]);
        
    // Concurrent type cache
    private readonly ConcurrentDictionary<Type, TypeInfo> _typeCache = 
        new();
}

2. Buffer Management

public class ThreadSafeBuffer
{
    private readonly ArrayPool<byte> _pool;
    private readonly SemaphoreSlim _semaphore;

    public async Task<byte[]> RentAsync()
    {
        await _semaphore.WaitAsync();
        try
        {
            return _pool.Rent(8192);
        }
        finally
        {
            _semaphore.Release();
        }
    }
}

3. Type Registry

public class TypeRegistry
{
    private readonly ReaderWriterLockSlim _lock = new();
    private readonly Dictionary<int, Type> _types = new();

    public void Register<T>(int typeCode)
    {
        _lock.EnterWriteLock();
        try
        {
            _types[typeCode] = typeof(T);
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }
}

Concurrent Operations

1. Parallel Processing

public class ParallelProcessor
{
    private readonly FlexonSerializer _serializer;
    private readonly ParallelOptions _options;

    public async Task ProcessBatch<T>(IEnumerable<T> items)
    {
        await Parallel.ForEachAsync(items, _options, async (item, ct) =>
        {
            await ProcessItem(item);
        });
    }
}

2. Producer-Consumer Pattern

public class SerializationQueue
{
    private readonly BlockingCollection<WorkItem> _queue;
    private readonly CancellationTokenSource _cts;
    private readonly Task[] _workers;

    public async Task Enqueue(object item)
    {
        var workItem = new WorkItem(item);
        _queue.Add(workItem);
        await workItem.Completion;
    }

    private async Task ProcessQueue()
    {
        foreach (var item in _queue.GetConsumingEnumerable())
        {
            try
            {
                var result = await ProcessItem(item);
                item.SetResult(result);
            }
            catch (Exception ex)
            {
                item.SetException(ex);
            }
        }
    }
}

3. Async IO

public class AsyncSerializer
{
    private readonly SemaphoreSlim _throttle;

    public async Task SerializeToFile<T>(T item, string path)
    {
        await _throttle.WaitAsync();
        try
        {
            using var file = File.OpenWrite(path);
            await SerializeAsync(item, file);
        }
        finally
        {
            _throttle.Release();
        }
    }
}

Synchronization

1. Lock Management

public class LockManager
{
    private readonly Dictionary<string, SemaphoreSlim> _locks = new();
    private readonly object _syncRoot = new();

    public async Task<IDisposable> AcquireLock(string key)
    {
        SemaphoreSlim semaphore;
        lock (_syncRoot)
        {
            if (!_locks.TryGetValue(key, out semaphore))
            {
                semaphore = new SemaphoreSlim(1, 1);
                _locks[key] = semaphore;
            }
        }

        await semaphore.WaitAsync();
        return new LockReleaser(semaphore);
    }
}

2. Resource Management

public class ResourceManager
{
    private readonly ConcurrentDictionary<string, Resource> _resources;
    private readonly SemaphoreSlim _globalLock;

    public async Task<Resource> GetResource(string key)
    {
        if (_resources.TryGetValue(key, out var resource))
            return resource;

        await _globalLock.WaitAsync();
        try
        {
            return _resources.GetOrAdd(key, CreateResource);
        }
        finally
        {
            _globalLock.Release();
        }
    }
}

3. Event Handling

public class EventManager
{
    private readonly ConcurrentDictionary<string, List<Action<Event>>> _handlers;
    private readonly AsyncReaderWriterLock _lock = new();

    public async Task Subscribe(string eventType, Action<Event> handler)
    {
        using (await _lock.WriterLockAsync())
        {
            var handlers = _handlers.GetOrAdd(eventType, _ => new List<Action<Event>>());
            handlers.Add(handler);
        }
    }

    public async Task Publish(Event evt)
    {
        using (await _lock.ReaderLockAsync())
        {
            if (_handlers.TryGetValue(evt.Type, out var handlers))
            {
                foreach (var handler in handlers)
                {
                    handler(evt);
                }
            }
        }
    }
}

Performance Optimization

1. Thread Pool Configuration

public class ThreadPoolConfig
{
    public static void Configure()
    {
        ThreadPool.GetMinThreads(out int workerThreads, out int completionPortThreads);
        ThreadPool.SetMinThreads(workerThreads * 2, completionPortThreads * 2);
    }
}

2. Batch Processing

public class BatchProcessor
{
    private readonly int _batchSize;
    private readonly ConcurrentQueue<WorkItem> _queue;
    private readonly Task[] _processors;

    public async Task ProcessQueue()
    {
        var batch = new List<WorkItem>(_batchSize);
        while (await WaitForItems())
        {
            while (batch.Count < _batchSize && _queue.TryDequeue(out var item))
            {
                batch.Add(item);
            }

            if (batch.Count > 0)
            {
                await ProcessBatch(batch);
                batch.Clear();
            }
        }
    }
}

3. Memory Management

public class MemoryManager
{
    private readonly ArrayPool<byte> _arrayPool;
    private readonly ObjectPool<FlexonWriter> _writerPool;

    public async Task ProcessLargeData(Stream data)
    {
        var buffer = _arrayPool.Rent(81920);
        try
        {
            var writer = _writerPool.Get();
            try
            {
                await ProcessWithBuffer(data, buffer, writer);
            }
            finally
            {
                _writerPool.Return(writer);
            }
        }
        finally
        {
            _arrayPool.Return(buffer);
        }
    }
}

Best Practices

1. Threading Guidelines

• Use async/await for IO operations
• Avoid blocking operations
• Implement proper cancellation
• Handle exceptions appropriately

2. Resource Management

• Pool frequently used objects
• Implement proper disposal
• Use appropriate synchronization
• Monitor resource usage

3. Performance

• Configure thread pool appropriately
• Use batch processing when possible
• Monitor thread usage
• Profile performance bottlenecks

4. Error Handling

• Handle exceptions in all threads
• Implement proper logging
• Monitor thread health
• Implement circuit breakers

Common Pitfalls

1. Deadlocks

   • Avoid nested locks
   • Use timeouts
   • Follow lock ordering

2. Resource Leaks

   • Properly dispose resources
   • Use using statements
   • Implement IDisposable

3. Thread Starvation

   • Monitor thread usage
   • Implement timeouts
   • Use proper throttling

4. Race Conditions

   • Use proper synchronization
   • Implement atomic operations
   • Validate thread safety