Task.WaitAll() при выполнении задачи только задерживает выполнение исходной задачи?

Question

фон:

У меня есть консольное приложение, которое создает Tasks для обработки данных из БДА (давайте называть их LEVEL1 задачами). Каждая из задач снова создает свои собственные Задачи для обработки каждой части данных, которые ей были назначены (задачи уровня 2).

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

Я на .NET 4.0 (нет асинхронной/ждать)

Выпуск:

Это создало проблему, хотя - оказалось, что если сделано таким образом, ни одна из задач LEVEL2 не начато до все запланированные задачи Level1 были запланированы. Это не оптимально.

Вопрос:

Это, кажется, фиксируется с изменением кода ждать как исходной задачи Level2 и продолжение задачи. Однако я не совсем уверен, почему это так?

Есть ли у вас идеи?

Единственное, что я мог придумать, это то, что, поскольку задача продолжения не началась, нет смысла ждать ее завершения? Но даже если бы это было так, я ожидал бы, по крайней мере, НЕКОТОРЫХ из задач Level2, которые нужно было запустить. Которые они никогда не делали.

Пример:

Я создал образец консольного приложения, которые продемонстрировали именно такое поведение:

1. Run это как есть, и вы увидите, что это планирование всех задач, во-первых, и только потом вы начинаете получать фактические сценарии из задач Level2.

2. Но закомментируйте отмеченный блок кода и раскомментируйте замену, и все работает как ожидалось.

Можете ли вы сказать мне, почему?

public class Program 
{ 
    static void Main(string[] args) 
    { 
     for (var i = 0; i < 100; i++) 
     { 
      Task.Factory.StartNew(() => SomeMethod()); 
      //Thread.Sleep(1000); 
     } 

     Console.ReadLine(); 
    } 

