Многопоточность в C#: Threadpool

ThreadPool.QueueUserWorkItem(state => {
    Console.WriteLine($"Выполняется работа в потоке {Thread.CurrentThread.ManagedThreadId}");
    // Здесь могла бы быть ваша задача
});

// Создание обычного потока
Thread thread = new Thread(() => {
    Console.WriteLine("Выполняется в новом потоке");
});
thread.Start();
thread.Join(); // Ожидание завершения потока
 
// С использованием ThreadPool
ThreadPool.QueueUserWorkItem(_ => {
    Console.WriteLine("Выполняется в потоке из пула");
});
// Нет прямого способа дождаться завершения

// Можно узнать текущие настройки пула
int workerThreads, completionPortThreads;
ThreadPool.GetMaxThreads(out workerThreads, out completionPortThreads);
Console.WriteLine($"Макс. количество рабочих потоков: {workerThreads}");
Console.WriteLine($"Макс. количество потоков ввода-вывода: {completionPortThreads}");

ThreadPool.QueueUserWorkItem(_ => {
    Console.WriteLine("Начало длительной операции");
    // Блокировка потока на 10 секунд
    Thread.Sleep(10000);
    Console.WriteLine("Завершение длительной операции");
});

Thread heavyWorkThread = new Thread(() => {
    // Длительная вычислительная задача
    PerformHeavyCalculations();
});
heavyWorkThread.Start();

// Получение текущих параметров пула
int minWorker, minIOCP;
ThreadPool.GetMinThreads(out minWorker, out minIOCP);
Console.WriteLine($"Мин. рабочих потоков: {minWorker}, мин. потоков I/O: {minIOCP}");
 
int maxWorker, maxIOCP; 
ThreadPool.GetMaxThreads(out maxWorker, out maxIOCP);
Console.WriteLine($"Макс. рабочих потоков: {maxWorker}, макс. потоков I/O: {maxIOCP}");
 
// Получение текущего состояния
int availableWorker, availableIOCP;
ThreadPool.GetAvailableThreads(out availableWorker, out availableIOCP);
Console.WriteLine($"Доступно рабочих потоков: {availableWorker}, доступно I/O: {availableIOCP}");

using System.Diagnostics;
 
// Создание счетчика для отслеживания количества потоков в пуле
PerformanceCounter threadCount = new PerformanceCounter(
    "Пул потоков .NET CLR", 
    "# Рабочих потоков",
    Process.GetCurrentProcess().ProcessName);
 
// Периодическое считывание значения
while (true) {
    Console.WriteLine($"Текущее количество потоков: {threadCount.NextValue()}");
    Thread.Sleep(1000);
}

// Асинхронная операция I/O, использующая ThreadPool через IOCP
async Task<string> ReadFileAsync(string path)
{
    using (FileStream stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, 
                                            4096, FileOptions.Asynchronous))
    {
        byte[] buffer = new byte[stream.Length];
        await stream.ReadAsync(buffer, 0, buffer.Length);
        return Encoding.UTF8.GetString(buffer);
    }
    // Здесь кодом управляет I/O поток из ThreadPool
}

ThreadPool.QueueUserWorkItem(callBack, state);

RegisteredWaitHandle handle = ThreadPool.RegisterWaitForSingleObject(
    waitObject,      // Объект для ожидания
    callBack,        // Функция обратного вызова
    state,           // Состояние, передаваемое в callback
    timeout,         // Время ожидания
    executeOnlyOnce  // Выполнить один раз или многократно
);

using System;
using System.Threading;
 
class Program
{
    static void Main()
    {
        // Простая демонстрация QueueUserWorkItem
        ThreadPool.QueueUserWorkItem(ProcessData, "Пример данных");
        
        Console.WriteLine("Работа поставлена в очередь");
        Console.ReadLine(); // Предотвращаем завершение программы
    }
    
    static void ProcessData(object data)
    {
        string inputData = data as string;
        Console.WriteLine($"Обработка данных '{inputData}' в потоке {Thread.CurrentThread.ManagedThreadId}");
        
        // Имитация долгой работы
        Thread.Sleep(1000);
        
        Console.WriteLine($"Обработка завершена в потоке {Thread.CurrentThread.ManagedThreadId}");
    }
}

class ImageProcessor
{
    public void ProcessImages(string[] imagePaths)
    {
        foreach (string path in imagePaths)
        {
            // Создаем параметры для передачи в поток
            var parameters = new ProcessingParameters
            {
                ImagePath = path,
                Quality = 80,
                StartTime = DateTime.Now
            };
            
            ThreadPool.QueueUserWorkItem(ProcessSingleImage, parameters);
        }
    }
    
    private void ProcessSingleImage(object state)
    {
        var parameters = (ProcessingParameters)state;
        Console.WriteLine($"Начата обработка {parameters.ImagePath}");
        
        // Здесь был бы код обработки изображения
        Thread.Sleep(2000); // Имитация работы
        
        TimeSpan processingTime = DateTime.Now - parameters.StartTime;
        Console.WriteLine($"Изображение {parameters.ImagePath} обработано за {processingTime.TotalMilliseconds} мс");
    }
    
    class ProcessingParameters
    {
        public string ImagePath { get; set; }
        public int Quality { get; set; }
        public DateTime StartTime { get; set; }
    }
}

public class SimpleServer
{
    private TcpListener _listener;
    private bool _isRunning;
    
