Параллельно в .NET 3.5 - потоки, не завершающие свои задачи в Nested For

Question

Некоторое время назад я скачал Parallels Extensions для .NET 3.5 SP1 на DevLabs и начал его использовать.Казалось, что все работает нормально, пока я не заметил в журналах некоторые исключения, которые мне трудно понять, почему они произошли.

Вот фрагмент параллельной задачи:

Parallel.ForEach(myJobArray, currentJob =>
{
    JobElements myJobElements = GetJobElements(currentJob);

    Parallel.For(0, myJobElements.Length, (currentIndex, loopState) =>
    {
        if (MyFunction(param1, myJobElements[currentIndex]))
        {
            loopState.Stop();
        }
    }
    );
}
);

Вотпсевдокод MyFunction:

private bool MyFunction(MyObject1 param1, MyObject2 param2)
{
    log(string.Format("start SubFunction1() from thread {0}", Thread.CurrentThread.ManagedThreadId));
    SubFunction1(); //which uses System.Diagnostics.Process to start a batch file (.bat) to execute a Perl script. If successful, a file will be generated.
    log(string.Format("end SubFunction1() from thread {0}", Thread.CurrentThread.ManagedThreadId));

    log(string.Format("start SubFunction2() from thread {0}", Thread.CurrentThread.ManagedThreadId));
    SubFunction2(); //which again, uses System.Diagnostics.Process to start another batch file (.bat) to execute a Perl script which transforms the file from step #1 to a new file.
    log(string.Format("end SubFunction2() from thread {0}", Thread.CurrentThread.ManagedThreadId));

}

В журналах показано, что один поток запускает SubFunction1 (), но никогда не завершается;т. е. не было записи в журнале, в которой говорилось бы что-то вроде «end SubFunction1 ()» с тем же идентификатором потока.Фактически, тот же поток, похоже, перешел к следующему заданию в массиве и снова вызвал SubFunction1 ().Исключение произошло, когда другой поток попытался заменить предыдущий поток для запуска SubFunction2 () и не смог найти файл, сгенерированный из SubFunction1 ().

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

Есть мысли?

Answer 1

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

Во-первых, это не гарантируется, поскольку вы используете loopState.Stop();, и из вашего кода неясно, как, когда и что возвращает MyFunction().

Тогда, что делает задачи / потоки и что записывается в журнал (файл) - это две разные вещи, которые, скорее всего, не отражают друг друга:

• Если журнал пишет в один и тот же файл, то наиболее вероятно, что журнал использует отдельный поток для записи. Тогда log(string.Format("end SubFunction1() from thread {0}", Thread.CurrentThread.ManagedThreadId)); не показывает фактический идентификатор потока задачи (но поток журнала). Если он действительно показывает идентификатор потока фактической задачи, тогда неясно (из вашего кода), почему вы не получаете межпотоковое исключение, записывая в журнал несколько потоков
• В любом случае, запись в один и тот же файл должна была быть синхронизирована, в противном случае неудивительно, что следующая задача, повторно использующая тот же поток, перезаписывает выходные данные предыдущей задачи (как в содержимом буфера, так и в содержимом файла)

Вы должны синхронизировать запись в журнале:

• с использованием вывода в массив (который является потокобезопасным) и его регистрации после завершения задачи
• упаковка вашего журнала в ActionBlock потока данных TPL (или, что то же самое, с использованием TP с очередью)
• использование блокировки внутри Parallel.For очистки потока выходного потока в конце каждого цикла для защиты доступа к журналу

Answer 2

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

Этого не произойдет, если SubFunction1() сгенерирует исключение.