    private static void SomeMethod() 
    { 
     var numbers = new List<int>(); 

     for (var i = 0; i < 10; i++) 
     { 
      numbers.Add(i); 
     } 

     var tasks = new List<Task>(); 

     foreach (var number in numbers) 
     { 
      Console.WriteLine("Before start task"); 

      var numberSafe = number; 

      /* Code to be replaced START */ 

      var nextTask = Task.Factory.StartNew(() => 
      { 
       Console.WriteLine("Got number: {0}", numberSafe); 
      }) 
       .ContinueWith(task => 
       { 
        Console.WriteLine("Continuation {0}", task.Id); 
       }); 

      tasks.Add(nextTask); 

      /* Code to be replaced END */ 

      /* Replacement START */ 

      //var originalTask = Task.Factory.StartNew(() => 
      //{ 
      // Console.WriteLine("Got number: {0}", numberSafe); 
      //}); 

      //var contTask = originalTask 
      // .ContinueWith(task => 
      // { 
      //  Console.WriteLine("Continuation {0}", task.Id); 
      // }); 

      //tasks.Add(originalTask); 
      //tasks.Add(contTask); 

      /* Replacement END */ 
     } 

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

Answer 1

Я думаю, что вы видите поведение Task Inlining. Цитируя MSDN:

В некоторых случаях, когда задача прислуживала, она может быть выполнена синхронно на тему, выполняющая операцию ожидания. Это повышает производительность, поскольку это предотвращает необходимость в дополнительном потоке, используя существующий Thread, который заблокировал бы, в противном случае.Чтобы предотвратить ошибки из-за повторного входа, сглаживание задачи происходит только тогда, когда цель ожидания найдена в локальной очереди соответствующего потока.

Вам не нужно 100 задач, чтобы это увидеть. Я изменил вашу программу, чтобы иметь 4 задания уровня 1 (у меня есть четырехъядерный процессор). Каждая задача уровня 1 создает только одну задачу уровня 2.

static void Main(string[] args) 
{ 
    for (var i = 0; i < 4; i++) 
    { 
     int j = i; 
     Task.Factory.StartNew(() => SomeMethod(j)); // j as level number 
    } 
} 

В исходной программе nextTask задача продолжения - так что я просто упростил метод.

private static void SomeMethod(int num) 
{ 
    var numbers = new List<int>(); 

    // create only one level 2 task for representation purpose 
    for (var i = 0; i < 1; i++) 
    { 
     numbers.Add(i); 
    } 

    var tasks = new List<Task>(); 

    foreach (var number in numbers) 
    { 
     Console.WriteLine("Before start task: {0} - thread {1}", num, 
           Thread.CurrentThread.ManagedThreadId); 

     var numberSafe = number; 

     var originalTask = Task.Factory.StartNew(() => 
     { 
      Console.WriteLine("Got number: {0} - thread {1}", num, 
            Thread.CurrentThread.ManagedThreadId); 
     }); 

     var contTask = originalTask 
      .ContinueWith(task => 
      { 
       Console.WriteLine("Continuation {0} - thread {1}", num, 
            Thread.CurrentThread.ManagedThreadId); 
      }); 

     tasks.Add(originalTask); // comment and un-comment this line to see change in behavior 

     tasks.Add(contTask); // same as adding nextTask in your original prog. 

    } 

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

Вот пример вывода - на комментирование tasks.Add(originalTask); - что это ваш первый блок.

Before start task: 0 - thread 4 
Before start task: 2 - thread 3 
Before start task: 3 - thread 6 
Before start task: 1 - thread 5 
Got number: 0 - thread 7 
Continuation 0 - thread 7 
Got number: 1 - thread 7 
Continuation 1 - thread 7 
Got number: 3 - thread 7 
Continuation 3 - thread 7 
Got number: 2 - thread 4 
Continuation 2 - thread 4 

И пример вывода - на сохранении tasks.Add(originalTask); что ваш второй блок

Before start task: 0 - thread 4 
Before start task: 1 - thread 6 
Before start task: 2 - thread 5 
Got number: 0 - thread 4 
Before start task: 3 - thread 3 
Got number: 3 - thread 3 
Got number: 1 - thread 6 
Got number: 2 - thread 5 
Continuation 0 - thread 7 
Continuation 1 - thread 7 
Continuation 3 - thread 7 
Continuation 2 - thread 4 

Как вы можете видеть, во втором случае, когда вы ждете originalTask на том же потоке, который начал его, task inlining сделает он запускается в том же потоке - вот почему вы видите сообщения Got Number.. раньше.

Answer 2

Я должен сказать, что этот код действительно не оптимистичны, как вы создаете 100 задач, и это не означает, что вы будете иметь 100 потоков, и внутри каждой задачи вы создаете две новые задачи, вы более superscribing планировщик. если эти задачи связаны с чтением db, почему бы не отметить их как длительную обработку и отказаться от внутренних задач?

Answer 3

Если я правильно помню, ожидание выполнения задачи, которая еще не была запланирована, может выполняться синхронно. (см. here). Было бы неудивительно, что это поведение применимо к вашему коду в альтернативном случае.

Имея в виду, что поведение потоковая сильно реализации- и машинно-зависимый, что здесь происходит, вероятно, что-то в строках этого:

• Учитывая задержку между вызовом Task.StartNew и фактическим исполнением задачи в threadpool, большинство заданий «Уровень 1», если не все из них, запланированы до того, как первая из них фактически будет выполнена.
• Поскольку планировщик задач по умолчанию использует .NET ThreadPool, все запланированные задачи, скорее всего, будут выполняться в потоке ThreadPool.
• После выполнения задачи «Уровень 1» очереди (-ы) планирования заполняются задачами «Уровень 1».
• Каждый раз, когда выполняется задача «Уровень 1», она назначает столько задач «Уровень 2», сколько требуется, но все они запланированы после задач «Уровень 1».
• Когда задача «Уровень 1» переходит в ожидании всех продолжений задач «Уровень 2», исполняемый поток переходит в состояние ожидания.
• Со многими Threadpool потоков в состоянии ожидания, программа быстро достигает ThreadPool голодание, заставляя ThreadPool выделить новые темы (вероятно, более 100 в общей сложности), чтобы разрешить голодание
• После того, как последний из «Уровня 1» задачи получают вызов состояния ожидания, ThreadPool выделяет по крайней мере еще один дополнительный поток.
• Этот выделенный выделенный поток теперь может выполнять задачи «Уровень 2» и их продолжение в первый раз, поскольку все задачи «Уровень 1» выполняются с.
• Спустя некоторое время одна задача «Уровень 1» будет содержать все его задачи «Уровень 2». Эта задача «Уровень 1» затем проснется от ожидания и завершит ее выполнение, освободив тем самым поток ThreadPool и ускоряя выполнение оставшихся задач и продолжения «Уровня 2».

Что изменилось при использовании вашего альтернативного метода, так это то, что вы ссылаетесь на задачу «Уровень 2» непосредственно в массиве задач для ожидания, метод Task.WaitAll получает возможность выполнить «Уровень 2 «задачи синхронно, а не простоя. Это не может произойти в начальном случае, поскольку задачи продолжения не могут выполняться синхронно.

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

Чтобы решить вашу первоначальную проблему, вам следует лучше следовать предложению lil-raz о том, чтобы избавиться от ваших внутренних задач.

Если у вас есть доступ к C# 5.0, вы также можете рассмотреть возможность использования шаблона async/await для написания кода, не полагаясь на ожидание.

Answer 4

Проблема с вашим кодом блокировкаTask.WaitAll(tasks.ToArray()). Планировщик заданий по умолчанию TPL не будет использовать новый поток пула для каждой задачи, которую вы начинаете с Factory.StartNew. И вы запускаете 100 задач Level1, каждый из которых блокирует поток с помощью Task.WaitAll.

Это создает узкое место. С размером по умолчанию ThreadPool я получаю ~ 20 потоков, работающих одновременно, причем только 4 из них фактически выполняются одновременно (количество ядер процессора).

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

static void Main(string[] args) 
{ 
    for (var i = 0; i < 100; i++) 
    { 
     Task.Factory.StartNew(() => SomeMethod(), 
      TaskCreationOptions.LongRunning); 
    } 

    Console.ReadLine(); 
} 

TaskCreationOptions.LongRunning даст вам желаемое поведение, но это, конечно, было бы неправильно решения.

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

Чтобы решить эту проблему, ваш код можно переустановить, как показано ниже. Обратите внимание на использование ContinueWhenAll, Unwrap и (необязательно) ExecuteSynchronously, что помогает устранить блокирующий код и уменьшить количество задействованных потоков пулов. Эта версия работает намного лучше.

using System; 
using System.Collections.Generic; 
using System.Threading.Tasks; 

public class Program 
{ 
    static void Main(string[] args) 
    { 
     var tasks = new List<Task>(); 

     for (var i = 0; i < 100; i++) 
     { 
      tasks.Add(Task.Factory.StartNew(() => SomeMethod(i)).Unwrap()); 
     } 

     // blocking at the topmost level 
     Task.WaitAll(tasks.ToArray()); 

     Console.WriteLine("Enter to exit..."); 
     Console.ReadLine(); 
    } 

    private static Task<Task[]> SomeMethod(int n) 
    { 
     Console.WriteLine("SomeMethod " + n); 

     var numbers = new List<int>(); 

     for (var i = 0; i < 10; i++) 
     { 
      numbers.Add(i); 
     } 

     var tasks = new List<Task>(); 

     foreach (var number in numbers) 
     { 
      Console.WriteLine("Before start task " + number); 

      var numberSafe = number; 

      var nextTask = Task.Factory.StartNew(() => 
      { 
       Console.WriteLine("Got number: {0}", numberSafe); 
      }) 
      .ContinueWith(task => 
      { 
       Console.WriteLine("Continuation {0}", task.Id); 
      }, TaskContinuationOptions.ExecuteSynchronously); 

      tasks.Add(nextTask); 
     } 

     return Task.Factory.ContinueWhenAll(tasks.ToArray(), 
      result => result, TaskContinuationOptions.ExecuteSynchronously); 
    } 
} 

В идеале, в проекте реальной жизни вы должны придерживаться естественно асинхронного API, везде, где это возможно (например, "Using SqlDataReader’s new async methods in .Net 4.5"), и использовать Task.Run/Task.Factory.StartNew только для CPU переплета вычислительных задач. А для серверных приложений (например, ASP.NET Web API) Task.Run/Task.Factory.StartNew обычно будет добавлять дополнительные служебные данные для избыточного переключения потоков. Это не ускорит завершение HTTP-запроса, если вам действительно не нужно выполнять несколько заданий с привязкой к процессору параллельно, что ухудшает масштабируемость.

Я понимаю, что следующий вариант может оказаться нецелесообразным, но я настоятельно рекомендую перейти на VS2012 + и использовать async/await для реализации такой логики.Это было бы очень полезно для инвестиций, поскольку это значительно ускоряет процесс кодирования и создает более простой, более чистый и менее подверженный ошибкам код. Вы по-прежнему сможете настроить .NET 4.0 на Microsoft.Bcl.Async.