Использование миллионов Task-ов приводит к ошибке System.OutOfMemoryException (C#)

Question

У меня есть 6 миллионов небольших (средний размер около 15 байт) файлов, которые мне нужно прочитать и далее обработать с помощью процессора. Я ранее реализовал это с помощью Task.Factory и всё работало без проблем на asp .net core 2.1. Занимало по времени около 20 часов.

Теперь я перенес приложение на asp.net 6 и на тестовом сервере мое веб-приложение перестает отвечать на любые запросы после запуска этих файловых операций и аварийно завершается. В логах я вижу ошибку System.OutOfMemoryException.

Я полагаю, что мой способ реализации далек от идеала. Я хотел бы узнать какие-нибудь иные способы многопоточной реализации этой работы или ваши замечания по текущему коду.

Метод контроллера ImportSignatures:

[HttpPost("ImportSignatures")]
public async Task<JsonResult> ImportSignatures()
{
    try
    {
        ImportSigningCertsResult res = await SignatureImportService.ImportSigningCerts();
        return Json(res);
    }
    catch (Exception e)
    {
        LogsHelper.WriteLog("api/Settings/ImportSignatures", e);
        return Json(new ImportSigningCertsResult(e.Message, SignatureImportService.WasCancelled));
    }
}

Метод ImportSigningCerts:

public static async Task<ImportSigningCertsResult> ImportSigningCerts()
{
    LogsHelper.WriteEventLog("Запуск SignatureImportService");
    WasCancelled = false;
    IsWorking = true;
    ResultStr = "";
    totalSignatures = 0;
    processedSignatures = 0;
var cancelMsg = &quot;Импорт сертификатов был прерван. \n&quot;;
var endMsg = &quot;Импорт сертификатов успешно завершён. \n&quot;;
var toDelete = new List&lt;string&gt;();

try
{
    var configuration = SignatureImportConfiguration.FromCfg();

    using (s_tokenSource = new CancellationTokenSource())
    {
        IEnumerable&lt;string&gt; signatures = Directory.EnumerateFiles(configuration.Path, &quot;*.sig&quot;);
        totalSignatures = signatures.Count();

        Store mainStore = StoreMan.GetStore(&quot;Main&quot;);
        var importStats = new ImportStats();
        List&lt;Task&gt; tasks = new();

        int saveIndex = 1;
        const int proccessedForSave = 100000; // Через какое кол-во обработанных подписей произвести промежуточное сохранение хранилища и удаление подписей
        CancellationToken token = s_tokenSource.Token;

        ThreadPool.GetMinThreads(out int minWorkerThreads, out _);
        using SemaphoreSlim semaphore = new(minWorkerThreads);

        foreach (string path in signatures)
        {
            semaphore.Wait();

            if (WasCancelled)
                break;

            tasks.Add(Task.Factory.StartNew(() =&gt;
            {
                try
                {
                    token.ThrowIfCancellationRequested();

                    if (UploadSigningCerts(mainStore, path, importStats))
                    {
                        if (configuration.NeedCleaning)
                        {
                            lock (s_toDeleteListLockObj)
                                toDelete.Add(path);
                        }
                    }

                    Interlocked.Increment(ref processedSignatures);
                    lock (s_intermediateSaveLockObj)
                    {
                        if (processedSignatures &gt; proccessedForSave * saveIndex)
                        {
                            LogsHelper.WriteEventLog(&quot;Промежуточное сохранение хранилища сертификатов...&quot;);

                            mainStore.WriteIfChanged();
                            StartRemovingSignatures(toDelete);
                            saveIndex++;
                        }
                    }
                }
                catch (Exception e)
                {
                    if (e is not OperationCanceledException)
                        LogsHelper.WriteLog(&quot;SignatureImportService/ImportSigningCerts:Task.Factory.StartNew&quot;, e);
                }
                finally
                {
                    semaphore.Release();
                }
            }, token));
        }

        try
        {
            await Task.WhenAll(tasks);
        }
        catch (OperationCanceledException) { }

        mainStore.WriteIfChanged();
        StartRemovingSignatures(toDelete);
        ResultStr = (WasCancelled ? cancelMsg : endMsg) + $&quot;Certificates found: {importStats.all}. Was imported: {importStats.imported}.&quot; + (importStats.parsingFailed &gt; 0 ? $&quot; Unrecognized files: {importStats.parsingFailed}&quot; : &quot;&quot;);
    }

    LogsHelper.WriteEventLog(ResultStr);
    return s_tokenSource == null ? new ImportSigningCertsResult(ResultStr) : new ImportSigningCertsResult(ResultStr, WasCancelled);
}
finally
{
    IsWorking = false;
}

}

Метод UploadSigningCerts:

private static bool UploadSigningCerts(Store store, string path, ImportStats importStats)
{
    bool toBeDeleted = true;
    CryptoClient client = CryptoServiceContext.DefaultInstance.CryptoClient;
try
{
    List<CertInfo> certs = client.GetSignCmsInfo(File.ReadAllBytes(path)).Certs.ToList();

    Interlocked.Add(ref importStats.all, certs.Count);

    for (int i = 0; i < certs.Count; i++)
    {
        lock (s_importLockObj)
        {
            // Код по валидации каждого сертификата из файла, принятие решения об импорте, импорт в хранилище...
        }
    }
    return toBeDeleted;
}
catch (Exception e)
{
    LogsHelper.WriteLog("SignatureImportService/UploadSigningCerts", e);
    LogsHelper.WriteEventLog($"Ошибка импорта сертификата из подписи: {Path.GetFileName(path)};");
    Interlocked.Increment(ref importStats.errors);
    return false;
}

}

Метод StartRemovingSignatures:

private static void StartRemovingSignatures(List<string> toDelete)
{
    if (toDelete.Count > 0)
    {
        List<string> tempToDelete;
        lock (s_toDeleteListLockObj)
        {
            tempToDelete = new List<string>(toDelete);
            toDelete.Clear();
        }
    LogsHelper.WriteEventLog(&quot;Удаление успешно обработанных файлов подписей...&quot;);

    Task.Factory.StartNew(() =&gt;
    {
        tempToDelete.ForEach(path =&gt;
        {
            try
            {
                File.Delete(path);
            }
            catch (Exception e)
            {
                LogsHelper.WriteLog(&quot;ImportResult/DeleteSignatures&quot;, e);
            }
        });
    });
}

}

Текст ошибки:

20.08.2023 11:58:01 api/Settings/ImportSignatures
Exception of type 'System.OutOfMemoryException' was thrown.
   at System.Threading.Tasks.Task.EnsureContingentPropertiesInitializedUnsafe()
   at System.Threading.Tasks.Task.AssignCancellationToken(CancellationToken cancellationToken, Task antecedent, TaskContinuation continuation)
   at System.Threading.Tasks.Task.TaskConstructorCore(Delegate action, Object state, CancellationToken cancellationToken, TaskCreationOptions creationOptions, InternalTaskOptions internalOptions, TaskScheduler scheduler)
   at Store.Services.SignatureImportService.<>c__DisplayClass20_0.<ImportSigningCerts>b__0(String path)
   at System.Collections.Generic.List`1.ForEach(Action`1 action)
   at Store.Services.SignatureImportService.ImportSigningCerts()
   at Store.Controllers.SettingsController.ImportSignatures()