    public void Start(int port)
    {
        _listener = new TcpListener(IPAddress.Any, port);
        _listener.Start();
        _isRunning = true;
        
        Console.WriteLine($"Сервер запущен на порту {port}");
        
        // Начинаем принимать клиентов в отдельном потоке
        ThreadPool.QueueUserWorkItem(_ => AcceptClientsLoop());
    }
    
    private void AcceptClientsLoop()
    {
        while (_isRunning)
        {
            try
            {
                // Принимаем нового клиента
                TcpClient client = _listener.AcceptTcpClient();
                
                // Обрабатываем клиента в потоке из пула
                ThreadPool.QueueUserWorkItem(_ => HandleClient(client));
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Ошибка при приеме клиента: {ex.Message}");
            }
        }
    }
    
    private void HandleClient(object clientObj)
    {
        TcpClient client = (TcpClient)clientObj;
        
        try
        {
            using (NetworkStream stream = client.GetStream())
            using (StreamReader reader = new StreamReader(stream))
            using (StreamWriter writer = new StreamWriter(stream) { AutoFlush = true })
            {
                // Читаем запрос
                string request = reader.ReadLine();
                Console.WriteLine($"Получен запрос: {request}");
                
                // Обрабатываем запрос (в реальном приложении здесь была бы логика)
                string response = $"Ответ на '{request}'";
                
                // Отправляем ответ
                writer.WriteLine(response);
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Ошибка при обработке клиента: {ex.Message}");
        }
        finally
        {
            client.Close();
        }
    }
    
    public void Stop()
    {
        _isRunning = false;
        _listener.Stop();
        Console.WriteLine("Сервер остановлен");
    }
}

public async Task<string> FetchDataAsync(string url)
{
    // Внутри Task.Run используется ThreadPool
    return await Task.Run(() =>
    {
        using (WebClient client = new WebClient())
        {
            return client.DownloadString(url);
        }
    });
}

public class ThreadPoolMonitor
{
    private Timer _monitoringTimer;
    private int _lastPendingWorkItems = 0;
    
    public void Start(TimeSpan interval)
    {
        _monitoringTimer = new Timer(MonitorThreadPool, null, TimeSpan.Zero, interval);
    }
    
    private void MonitorThreadPool(object state)
    {
        // Получаем информацию о текущем состоянии пула потоков
        ThreadPool.GetMaxThreads(out int maxWorker, out int maxIO);
        ThreadPool.GetAvailableThreads(out int availWorker, out int availIO);
        
        // Вычисляем использованные потоки
        int usedWorker = maxWorker - availWorker;
        int usedIO = maxIO - availIO;
        
        // Оцениваем количество ожидающих задач
        // Данное значение не предоставляется напрямую из ThreadPool API
        int pendingWorkItems = Process.GetCurrentProcess().Threads.Count - usedWorker - usedIO;
        
        // Вычисляем скорость прироста очереди
        int queueGrowthRate = pendingWorkItems - _lastPendingWorkItems;
        _lastPendingWorkItems = pendingWorkItems;
        
        // Логгируем информацию
        Console.WriteLine($"[{DateTime.Now:HH:mm:ss}] ThreadPool статистика:");
        Console.WriteLine($"Рабочие потоки: {usedWorker}/{maxWorker}");
        Console.WriteLine($"I/O потоки: {usedIO}/{maxIO}");
        Console.WriteLine($"Ожидающие задачи: {pendingWorkItems} (изменение: {queueGrowthRate})");
        
        // Проверяем потенциальные проблемы
        if (usedWorker > maxWorker * 0.8)
        {
            Console.WriteLine("ПРЕДУПРЕЖДЕНИЕ: Высокая нагрузка на пул рабочих потоков!");
        }
        
        if (queueGrowthRate > 10)
        {
            Console.WriteLine("ПРЕДУПРЕЖДЕНИЕ: Очередь задач быстро растет!");
        }
    }
    
    public void Stop()
    {
        _monitoringTimer?.Dispose();
        _monitoringTimer = null;
    }
}

public class ThreadPoolMetrics
{
    private readonly Counter _taskQueuedCounter;
    private readonly Gauge _threadsInUseGauge;
    private readonly Gauge _queueLengthGauge;
    
    public ThreadPoolMetrics()
    {
        // Создаем метрики для Prometheus
        _taskQueuedCounter = Metrics.CreateCounter(
            "threadpool_tasks_queued_total", 
            "Общее число задач, поставленных в очередь ThreadPool");
        
        _threadsInUseGauge = Metrics.CreateGauge(
            "threadpool_threads_in_use", 
            "Текущее количество используемых потоков в ThreadPool");
        
        _queueLengthGauge = Metrics.CreateGauge(
            "threadpool_queue_length", 
            "Приблизительная длина очереди ThreadPool");
        
        // Запускаем фоновое обновление метрик
        StartMetricsCollection();
    }
    
    private void StartMetricsCollection()
    {
        Task.Run(async () =>
        {
            while (true)
            {
                UpdateMetrics();
                await Task.Delay(1000);
            }
        });
    }
    
    private void UpdateMetrics()
    {
        ThreadPool.GetMaxThreads(out int maxWorker, out int maxIO);
        ThreadPool.GetAvailableThreads(out int availWorker, out int availIO);
        
        int usedWorker = maxWorker - availWorker;
        int usedIO = maxIO - availIO;
        
        _threadsInUseGauge.Set(usedWorker + usedIO);
        
        // Оценка длины очереди на основе косвенных данных
        // В реальном приложении используйте более точные методы
        // ...
    }
    
