Есть ли способ дросселировать потоки, используемые параллельной библиотекой задач?

Question

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

Я стараюсь не introduce a wrapper над этим, поскольку я чувствую, что это может ввести проблемы.

Я понимаю, что вы можете установить близость процессора в TPL, но было бы очень приятно установить максимальный поток (возможно, для домена приложения). Поэтому при установке максимального значения потока в 1 TPL будет принудительно использовать любой поток, на котором он был использован.

Как вы думаете? Возможно ли это (я уверен, что это не так), и должно это возможно?

Edit: вот пример

public class Foo 
{ 
    public Foo() 
    { 
     Task.Factory.StartNew(() => somethingLong()) 
      .ContinueWith(a => Bar = 1) ; 
    } 
} 

[Test] public void Foo_should_set_Bar_to_1() 
{ 
    Assert.Equal(1, new Foo().Bar) ; 
} 

Тест вероятно не пройдет, пока я не ввести задержку. Я хотел бы иметь что-то вроде Task.MaximumThreads=1, так что TPL будет работать серийно.

Answer 1

Вы можете создать свой собственный класс TaskScheduler исходя из TaskScheduler, передать его в TaskFactory. Теперь у вас могут быть любые объекты Task, которые вы создаете, запустили , что планировщик.

Не нужно устанавливать его для использования в одном потоке.

Затем, прямо перед вашими утверждениями, просто позвоните по телефону Dispose(). Внутренне он будет делать что-то вроде этого, если вы будете следовать образцам там для написания TaskScheduler: -

public void Dispose() 
{ 
    if (tasks != null) 
    { 
     tasks.CompleteAdding(); 

     foreach (var thread in threads) thread.Join(); 

     tasks.Dispose(); 
     tasks = null; 
    } 
} 

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

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

Answer 2

Действительно, это больше проблема с тестируемостью лямбда-кода, чем с TPL. Предложение Hightechrider является хорошим, но, по сути, ваши тесты все еще тестируют TPL так же, как и ваш код. Вам не нужно проверять это, когда заканчивается первая задача, а ContinueWith запускает следующую задачу.

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

Сказав, что я хотел бы посмотреть, будет ли работать планировщик. Вот реализация с использованием модифицированного StaTaskScheduler от http://code.msdn.microsoft.com/ParExtSamples

using System; 
    using System.Collections.Concurrent; 
    using System.Collections.Generic; 
    using System.Linq; 
    using System.Threading; 
    using System.Threading.Tasks; 
    using Xunit; 

    namespace Example 
    { 
     public class Foo 
     { 
     private TaskScheduler _scheduler; 

    public int Bar { get; set; } 

    private void SomethingLong() 
    { 
     Thread.SpinWait(10000); 
    } 

    public Foo() 
     : this(TaskScheduler.Default) 
    { 
    } 

    public Foo(TaskScheduler scheduler) 
    { 
     _scheduler = scheduler; 
    } 

    public void DoWork() 
    { 
     var factory = new TaskFactory(_scheduler); 

     factory.StartNew(() => SomethingLong()) 
     .ContinueWith(a => Bar = 1, _scheduler); 
    } 
    } 

    public class FooTests 
    { 
    [Fact] 
    public void Foo_should_set_Bar_to_1() 
    { 
     var sch = new StaTaskScheduler(3); 
     var target = new Foo(sch); 
     target.DoWork(); 

     sch.Dispose(); 
     Assert.Equal(1, target.Bar); 
    } 
    } 

    public sealed class StaTaskScheduler : TaskScheduler, IDisposable 
    { 
    /// <summary>Stores the queued tasks to be executed by our pool of STA threads.</summary> 
    private BlockingCollection<Task> _tasks; 
    /// <summary>The STA threads used by the scheduler.</summary> 
    private readonly List<Thread> _threads; 

    /// <summary>Initializes a new instance of the StaTaskScheduler class with the specified concurrency level.</summary> 
    /// <param name="numberOfThreads">The number of threads that should be created and used by this scheduler.</param> 
    public StaTaskScheduler(int numberOfThreads) 
    { 
     // Validate arguments 
     if (numberOfThreads < 1) throw new ArgumentOutOfRangeException("concurrencyLevel"); 

     // Initialize the tasks collection 
     _tasks = new BlockingCollection<Task>(); 

     // Create the threads to be used by this scheduler 
     _threads = Enumerable.Range(0, numberOfThreads).Select(i => 
     { 
     var thread = new Thread(() => 
     { 
      // Continually get the next task and try to execute it. 
      // This will continue until the scheduler is disposed and no more tasks remain. 
      foreach (var t in _tasks.GetConsumingEnumerable()) 
      { 
      TryExecuteTask(t); 
      } 
     }); 
     thread.IsBackground = true; 
     // NO STA REQUIREMENT! 
     // thread.SetApartmentState(ApartmentState.STA); 
     return thread; 
     }).ToList(); 

     // Start all of the threads 
     _threads.ForEach(t => t.Start()); 
    } 

    /// <summary>Queues a Task to be executed by this scheduler.</summary> 
    /// <param name="task">The task to be executed.</param> 
    protected override void QueueTask(Task task) 
    { 
     // Push it into the blocking collection of tasks 
     _tasks.Add(task); 
    } 

    /// <summary>Provides a list of the scheduled tasks for the debugger to consume.</summary> 
    /// <returns>An enumerable of all tasks currently scheduled.</returns> 
    protected override IEnumerable<Task> GetScheduledTasks() 
    { 
     // Serialize the contents of the blocking collection of tasks for the debugger 
     return _tasks.ToArray(); 
    } 

    /// <summary>Determines whether a Task may be inlined.</summary> 
    /// <param name="task">The task to be executed.</param> 
    /// <param name="taskWasPreviouslyQueued">Whether the task was previously queued.</param> 
    /// <returns>true if the task was successfully inlined; otherwise, false.</returns> 
    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) 
    { 
     // Try to inline if the current thread is STA 
     return 
     Thread.CurrentThread.GetApartmentState() == ApartmentState.STA && 
     TryExecuteTask(task); 
    } 

    /// <summary>Gets the maximum concurrency level supported by this scheduler.</summary> 
    public override int MaximumConcurrencyLevel 
    { 
     get { return _threads.Count; } 
    } 

    /// <summary> 
    /// Cleans up the scheduler by indicating that no more tasks will be queued. 
    /// This method blocks until all threads successfully shutdown. 
    /// </summary> 
    public void Dispose() 
    { 
     if (_tasks != null) 
     { 
     // Indicate that no new tasks will be coming in 
     _tasks.CompleteAdding(); 

     // Wait for all threads to finish processing tasks 
     foreach (var thread in _threads) thread.Join(); 

     // Cleanup 
     _tasks.Dispose(); 
     _tasks = null; 
     } 
    } 
    } 
} 

Answer 3

Если вы желаете, чтобы избавиться от необходимости перегружать конструктор вы можете обернуть блок тестового кода в Task.Factory.ContinueWhenAll (...).

public class Foo 
{ 
    public Foo() 
    { 
     Task.Factory.StartNew(() => somethingLong()) 
      .ContinueWith(a => Bar = 1) ; 
    } 
} 

[Test] public void Foo_should_set_Bar_to_1() 
{ 
    Foo foo; 
    Task.Factory.ContinueWhenAll(
     new [] { 
      new Task(() => { 
       foo = new Foo(); 
      }) 
     }, 
     asserts => { 
      Assert.Equal(1, foo.Bar) ; 
     } 
    ).Wait; 
} 

Было бы интересно услышать некоторые ответы на этот подход.