Как отменить parallel.foreach цикл изнутри задачи

Question

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

Ниже приведен код, у меня есть на данный момент

С моей консольного приложения: метод

using (IUnityContainer container = Bootstrapper.Initialise()) 
      { 
       IImageProcessor processor = container.Resolve<IImageProcessor>(); 

       processor.Container = container; 

       try 
       { 
        InputArguments arguments = new InputArguments(args); 

        if (arguments.Contains("fs")) 
        { 
         processor.Initialise(arguments["fs"]); 
        } 
        else 
        { 
         processor.Initialise(); 
        } 

        processor.Run(); 

        Console.WriteLine("\r\n\n\nPress any key to Exit"); 
        Console.ReadLine(); 

        return (int)ExitCode.Success; 
       } 
       catch (Exception ex) 
       { 
        Console.WriteLine(ex.Message); 

        Console.WriteLine("\r\n\n\nPress any key to Exit"); 
        Console.ReadLine(); 

        return (int)ExitCode.UnknownError; 
       } 
      } 

The Run

public void Run() 
     { 
      List<Task> tasks = new List<Task>(); 

      foreach (IFileService fileservice in this.fileServices) 
      { 
       Task task = Task.Factory.StartNew((arg) => 
       { 
        IFileService fs = (IFileService)arg; 

        string msg = $"Processing {fs.ToString()}..."; 

        FileLogger.Write(msg, fs.ToString()); 
        ConsoleLogger.WriteLine(msg); 
        fs.ProcessFiles(); 
        //fileservice.ReprocessUnMatchedData(); 
       }, fileservice); 

       tasks.Add(task); 
      } 

      Task.WaitAll(tasks.ToArray()); 
     } 

и внутри каждой файловой службы у меня есть вызывать этот метод:

protected bool ProcessFileRecord<T>() where T : IDataRecord 
     { 
      int matched = 0; 
      int notMatched = 0; 
      int skipped = 0; 
      bool result; 
      object lockObject = new object(); 

      try 
      { 
       processTracker = GetTracker(); 

       if (databaseHelper.TrackerFullyProcessed(processTracker)) 
       { 
        LoggingService.Write("File already processed... Skipping.", LoggingTarget.All, serviceName); 

        result = true; 
       } 

       LoggingService.Write($"\r\nProcessing index file {fileRecord.IndexFileName}", LoggingTarget.File, serviceName); 

       Parallel.ForEach(
        fileRecord.DataRecords, 
        new ParallelOptions() { MaxDegreeOfParallelism = Thread.CurrentThread.ManagedThreadId }, 
        (item) => 
        { 
         switch ((RecordProcessResult)ProcessRecord(item)) 
         { 
          case RecordProcessResult.Invalid: 
           break; 

          case RecordProcessResult.Matched: 
           Increment(ref matched); 
           break; 

          case RecordProcessResult.NotMatched: 
           Increment(ref notMatched); 
           break; 

          case RecordProcessResult.Skipped: 
           Increment(ref skipped); 
           break; 

          default: 
           break; 
         } 

         lock (lockObject) 
         { 
          if ((matched + notMatched + skipped) % 100 == 0) 
          { 
           LoggingService.Write($"\tMatched: {matched}\tNot Matched: {notMatched}\tSkipped: {skipped}\t total: {matched + notMatched + skipped}", LoggingTarget.Trace & LoggingTarget.Console, serviceName); 
          } 
         } 
        }); 

       LoggingService.Write($"Total Lines: {matched + notMatched + skipped} \r\nMatched: {matched} \r\nNot Matched: {notMatched} \r\nSkipped: {skipped}", LoggingTarget.All, serviceName); 

       this.databaseHelper.UpdateTracker(this.processTracker, matched, notMatched); 

       result = true; 
      } 
      catch (Exception ex) 
      { 
       LoggingService.Write($"Error processing data file:{fileRecord.IndexFileName}", LoggingTarget.All, serviceName); 

       LoggingService.Write($"{ex.ExceptionTreeAsString()}", LoggingTarget.All, serviceName); 

       LoggingService.Write($"Total Lines: {(matched + notMatched + skipped)} \r\nMatched: {matched} \r\nNot Matched: {notMatched} \r\nSkipped: {skipped}", LoggingTarget.All, serviceName); 

       this.databaseHelper.UpdateTracker(this.processTracker, matched, notMatched); 

       result = false; 
      } 

      return result; 
     } 

Спасибо.

Answer 1

Пожалуйста, посмотрите пример на этой странице CancellationTokenSource