使用数百万个任务会导致 System.OutOfMemoryException (C#)

我有 600 万个小文件（平均大小约为 15 字节），我需要读取这些文件，然后使用处理器进行处理。我之前已经实现过这个，Task.Factory并且它在 asp .net core 2.1 上运行没有问题。花了大约20个小时。

我现在已将应用程序移植到 asp.net 6，并且在测试服务器上，我的 Web 应用程序在运行这些文件操作并崩溃后停止响应任何请求。我在日志中看到错误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 = "Импорт сертификатов был прерван. \n";
    var endMsg = "Импорт сертификатов успешно завершён. \n";
    var toDelete = new List<string>();

    try
    {
        var configuration = SignatureImportConfiguration.FromCfg();

        using (s_tokenSource = new CancellationTokenSource())
        {
            IEnumerable<string> signatures = Directory.EnumerateFiles(configuration.Path, "*.sig");
            totalSignatures = signatures.Count();

            Store mainStore = StoreMan.GetStore("Main");
            var importStats = new ImportStats();
            List<Task> 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(() =>
                {
                    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 > proccessedForSave * saveIndex)
                            {
                                LogsHelper.WriteEventLog("Промежуточное сохранение хранилища сертификатов...");

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

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

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

        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("Удаление успешно обработанных файлов подписей...");

        Task.Factory.StartNew(() =>
        {
            tempToDelete.ForEach(path =>
            {
                try
                {
                    File.Delete(path);
                }
                catch (Exception e)
                {
                    LogsHelper.WriteLog("ImportResult/DeleteSignatures", 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()

更新：代码已被编辑