    public void TrackQueuedTask()
    {
        _taskQueuedCounter.Inc();
    }
}

public class ThreadPoolOptimizer
{
    public void OptimizeForHighConcurrency()
    {
        // Получаем текущие настройки
        ThreadPool.GetMinThreads(out int minWorker, out int minIO);
        
        // Увеличиваем минимальное количество потоков
        // чтобы избежать задержки при создании новых потоков
        int newMinWorker = Environment.ProcessorCount * 4;
        int newMinIO = Environment.ProcessorCount * 4;
        
        ThreadPool.SetMinThreads(newMinWorker, newMinIO);
        
        Console.WriteLine($"ThreadPool настроен для высокой конкурентности:");
        Console.WriteLine($"Мин. рабочих потоков: {minWorker} -> {newMinWorker}");
        Console.WriteLine($"Мин. потоков I/O: {minIO} -> {newMinIO}");
    }
    
    public void OptimizeForMemoryConstraints()
    {
        // Получаем текущие настройки
        ThreadPool.GetMaxThreads(out int maxWorker, out int maxIO);
        
        // Ограничиваем максимальное количество потоков
        int newMaxWorker = Math.Min(maxWorker, Environment.ProcessorCount * 8);
        int newMaxIO = Math.Min(maxIO, Environment.ProcessorCount * 8);
        
        ThreadPool.SetMaxThreads(newMaxWorker, newMaxIO);
        
        Console.WriteLine($"ThreadPool настроен для экономии памяти:");
        Console.WriteLine($"Макс. рабочих потоков: {maxWorker} -> {newMaxWorker}");
        Console.WriteLine($"Макс. потоков I/O: {maxIO} -> {newMaxIO}");
    }
}

public class WorkDistributor
{
    private readonly List<string> _workerNodes;
    private int _nextNodeIndex = 0;
    
    public WorkDistributor(List<string> workerNodes)
    {
        _workerNodes = workerNodes;
    }
    
    public async Task<string> ProcessWorkItem(string data)
    {
        // Выбираем узел по алгоритму "round-robin"
        string selectedNode;
        lock(this) // Защищаем от состояния гонки при выборе узла
        {
            selectedNode = _workerNodes[_nextNodeIndex];
            _nextNodeIndex = (_nextNodeIndex + 1) % _workerNodes.Count;
        }
        
        // В реальном сценарии здесь был бы код для отправки задачи на выбранный узел
        return await SimulateNodeProcessing(selectedNode, data);
    }
    
    private async Task<string> SimulateNodeProcessing(string node, string data)
    {
        Console.WriteLine($"Задача '{data}' отправлена на узел {node}");
        
        // Имитация обработки на удаленном узле
        await Task.Delay(1000);
        
        return $"Результат обработки '{data}' на узле {node}";
    }
}

public class ScalableWorkProcessor
{
    private readonly IQueueClient _inputQueue;
    private readonly IQueueClient _outputQueue;
    private readonly IScalingManager _scalingManager;
    private int _activeWorkers = 0;
    
    // ... инициализация в конструкторе ...
    
    public async Task StartProcessing()
    {
        while (true)
        {
            // Проверяем длину очереди
            int queueLength = await _inputQueue.GetApproximateMessageCount();
            
            // Оцениваем оптимальное количество воркеров на основе длины очереди
            int optimalWorkers = Math.Max(1, queueLength / 100);
            
            // Динамически масштабируем инфраструктуру
            if (_activeWorkers < optimalWorkers)
            {
                await _scalingManager.ScaleUp(optimalWorkers - _activeWorkers);
                _activeWorkers = optimalWorkers;
            }
            else if (_activeWorkers > optimalWorkers && _activeWorkers > 1)
            {
                await _scalingManager.ScaleDown(_activeWorkers - optimalWorkers);
                _activeWorkers = optimalWorkers;
            }
            
            await Task.Delay(30000); // Проверяем раз в 30 секунд
        }
    }
}

public class LegacyImageProcessor
{
    public void ProcessBatch(string[] files)
    {
        List<Thread> threads = new List<Thread>();
        
        foreach (string file in files)
        {
            Thread t = new Thread(() => ProcessImage(file));
            threads.Add(t);
            t.Start();
        }
        
        // Ожидаем завершения всех потоков
        foreach (Thread t in threads)
        {
            t.Join();
        }
        
        Console.WriteLine("Все изображения обработаны");
    }
    
    private void ProcessImage(string file)
    {
        Console.WriteLine($"Начата обработка {file}");
        // Имитация работы
        Thread.Sleep(2000);
        Console.WriteLine($"Файл {file} обработан");
    }
}

public class ModernImageProcessor
{
    private ManualResetEvent _doneEvent;
    private int _pendingFiles;
    
    public void ProcessBatch(string[] files)
    {
        _pendingFiles = files.Length;
        _doneEvent = new ManualResetEvent(false);
        
        foreach (string file in files)
        {
            ThreadPool.QueueUserWorkItem(ProcessImageCallback, file);
        }
        
        // Ожидаем завершения всех задач
        _doneEvent.WaitOne();
        Console.WriteLine("Все изображения обработаны");
    }
    
