Многопоточная обработка файла с.NET

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

EDIT: В идеале, вы должны изящно обрабатывать все исключения/ошибки (если они есть), чтобы у вас не было точек отказа.

Как альтернатива, вы можете иметь несколько потоков, каждый из которых "претендует" на файл, переименовывая его перед обработкой, таким образом, файловая система становится реализацией заблокированного доступа. Не знаю, будет ли это более производительным, чем мой первоначальный ответ, только тестирование покажет.

Дизайн

Шаблон "Производитель / Потребитель", вероятно, будет наиболее полезным в этой ситуации. Вы должны создать достаточно потоков, чтобы максимизировать пропускную способность.

Вот несколько вопросов о шаблоне «Производитель / Потребитель», чтобы дать вам представление о том, как он работает:

• Шаблон C # Производитель / Потребитель
• Производитель / Потребитель C #

Вы должны использовать очередь блокировки и производитель должны добавлять файлы в очередь, пока потребители обрабатывают файлы из очереди. Очередь с блокировкой не требует блокировки, поэтому это наиболее эффективный способ решения вашей проблемы.

Если вы используете .NET 4.0 есть несколько параллельных коллекций , которые вы можете использовать из коробки:

• ConcurrentQueue: http://msdn.microsoft.com/en-us/library/dd267265%28v=VS .100% 29.aspx
• BlockingCollection: http://msdn.microsoft.com/en-us/library/dd267312%28VS.100%29.aspx

Threading

Один поток производителя вероятно, будет наиболее эффективным способом загрузить файлы с диска и поместить их в очередь; впоследствии несколько потребителей будут извлекать элементы из очереди и обрабатывать их. Я бы посоветовал вам попробовать 2-4 потребительских потока на ядро ​​и провести некоторые измерения производительности, чтобы определить, какой из них наиболее оптимален (то есть количество потоков, обеспечивающих максимальную пропускную способность). Я бы не рекомендовал использовать ThreadPool для этого конкретного примера.

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

Вообще говоря, 1000 маленьких файлов (насколько маленьких, кстати?) не должны обрабатываться шесть минут. В качестве быстрого теста выполните find "foobar" * в каталоге, содержащем файлы (первый аргумент в кавычках не имеет значения; он может быть любым), и посмотрите, сколько времени потребуется на обработку каждого файла. Если это займет более одной секунды, я буду разочарован.

Если этот тест подтвердит мои подозрения, то процесс работает на CPU, и вы не получите никакого улучшения от разделения чтения на отдельный поток. Вам следует:

1. Выяснить, почему обработка небольшого ввода занимает в среднем более 350 мс, и, надеюсь, улучшить алгоритм.
2. Если нет способа ускорить алгоритм и у вас есть многоядерная машина (в наши дни почти у всех), используйте пул потоков, чтобы назначить 1000 задач на чтение одного файла.

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

Код примера является сокращенным демонстратором, но он работает.

class Program
{
    static void Main(string[] args)
    {
        Supervisor super = new Supervisor();
        super.LaunchWaitingThreads();

        while (!super.Done) { Thread.Sleep(200); }
        Console.WriteLine("\nDone");
        Console.ReadKey();
    }
}

public delegate void StartCallbackDelegate(int idArg, Worker workerArg);
public delegate void DoneCallbackDelegate(int idArg);

public class Supervisor
{
    Queue<Thread> waitingThreads = new Queue<Thread>();
    Dictionary<int, Worker> runningThreads = new Dictionary<int, Worker>();
    int maxThreads = 20;
    object locker = new object();

    public bool Done { 
        get { 
            lock (locker) {
                return ((waitingThreads.Count == 0) && (runningThreads.Count == 0)); 
            } 
        } 
    }

    public Supervisor()
    {
        // queue up a thread for each file
        Directory.GetFiles("C:\\folder").ToList().ForEach(n => waitingThreads.Enqueue(CreateThread(n)));
    }

    Thread CreateThread(string fileNameArg)
    {
        Thread thread = new Thread(new Worker(fileNameArg, WorkerStart, WorkerDone).ProcessFile);
        thread.IsBackground = true;
        return thread;
    }

    // called when a worker starts
    public void WorkerStart(int threadIdArg, Worker workerArg)
    {
        lock (locker)
        {
            // update with worker instance
            runningThreads[threadIdArg] = workerArg;
        }
    }

    // called when a worker finishes
    public void WorkerDone(int threadIdArg)
    {
        lock (locker)
        {
            runningThreads.Remove(threadIdArg);
        }
        Console.WriteLine(string.Format("  Thread {0} done", threadIdArg.ToString()));
        LaunchWaitingThreads();
    }

    // launches workers until max is reached
    public void LaunchWaitingThreads()
    {
        lock (locker)
        {
            while ((runningThreads.Count < maxThreads) && (waitingThreads.Count > 0))
            {
                Thread thread = waitingThreads.Dequeue();
                runningThreads.Add(thread.ManagedThreadId, null); // place holder so count is accurate
                thread.Start();
            }
        }
    }
}

public class Worker
{
    string fileName;
    StartCallbackDelegate startCallback;
    DoneCallbackDelegate doneCallback;
    public Worker(string fileNameArg, StartCallbackDelegate startCallbackArg, DoneCallbackDelegate doneCallbackArg)
    {
        fileName = fileNameArg;
        startCallback = startCallbackArg;
        doneCallback = doneCallbackArg;
    }

    public void ProcessFile()
    {
        startCallback(Thread.CurrentThread.ManagedThreadId, this);
        Console.WriteLine(string.Format("Reading file {0} on thread {1}", fileName, Thread.CurrentThread.ManagedThreadId.ToString()));
        File.ReadAllBytes(fileName);
        doneCallback(Thread.CurrentThread.ManagedThreadId);
    }
}

Вы можете рассмотреть очередь файлов для обработки. Заполните очередь один раз, сканируя каталог при запуске и обновив очередь с помощью FileSystemWatcher , чтобы эффективно добавлять новые файлы в очередь без постоянного повторного сканирования каталога.

По возможности считывайте и записывайте на разные физические диски. Это даст вам максимальную производительность ввода-вывода.

Если у вас есть первоначальный пакет из большого количества файлов для обработки, а затем неравномерная скорость добавления новых файлов, и все это происходит на одном диске (чтение / запись), вы можете рассмотреть возможность буферизации обработанных файлы в память, пока не будет выполнено одно из двух условий:

• Нет (временно) новых файлов
• Вы поместили в буфер так много файлов, что вы не хотите использовать больше памяти для {{1} } буферизация (в идеале настраиваемый порог)

Если ваша фактическая обработка файлов требует интенсивной работы ЦП, вы можете подумать о наличии одного потока обработки на ядро ​​ЦП. Однако для «нормальной» обработки время ЦП будет тривиальным по сравнению с временем ввода-вывода, и сложность не будет стоить каких-либо незначительных выигрышей.