Как обрабатывать файлы каталога в параллельной библиотеке задач?

Question

У меня есть сценарий, в котором я должен обрабатывать несколько файлов (например, 30) параллельно на основе процессорных ядер. Я должен назначить эти файлы для разделения задач на основе отсутствия процессорных ядер. Я не знаю, как сделать начальный и конечный предел каждой задачи для обработки. Например, каждая задача знает, сколько файлов она должна обрабатывать.

private void ProcessFiles(object e) 
    { 
     try 
     { 
      var diectoryPath = _Configurations.Descendants().SingleOrDefault(Pr => Pr.Name == "DirectoryPath").Value; 

      var FilePaths = Directory.EnumerateFiles(diectoryPath); 
      int numCores = System.Environment.ProcessorCount; 
      int NoOfTasks = FilePaths.Count() > numCores ? (FilePaths.Count()/ numCores) : FilePaths.Count(); 


      for (int i = 0; i < NoOfTasks; i++) 
      { 
       Task.Factory.StartNew(
         () => 
         { 
          int startIndex = 0, endIndex = 0; 
          for (int Count = startIndex; Count < endIndex; Count++) 
          { 
           this.ProcessFile(FilePaths); 
          } 
         }); 

      } 
     } 
     catch (Exception ex) 
     { 
      throw; 
     } 
    } 

Answer 1

Для таких проблем, как у тебя, есть параллельные структуры данных доступно на C#. Вы хотите использовать BlockingCollection и хранить в нем все имена файлов.

Ваша идея расчета количества задач с использованием количества ядер, доступных на машине, не очень хороша. Зачем? Потому что ProcessFile() может не принимать одинаковое время для каждого файла. Итак, было бы лучше начать количество задач, как количество ядер, которые у вас есть. Затем пусть каждая задача будет читать имя файла один за другим из BlockingCollection и затем обрабатывать файл, пока BlockingCollection не будет пустым.

try 
{ 
    var directoryPath = _Configurations.Descendants().SingleOrDefault(Pr => Pr.Name == "DirectoryPath").Value; 

    var filePaths = CreateBlockingCollection(directoryPath); 
    //Start the same #tasks as the #cores (Assuming that #files > #cores) 
    int taskCount = System.Environment.ProcessorCount; 

    for (int i = 0; i < taskCount; i++) 
    { 
     Task.Factory.StartNew(
       () => 
       { 
        string fileName; 
        while (!filePaths.IsCompleted) 
        { 
         if (!filePaths.TryTake(out fileName)) continue; 
         this.ProcessFile(fileName); 
        } 
       }); 
    } 
} 

И CreateBlockingCollection() будет выглядеть следующим образом:

private BlockingCollection<string> CreateBlockingCollection(string path) 
{ 
    var allFiles = Directory.EnumerateFiles(path); 
    var filePaths = new BlockingCollection<string>(allFiles.Count); 
    foreach(var fileName in allFiles) 
    { 
     filePaths.Add(fileName); 
    } 
    filePaths.CompleteAdding(); 
    return filePaths; 
} 

Вы должны изменить ваш ProcessFile() получить имя файла в настоящее время вместо того, чтобы все пути к файлам и обработки его кусок.

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

---

_{Я не запускать код сам, так что может быть какая-то ошибка синтаксиса в моем коде. Не стесняйтесь исправить ошибку, если вы столкнетесь с ней.}

Answer 2

Основываясь на моем заведомо ограниченное понимании TPL, я думаю, что ваш код может быть переписан так:

private void ProcessFiles(object e) 
{ 
    try 
    { 
     var diectoryPath = _Configurations.Descendants().SingleOrDefault(Pr => Pr.Name == "DirectoryPath").Value; 

     var FilePaths = Directory.EnumerateFiles(diectoryPath); 

     Parallel.ForEach(FilePaths, path => this.ProcessFile(path)); 

    } 
    catch (Exception ex) 
    { 
     throw; 
    } 
} 

приветы