    private void ProcessImageCallback(object state)
    {
        string file = (string)state;
        
        try
        {
            Console.WriteLine($"Начата обработка {file}");
            // Имитация работы
            Thread.Sleep(2000);
            Console.WriteLine($"Файл {file} обработан");
        }
        finally
        {
            // Уменьшаем счетчик незавершенных файлов
            if (Interlocked.Decrement(ref _pendingFiles) == 0)
            {
                // Если все файлы обработаны, сигнализируем основному потоку
                _doneEvent.Set();
            }
        }
    }
}

public class EnhancedImageProcessor
{
    private SemaphoreSlim _throttler;
    private CountdownEvent _counter;
    
    public async Task ProcessBatchAsync(string[] files, int maxConcurrent)
    {
        _throttler = new SemaphoreSlim(maxConcurrent);
        _counter = new CountdownEvent(files.Length);
        
        List<Task> tasks = new List<Task>();
        
        foreach (string file in files)
        {
            tasks.Add(ProcessFileAsync(file));
        }
        
        // Ожидаем завершения всех задач
        await Task.WhenAll(tasks);
        Console.WriteLine("Все изображения обработаны");
    }
    
    private async Task ProcessFileAsync(string file)
    {
        try
        {
            await _throttler.WaitAsync();
            
            // Запускаем вычислительную работу через ThreadPool
            await Task.Run(() =>
            {
                Console.WriteLine($"Начата обработка {file}");
                // Имитация работы
                Thread.Sleep(2000);
                Console.WriteLine($"Файл {file} обработан");
            });
        }
        finally
        {
            _throttler.Release();
            _counter.Signal();
        }
    }
}

// Получаем текущие настройки
int minWorkerThreads, minIOThreads;
ThreadPool.GetMinThreads(out minWorkerThreads, out minIOThreads);
 
// Устанавливаем новые значения
int newMinWorkerThreads = Environment.ProcessorCount * 2;
int newMinIOThreads = Environment.ProcessorCount * 2;
ThreadPool.SetMinThreads(newMinWorkerThreads, newMinIOThreads);

int maxWorkerThreads, maxIOThreads;
ThreadPool.GetMaxThreads(out maxWorkerThreads, out maxIOThreads);
 
// Устанавливаем ограничения
ThreadPool.SetMaxThreads(
    Math.Min(maxWorkerThreads, Environment.ProcessorCount * 10), 
    Math.Min(maxIOThreads, Environment.ProcessorCount * 15)
);

try
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        throw new InvalidOperationException("Что-то пошло не так");
    });
}
catch (InvalidOperationException ex)
{
    // Этот код никогда не выполнится!
    Console.WriteLine($"Перехвачено: {ex.Message}");
}

ThreadPool.QueueUserWorkItem(_ =>
{
    try
    {
        // Опасный код
        DoSomethingRisky();
    }
    catch (Exception ex)
    {
        // Логирование ошибки
        Console.WriteLine($"Ошибка в потоке из пула: {ex.Message}");
    }
});

public static void QueueSafeWorkItem(WaitCallback callback, object state = null)
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        try
        {
            callback(state);
        }
        catch (Exception ex)
        {
            // Централизованная обработка ошибок
            LogException(ex);
        }
    }, state);
}
 
private static void LogException(Exception ex)
{
    // В реальном коде здесь был бы механизм логирования
    Console.WriteLine($"[{DateTime.Now}] Исключение в ThreadPool: {ex}");
}

// Этот код использует ThreadPool под капотом
async Task ProcessDataAsync(string data)
{
    // Task.Run помещает работу в ThreadPool
    var result = await Task.Run(() =>
    {
        return TransformData(data);
    });
    
    Console.WriteLine($"Результат: {result}");
}

public async Task<List<string>> ProcessFilesAsync(string[] filePaths)
{
    List<Task<string>> tasks = new List<Task<string>>();
    
    foreach (var path in filePaths)
    {
        // Запускаем задачи параллельно через пул потоков
        tasks.Add(Task.Run(() => ProcessSingleFile(path)));
    }
    
    // Ожидаем выполнения всех задач
    var results = await Task.WhenAll(tasks);
    
    return results.ToList();
}
 
private string ProcessSingleFile(string path)
{
    // Тяжелая вычислительная операция
    Thread.Sleep(1000); // Имитация работы
    return $"Результат обработки файла {path}";
}

// Правильно - используем встроенный асинхронный API
async Task<string> ReadFileAsync(string path)
{
    using var reader = new StreamReader(path);
    return await reader.ReadToEndAsync();
}
 
// Неправильно - заворачиваем синхронный API в Task.Run
async Task<string> ReadFileWrongWay(string path)
{
    return await Task.Run(() =>
    {
        using var reader = new StreamReader(path);
        return reader.ReadToEnd();
    });
}

// Получаем текущий контекст синхронизации
SynchronizationContext uiContext = SynchronizationContext.Current;
 
ThreadPool.QueueUserWorkItem(_ =>
{
    // Долгая операция в фоновом потоке
    var result = PerformHeavyCalculation();
    
    // Переключаемся на UI-поток для обновления интерфейса
    uiContext.Post(state =>
    {
        resultLabel.Text = result.ToString();
    }, null);
});

async void Button_Click(object sender, EventArgs e)
{
    resultLabel.Text = "Вычисление...";
    
    // Запускаем тяжелую работу в ThreadPool
    var result = await Task.Run(() => PerformHeavyCalculation());
    
    // Этот код автоматически выполнится в UI-потоке
    resultLabel.Text = result.ToString();
}