Updated: код был отредактирован

Answer 1

Давайте попробуем этот код стабилизировать

• нужно избавиться от лишних делегатов
• нет смысла гонять файловую систему из кучи потоков, она синхронная, она не может в рамках одного носителя выполнять кучу операций одновременно
• нет смысла трогать настройки пула потоков, он нормально настроен, просто нужно относиться к нему бережно
• нельзя блокировать поток контроллера, так вы вешаете весь сервер, для этого есть асинхронные операции
• много локов - плохо для понимания кода и производительности, используйте потокобезопасные коллекции
• процедура чтения int переменной атомарна сама по себе, поэтому lock не нужен для чтения, избавиться от лишних локов
• полностью избавиться от статики
• ограничить количество одновременно работающих задач до количества ядер сервера, умноженное на два (так исторически сложилось, что это оптимально, с этого надо начинать)
• каждый созданный вами Task должен быть ожидаемым, так вы получаете контроль над процессом и удобнее ловить исключения
• удаление файлов можно вообще через Producer/Consumer сделать, чтобы удалялись сразу как появляются в череди на удаление, при этом не тормозили основной код.

Вот и получается как-то так

[HttpPost("ImportSignatures")]
public async Task<JsonResult> ImportSignatures()
{
    SignatureImportService importService = new SignatureImportService();
    try
    {
        ImportSigningCertsResult res = await importService.ImportSigningCerts();
        return Json(res);
    }
    catch (Exception e)
    {
        LogsHelper.WriteLog("api/Settings/ImportSignatures", e);
        return Json(new ImportSigningCertsResult(e.Message, importService.WasCancelled));
    }
}