private object _lockObject = new object();
private Dictionary<string, int> _sharedData = new Dictionary<string, int>();
 
public void UpdateData(string key, int value)
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        lock (_lockObject)
        {
            _sharedData[key] = value;
        }
    });
}

private int _counter = 0;
 
public void IncrementCounter()
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        Interlocked.Increment(ref _counter);
    });
}

private ReaderWriterLockSlim _rwLock = new ReaderWriterLockSlim();
private Dictionary<string, object> _cache = new Dictionary<string, object>();
 
public object GetFromCache(string key)
{
    _rwLock.EnterReadLock();
    try
    {
        return _cache.ContainsKey(key) ? _cache[key] : null;
    }
    finally
    {
        _rwLock.ExitReadLock();
    }
}
 
public void UpdateCache(string key, object value)
{
    _rwLock.EnterWriteLock();
    try
    {
        _cache[key] = value;
    }
    finally
    {
        _rwLock.ExitWriteLock();
    }
}

object lockObject = new object();
 
// Плохой код - длительная блокировка в потоке из пула
ThreadPool.QueueUserWorkItem(_ =>
{
    lock (lockObject)
    {
        // Длительная операция внутри блокировки
        Thread.Sleep(10000);
        ProcessData();
    }
});

// Улучшенный вариант
ThreadPool.QueueUserWorkItem(_ =>
{
    // Быстрая операция под блокировкой - только для обновления данных
    lock (lockObject)
    {
        PrepareDataForProcessing();
    }
    
    // Длительная обработка без блокировки
    ProcessDataWithoutLock();
    
    // Ещё одна короткая блокировка для сохранения результата
    lock (lockObject)
    {
        SaveResults();
    }
});

public class PrioritizedThreadPool
{
    private readonly ConcurrentQueue<PriorityTask> _highPriorityTasks = new ConcurrentQueue<PriorityTask>();
    private readonly ConcurrentQueue<PriorityTask> _normalPriorityTasks = new ConcurrentQueue<PriorityTask>();
    private readonly ConcurrentQueue<PriorityTask> _lowPriorityTasks = new ConcurrentQueue<PriorityTask>();
    
    private int _activeThreads = 0;
    private readonly int _maxThreads;
    
    public PrioritizedThreadPool(int maxThreads)
    {
        _maxThreads = maxThreads;
    }
    
    public void QueueTask(Action task, TaskPriority priority)
    {
        var priorityTask = new PriorityTask { Action = task };
        
        switch (priority)
        {
            case TaskPriority.High:
                _highPriorityTasks.Enqueue(priorityTask);
                break;
            case TaskPriority.Normal:
                _normalPriorityTasks.Enqueue(priorityTask);
                break;
            case TaskPriority.Low:
                _lowPriorityTasks.Enqueue(priorityTask);
                break;
        }
        
        TryStartNewWorker();
    }
    
    private void TryStartNewWorker()
    {
        if (Interlocked.Increment(ref _activeThreads) <= _maxThreads)
        {
            ThreadPool.QueueUserWorkItem(_ => ProcessTasks());
        }
        else
        {
            Interlocked.Decrement(ref _activeThreads);
        }
    }
    
    private void ProcessTasks()
    {
        try
        {
            bool processed;
            
            do
            {
                processed = false;
                
                // Сначала пробуем задачи с высоким приоритетом
                if (_highPriorityTasks.TryDequeue(out var highTask))
                {
                    highTask.Action();
                    processed = true;
                }
                // Затем нормальный приоритет
                else if (_normalPriorityTasks.TryDequeue(out var normalTask))
                {
                    normalTask.Action();
                    processed = true;
                }
                // И наконец низкий приоритет
                else if (_lowPriorityTasks.TryDequeue(out var lowTask))
                {
                    lowTask.Action();
                    processed = true;
                }
            }
            while (processed);
        }
        finally
        {
            Interlocked.Decrement(ref _activeThreads);
        }
    }
    
    private class PriorityTask
    {
        public Action Action { get; set; }
    }
    
    public enum TaskPriority
    {
        Low,
        Normal,
        High
    }
}

public async Task<T> RunWithTimeout<T>(Func<T> function, TimeSpan timeout)
{
    using var cts = new CancellationTokenSource();
    var task = Task.Run(function, cts.Token);
    
    if (await Task.WhenAny(task, Task.Delay(timeout)) == task)
    {
        return await task;
    }
    else
    {
        cts.Cancel();
        throw new TimeoutException($"Операция превысила лимит времени ({timeout})");
    }
}

public class PersistentThreadWorker : IDisposable
{
    private readonly BlockingCollection<Action> _workItems = new BlockingCollection<Action>();
    private readonly Thread _workerThread;
    private bool _isDisposed = false;
    
    public PersistentThreadWorker(string threadName = null)
    {
        _workerThread = new Thread(WorkerLoop)
        {
            IsBackground = true,
            Name = threadName ?? $"PersistentWorker-{Guid.NewGuid()}"
        };
        _workerThread.Start();
    }
    
    public void QueueWork(Action workItem)
    {
        if (_isDisposed)
            throw new ObjectDisposedException(nameof(PersistentThreadWorker));
        
        _workItems.Add(workItem);
    }
    
    public Task QueueWorkAsync(Action workItem)
    {
        var tcs = new TaskCompletionSource<bool>();
        
        QueueWork(() =>
        {
            try
            {
                workItem();
                tcs.SetResult(true);
            }
            catch (Exception ex)
            {
                tcs.SetException(ex);
            }
        });
        
        return tcs.Task;
    }
    
    private void WorkerLoop()
    {
        foreach (var workItem in _workItems.GetConsumingEnumerable())
        {
            try
            {
                workItem();
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Ошибка в постоянном потоке: {ex.Message}");
            }
        }
    }
    
    public void Dispose()
    {
        if (!_isDisposed)
        {
            _isDisposed = true;
            _workItems.CompleteAdding();
            _workerThread.Join(1000); // Ждём завершения потока
        }
    }
}

public class ThreadPoolDiagnostics
{
    private readonly Timer _sampleTimer;
    private readonly List<ThreadPoolSnapshot> _snapshots = new List<ThreadPoolSnapshot>();
    private readonly int _maxSnapshots;
    
    public ThreadPoolDiagnostics(TimeSpan sampleInterval, int maxSnapshots = 100)
    {
        _maxSnapshots = maxSnapshots;
        _sampleTimer = new Timer(TakeSnapshot, null, TimeSpan.Zero, sampleInterval);
    }
    
    private void TakeSnapshot(object state)
    {
        ThreadPool.GetAvailableThreads(out int availWorkerThreads, out int availCompletionPortThreads);
        ThreadPool.GetMaxThreads(out int maxWorkerThreads, out int maxCompletionPortThreads);
        
        var snapshot = new ThreadPoolSnapshot
        {
            Timestamp = DateTime.UtcNow,
            ActiveWorkerThreads = maxWorkerThreads - availWorkerThreads,
            ActiveCompletionPortThreads = maxCompletionPortThreads - availCompletionPortThreads,
            ProcessorCount = Environment.ProcessorCount,
            ProcessThreadCount = Process.GetCurrentProcess().Threads.Count,
            ManagedThreadId = Thread.CurrentThread.ManagedThreadId
        };
        
        lock (_snapshots)
        {
            _snapshots.Add(snapshot);
            if (_snapshots.Count > _maxSnapshots)
                _snapshots.RemoveAt(0);
        }
    }
    
    public ThreadPoolSnapshot[] GetSnapshots()
    {
        lock (_snapshots)
        {
            return _snapshots.ToArray();
        }
    }
    
    public void DetectPotentialThreadStarvation()
    {
        var snapshots = GetSnapshots();
        if (snapshots.Length < 5) return;
        
        var recent = snapshots.TakeLast(5).ToArray();
        
        // Проверяем тренд использования потоков
        bool isGrowing = true;
        for (int i = 1; i < recent.Length; i++)
        {
            if (recent[i].ActiveWorkerThreads <= recent[i-1].ActiveWorkerThreads)
            {
                isGrowing = false;
                break;
            }
        }
        
        // Проверяем загрузку пула потоков
        var latest = recent.Last();
        ThreadPool.GetMaxThreads(out int maxWorkerThreads, out _);
        
        double utilization = (double)latest.ActiveWorkerThreads / maxWorkerThreads;
        
        if (isGrowing && utilization > 0.8)
        {
            // Выводим предупреждение
            Console.WriteLine("ПРЕДУПРЕЖДЕНИЕ: Обнаружен потенциальный дефицит потоков!");
            Console.WriteLine($"Использование пула: {utilization:P2}, активных потоков: {latest.ActiveWorkerThreads}");
            
            // Рекомендации
            if (latest.ActiveWorkerThreads > Environment.ProcessorCount * 10)
            {
                Console.WriteLine("Рекомендация: Проверьте наличие блокировок в коде или неоптимальных операций I/O");
            }
        }
    }
    
    public class ThreadPoolSnapshot
    {
        public DateTime Timestamp { get; set; }
        public int ActiveWorkerThreads { get; set; }
        public int ActiveCompletionPortThreads { get; set; }
        public int ProcessorCount { get; set; }
        public int ProcessThreadCount { get; set; }
        public int ManagedThreadId { get; set; }
    }
}

public class ThrottledTaskProcessor
{
    private readonly SemaphoreSlim _throttler;
    private readonly TimeSpan _interval;
    
    public ThrottledTaskProcessor(int maxConcurrentTasks, TimeSpan interval)
    {
        _throttler = new SemaphoreSlim(maxConcurrentTasks);
        _interval = interval;
    }
    
    public async Task ProcessAsync(Func<Task> action)
    {
        await _throttler.WaitAsync();
        
        try
        {
            await action();
        }
        finally
        {
            // Отложенное освобождение, чтобы ограничить частоту операций
            _ = Task.Run(async () =>
            {
                await Task.Delay(_interval);
                _throttler.Release();
            });
        }
    }
}

// Антипаттерн - блокировка потоков в пуле
for (int i = 0; i < 100; i++)
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        // Синхронная блокирующая операция в потоке из пула
        using (var client = new WebClient())
        {
            string data = client.DownloadString("http://api.example.com/data");
            ProcessData(data);
        }
    });
}

public static void MonitorThreadPoolStarvation()
{
    ThreadPool.GetMaxThreads(out int maxWorker, out int maxIO);
    ThreadPool.GetAvailableThreads(out int availWorker, out int availIO);
    
    int usedWorker = maxWorker - availWorker;
    int usedIO = maxIO - availIO;
    
    double workerUsagePercent = (double)usedWorker / maxWorker * 100;
    
    Console.WriteLine($"Использование потоков: {usedWorker}/{maxWorker} ({workerUsagePercent:F1}%)");
    
    if (workerUsagePercent > 90)
    {
        Console.WriteLine("ВНИМАНИЕ: Высокая нагрузка на пул потоков!");
        DumpRunningTasks(); // метод для анализа выполняемых задач
    }
}