public async Task<ImportSigningCertsResult> ImportSigningCerts()
{
    LogsHelper.WriteEventLog("Запуск SignatureImportService");
const string cancelMsg = &quot;Импорт сертификатов был прерван. \n&quot;;
const string endMsg = &quot;Импорт сертификатов успешно завершён. \n&quot;;
try
{
    using (_cts = new CancellationTokenSource())
    {
        var configuration = SignatureImportConfiguration.FromCfg();
        string[] signatures = Directory.GetFiles(configuration.Path, &quot;*.sig&quot;);
        totalSignatures = signatures.Length;
        processedSignatures = 0;

        Store mainStore = StoreMan.GetStore(&quot;Main&quot;);
        var importStats = new ImportStats();
        List&lt;Task&gt; tasks = new();

        int saveIndex = 1;
        const int proccessedForSave = 100000; // Через какое кол-во обработанных подписей произвести промежуточное сохранение хранилища и удаление подписей
        CancellationToken token = _cts.Token;
        Channel&lt;string&gt; channel = Channel.CreateUnbounded&lt;string&gt;();
        ChannelWriter&lt;string&gt; deleteQueue = channel.Writer;
        Task deleteTask = DeleteWorkerAsync(channel.Reader, token);
        object saveSync = new object();
        int maxJobs = Environment.ProcessorCount * 2;
        using SemaphoreSlim semaphore = new(maxJobs);
        try
        {
            foreach (string path in signatures)
            {
                await semaphore.WaitAsync(token);
                if (tasks.Count &gt; maxJobs * 2)
                {
                    for (int i = 0; i &lt; tasks.Count; i++) // не позволит списку тасков расти бесконечно
                    {
                        if (tasks[i].IsCompletedSuccessfully)
                        {
                            tasks.RemoveAt(i--);
                        }
                    }
                }
                tasks.Add(Task.Run(() =&gt;
                {
                    try
                    {
                        token.ThrowIfCancellationRequested();

                        if (UploadSigningCerts(mainStore, path, importStats) &amp;&amp; configuration.NeedCleaning)
                        {
                            deleteQueue.TryWrite(path);
                        }
                        bool needSave = false;
                        lock (saveSync)
                        {
                            if (++processedSignatures &gt; proccessedForSave * saveIndex)
                            {
                                saveIndex++;
                                needSave = true;
                            }
                        }
                        if (needSave)
                        {
                            LogsHelper.WriteEventLog(&quot;Промежуточное сохранение хранилища сертификатов...&quot;);
                            mainStore.WriteIfChanged();
                        }
                    }
                    catch (Exception e) when (e is not OperationCanceledException)
                    {
                        LogsHelper.WriteLog(&quot;SignatureImportService/ImportSigningCerts:Task.Factory.StartNew&quot;, e);
                    }
                    finally
                    {
                        semaphore.Release();
                    }
                }, token));
            }

            await Task.WhenAll(tasks);
            mainStore.WriteIfChanged();
            deleteQueue.Complete();
            await deleteTask;
        }
        catch (OperationCanceledException) { }

        string result = (token.IsCancellationRequested ? cancelMsg : endMsg) + $&quot;Certificates found: {importStats.all}. Was imported: {importStats.imported}.&quot; + (importStats.parsingFailed &gt; 0 ? $&quot; Unrecognized files: {importStats.parsingFailed}&quot; : &quot;&quot;);
        LogsHelper.WriteEventLog(result);

        return new ImportSigningCertsResult(result, token.IsCancellationRequested);
    }
}
finally
{
    _cts = null;
}

}
private bool UploadSigningCerts(Store store, string path, ImportStats importStats)
{
    bool toBeDeleted = true;
    CryptoClient client = CryptoServiceContext.DefaultInstance.CryptoClient;
try
{
    CertInfo[] certs = client.GetSignCmsInfo(File.ReadAllBytes(path)).Certs.ToArray(); // здесь непонятно, что такое за коллекция Certs и зачем ее клонировать

    Interlocked.Add(ref importStats.all, certs.Length);

    for (int i = 0; i &lt; certs.Length; i++)
    {

    }
    return toBeDeleted;
}
catch (Exception e)
{
    LogsHelper.WriteLog(&quot;SignatureImportService/UploadSigningCerts&quot;, e);
    LogsHelper.WriteEventLog($&quot;Ошибка импорта сертификата из подписи: {Path.GetFileName(path)};&quot;);
    Interlocked.Increment(ref importStats.errors);
    return false;
}

}
private async Task DeleteWorkerAsync(ChannelReader<string> reader, CancellationToken token)
{
    await foreach (string path in reader.ReadAllAsync(token))
    {
        try
        {
            File.Delete(path); // если файловая система медленная, здесь можно await Task.Run(() => File.Delete(path));
        }
        catch (Exception e)
        {
            LogsHelper.WriteLog("ImportResult/DeleteSignatures", e);
        }
    }
}

bool переменные тоже бессмысленные, у вас есть потокобезопасный CTS, его и используйте, например так

public bool WasCancelled => _cts?.IsCancellationRequested ?? false;
public bool IsWorking => !_cts?.IsCancellationRequested ?? false;
private CancellationTokenSource _cts;

Избавляйтесь от статики дальше. И отвыкайте от списков в пользу массивов, если не надо динамически добавлять и удалять элементы из списка. Массивы легче и работают быстрее.

Вот еще почитайте в тему: c# как ускорить добавление потоков используя async-await