// Правильный подход - асинхронные операции
async Task ProcessItemsAsync(IEnumerable<string> items)
{
    var tasks = new List<Task>();
    
    foreach (var item in items)
    {
        tasks.Add(ProcessSingleItemAsync(item));
    }
    
    await Task.WhenAll(tasks);
}
 
async Task ProcessSingleItemAsync(string item)
{
    using (var client = new HttpClient())
    {
        // Асинхронная операция, не блокирующая поток
        string data = await client.GetStringAsync($"http://api.example.com/data/{item}");
        ProcessData(data);
    }
}

public void StartProcessingWithCancellation(string data)
{
    var cts = new CancellationTokenSource();
    var token = cts.Token;
    
    // Сохраняем токен для возможности отмены позже
    _cancellationTokens[data] = cts;
    
    ThreadPool.QueueUserWorkItem(_ =>
    {
        try
        {
            // Периодически проверяем статус отмены
            for (int i = 0; i < 100; i++)
            {
                if (token.IsCancellationRequested)
                {
                    Console.WriteLine("Операция отменена");
                    return;
                }
                
                // Шаг обработки
                ProcessPartial(data, i);
                Thread.Sleep(100); // имитация работы
            }
            
            Console.WriteLine("Обработка завершена");
        }
        finally
        {
            // Очистка ресурсов
            lock (_cancellationTokens)
            {
                _cancellationTokens.Remove(data);
                cts.Dispose();
            }
        }
    });
}
 
public void CancelProcessing(string data)
{
    lock (_cancellationTokens)
    {
        if (_cancellationTokens.TryGetValue(data, out var cts))
        {
            cts.Cancel();
        }
    }
}

public void CalculateWithCallback(int x, int y, Action<int> onComplete)
{
    ThreadPool.QueueUserWorkItem(_ =>
    {
        int result = ComplexCalculation(x, y);
        onComplete(result);
    });
}
 
// Использование
CalculateWithCallback(10, 20, result =>
{
    Console.WriteLine($"Результат: {result}");
});

// Потенциально проблемный код
ThreadPool.QueueUserWorkItem(_ =>
{
    IntPtr handle = NativeMethods.Initialize();
    
    // Операции с нативным ресурсом
    ProcessWithNativeResource(handle);
    
    // Освобождение может произойти в другом потоке!
    ThreadPool.QueueUserWorkItem(__ =>
    {
        NativeMethods.Cleanup(handle); // может вызвать проблемы
    });
});

// Сравнение эквивалентных операций
 
// ThreadPool
ThreadPool.QueueUserWorkItem(_ =>
{
    try
    {
        var result = PerformCalculation(42);
        OnCalculationComplete(result);
    }
    catch (Exception ex)
    {
        OnCalculationError(ex);
    }
});
 
// Task Parallel Library
Task.Run(() => PerformCalculation(42))
    .ContinueWith(task =>
    {
        if (task.IsCompletedSuccessfully)
            OnCalculationComplete(task.Result);
        else
            OnCalculationError(task.Exception.InnerException);
    });
 
// Ещё лучше с async/await
async Task CalculateAsync()
{
    try
    {
        var result = await Task.Run(() => PerformCalculation(42));
        OnCalculationComplete(result);
    }
    catch (Exception ex)
    {
        OnCalculationError(ex);
    }
}

// В каждом микросервисе может быть такая настройка
ThreadPool.SetMinThreads(Environment.ProcessorCount * 2, Environment.ProcessorCount * 2);

public class CircuitBreaker
{
 private readonly Func<Task> _operation;
 private readonly int _failureThreshold;
 private readonly TimeSpan _resetTimeout;
 
 private int _failureCount = 0;
 private bool _isOpen = false;
 private DateTime _lastFailure = DateTime.MinValue;
 
 public CircuitBreaker(Func<Task> operation, int failureThreshold, TimeSpan resetTimeout)
 {
     _operation = operation;
     _failureThreshold = failureThreshold;
     _resetTimeout = resetTimeout;
 }
 
 public async Task ExecuteAsync()
 {
     if (_isOpen)
     {
         // Проверяем, не пора ли сбросить состояние
         if (DateTime.UtcNow - _lastFailure > _resetTimeout)
         {
             _isOpen = false;
         }
         else
         {
             throw new CircuitBreakerOpenException("Цепь разорвана");
         }
     }
     
     try
     {
         await _operation();
         // Сбрасываем счётчик при успешном выполнении
         _failureCount = 0;
     }
     catch (Exception)
     {
         _lastFailure = DateTime.UtcNow;
         _failureCount++;
         
         if (_failureCount >= _failureThreshold)
         {
             _isOpen = true;
         }
         
         throw;
     }
 }
}

public class BulkheadTaskScheduler : TaskScheduler
{
 private readonly int _maxDegreeOfParallelism;
 private readonly SemaphoreSlim _semaphore;
 private readonly ConcurrentQueue<Task> _tasks = new ConcurrentQueue<Task>();
 
 public BulkheadTaskScheduler(int maxDegreeOfParallelism)
 {
     _maxDegreeOfParallelism = maxDegreeOfParallelism;
     _semaphore = new SemaphoreSlim(maxDegreeOfParallelism);
 }
 
 protected override IEnumerable<Task> GetScheduledTasks()
 {
     return _tasks.ToArray();
 }
 
 protected override void QueueTask(Task task)
 {
     _tasks.Enqueue(task);
     ProcessTaskAsync();
 }
 
 private async void ProcessTaskAsync()
 {
     await _semaphore.WaitAsync();
     
     try
     {
         if (_tasks.TryDequeue(out Task task))
         {
             TryExecuteTask(task);
         }
     }
     finally
     {
         _semaphore.Release();
     }
 }
 
 protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
 {
     // Не выполняем задачи встроенным образом
     return false;
 }
}

// Потенциальная утечка при обработке UI-контроля
private void ProcessDataButton_Click(object sender, EventArgs e)
{
 // Захват контекста UI-потока
 var uiContext = SynchronizationContext.Current;
 
 ThreadPool.QueueUserWorkItem(_ =>
 {
     // Долгая операция
     var result = PerformHeavyCalculation();
     
     // Обновление UI через захваченный контекст
     uiContext.Post(__ =>
     {
         resultLabel.Text = result.ToString();
     }, null);
 });
}

// Утечка ресурсов из-за неосвобождённых EventWaitHandle
private void QueueLongProcessWithWaiting(string data)
{
 var completionEvent = new ManualResetEvent(false);
 
 ThreadPool.QueueUserWorkItem(_ =>
 {
     try
     {
         ProcessData(data);
     }
     finally
     {
         completionEvent.Set();
         // Забыли вызвать completionEvent.Close() или Dispose()
     }
 });
 
 // Ожидаем завершения
 completionEvent.WaitOne();
 
 // Если здесь возникнет исключение, handle не будет закрыт
 DoSomethingWithResult();
}

private void QueueLongProcessWithWaiting(string data)
{
 using (var completionEvent = new ManualResetEvent(false))
 {
     ThreadPool.QueueUserWorkItem(_ =>
     {
         try
         {
             ProcessData(data);
         }
         finally
         {
             completionEvent.Set();
         }
     });
     
     completionEvent.WaitOne();
     DoSomethingWithResult();
 } // Здесь event будет автоматически закрыт
}

private void ProcessDataButton_Click(object sender, EventArgs e)
{
 // Сохраняем только необходимые данные, а не весь контекст
 string inputData = inputTextBox.Text;
 
 ThreadPool.QueueUserWorkItem(_ =>
 {
     var result = PerformHeavyCalculation(inputData);
     
     // Используем BeginInvoke, который не захватывает лишних ссылок
     this.BeginInvoke(new Action(() =>
     {
         resultLabel.Text = result.ToString();
     }));
 });
}

// Создаём конвейер обработки
var downloadBlock = new TransformBlock<string, byte[]>(
 async url => await new HttpClient().GetByteArrayAsync(url),
 new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 });
 
var processBlock = new TransformBlock<byte[], ImageData>(
 data => ProcessImageBytes(data),
 new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount });
 
var saveBlock = new ActionBlock<ImageData>(
 async imageData => await SaveImageAsync(imageData),
 new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });
 
// Связываем блоки
downloadBlock.LinkTo(processBlock);
processBlock.LinkTo(saveBlock);
 
// Отправляем URL-адреса на обработку
foreach (var url in imageUrls)
{
 downloadBlock.Post(url);
}
downloadBlock.Complete();
 
// Ожидаем завершения всего конвейера
await saveBlock.Completion;

// Создаём канал для передачи сообщений
var channel = Channel.CreateBounded<WorkItem>(100);
 
// Задача производителя
async Task ProducerAsync(CancellationToken token)
{
 var writer = channel.Writer;
 for (int i = 0; i < 1000; i++)
 {
     if (token.IsCancellationRequested) break;
     
     await writer.WriteAsync(new WorkItem { Id = i }, token);
     await Task.Delay(10, token); // Имитация генерации данных
 }
 
 writer.Complete();
}
 
// Задача потребителя
async Task ConsumerAsync(CancellationToken token)
{
 var reader = channel.Reader;
 
 await foreach (var item in reader.ReadAllAsync(token))
 {
     ProcessWorkItem(item);
 }
}
 
// Запуск обработки
var cts = new CancellationTokenSource();
var producerTask = ProducerAsync(cts.Token);
var consumerTasks = Enumerable.Range(0, 4)
 .Select(_ => ConsumerAsync(cts.Token))
 .ToArray();
 
await Task.WhenAll(producerTask, Task.WhenAll(consumerTasks));

// Создаём поток данных
var dataStream = Observable.Interval(TimeSpan.FromMilliseconds(100))
 .Take(100)
 .Select(i => new DataPoint { Value = i })
 .ObserveOn(TaskPoolScheduler.Default) // Использование пула задач
 .Subscribe(
     dataPoint => ProcessDataPoint(dataPoint),
     ex => HandleError(ex),
     () => Console.WriteLine("Обработка завершена")
 );
 
// Канал управления
var controlStream = Observable.FromEventPattern<ControlEventArgs>(
 h => controlSource.ControlChanged += h,
 h => controlSource.ControlChanged -= h)
 .Throttle(TimeSpan.FromMilliseconds(500))
 .Subscribe(evt => HandleControlEvent(evt));