Task 有静态方法WaitAll和WaitAny，主要用于等待其他Task完成后做一些事情，先看看其实现部分吧：
public class Task : IThreadPoolWorkItem, IAsyncResult, IDisposable
{
//Waits for all of the provided Task objects to complete execution.
public static void WaitAll(params Task[] tasks)
{
WaitAll(tasks, Timeout.Infinite);
} //Waits for any of the provided Task objects to complete execution.Return The index of the completed task in the tasks array argument.
public static int WaitAny(params Task[] tasks)
{
int waitResult = WaitAny(tasks, Timeout.Infinite);
Contract.Assert(tasks.Length == || waitResult != -, "expected wait to succeed");
return waitResult;
} //true if all of the Task instances completed execution within the allotted time; otherwise, false.
public static bool WaitAll(Task[] tasks, int millisecondsTimeout, CancellationToken cancellationToken)
{
if (tasks == null)
{
throw new ArgumentNullException("tasks");
}
if (millisecondsTimeout < -)
{
throw new ArgumentOutOfRangeException("millisecondsTimeout");
}
Contract.EndContractBlock();
cancellationToken.ThrowIfCancellationRequested(); // early check before we make any allocations List<Exception> exceptions = null;
List<Task> waitedOnTaskList = null;
List<Task> notificationTasks = null; // If any of the waited-upon tasks end as Faulted or Canceled, set these to true.
bool exceptionSeen = false, cancellationSeen = false;
bool returnValue = true; // Collects incomplete tasks in "waitedOnTaskList"
for (int i = tasks.Length - ; i >= ; i--)
{
Task task = tasks[i];
if (task == null)
{
throw new ArgumentException(Environment.GetResourceString("Task_WaitMulti_NullTask"), "tasks");
} bool taskIsCompleted = task.IsCompleted;
if (!taskIsCompleted)
{
// try inlining the task only if we have an infinite timeout and an empty cancellation token
if (millisecondsTimeout != Timeout.Infinite || cancellationToken.CanBeCanceled)
{
AddToList(task, ref waitedOnTaskList, initSize: tasks.Length);
}
else
{
// We are eligible for inlining. If it doesn't work, we'll do a full wait.
taskIsCompleted = task.WrappedTryRunInline() && task.IsCompleted; // A successful TryRunInline doesn't guarantee completion
if (!taskIsCompleted) AddToList(task, ref waitedOnTaskList, initSize: tasks.Length);
}
}
if (taskIsCompleted)
{
if (task.IsFaulted) exceptionSeen = true;
else if (task.IsCanceled) cancellationSeen = true;
if (task.IsWaitNotificationEnabled) AddToList(task, ref notificationTasks, initSize: 1);
}
} if (waitedOnTaskList != null)
{
// Block waiting for the tasks to complete.
returnValue = WaitAllBlockingCore(waitedOnTaskList, millisecondsTimeout, cancellationToken); // If the wait didn't time out, ensure exceptions are propagated, and if a debugger is
// attached and one of these tasks requires it, that we notify the debugger of a wait completion.
if (returnValue)
{
foreach (var task in waitedOnTaskList)
{
if (task.IsFaulted) exceptionSeen = true;
else if (task.IsCanceled) cancellationSeen = true;
if (task.IsWaitNotificationEnabled) AddToList(task, ref notificationTasks, initSize: );
}
} GC.KeepAlive(tasks);
} if (returnValue && notificationTasks != null)
{
foreach (var task in notificationTasks)
{
if (task.NotifyDebuggerOfWaitCompletionIfNecessary()) break;
}
} // If one or more threw exceptions, aggregate and throw them.
if (returnValue && (exceptionSeen || cancellationSeen))
{
if (!exceptionSeen) cancellationToken.ThrowIfCancellationRequested(); // Now gather up and throw all of the exceptions.
foreach (var task in tasks) AddExceptionsForCompletedTask(ref exceptions, task);
Contract.Assert(exceptions != null, "Should have seen at least one exception");
throw new AggregateException(exceptions);
} return returnValue;
} public static int WaitAny(Task[] tasks, int millisecondsTimeout, CancellationToken cancellationToken)
{
if (tasks == null)
{
throw new ArgumentNullException("tasks");
}
if (millisecondsTimeout < -)
{
throw new ArgumentOutOfRangeException("millisecondsTimeout");
}
Contract.EndContractBlock();
cancellationToken.ThrowIfCancellationRequested(); // early check before we make any allocations int signaledTaskIndex = -;
for (int taskIndex = ; taskIndex < tasks.Length; taskIndex++)
{
Task task = tasks[taskIndex];
if (task == null)
{
throw new ArgumentException(Environment.GetResourceString("Task_WaitMulti_NullTask"), "tasks");
}
if (signaledTaskIndex == - && task.IsCompleted)
{
signaledTaskIndex = taskIndex;
}
} if (signaledTaskIndex == - && tasks.Length != )
{
Task<Task> firstCompleted = TaskFactory.CommonCWAnyLogic(tasks);
bool waitCompleted = firstCompleted.Wait(millisecondsTimeout, cancellationToken);
if (waitCompleted)
{
Contract.Assert(firstCompleted.Status == TaskStatus.RanToCompletion);
signaledTaskIndex = Array.IndexOf(tasks, firstCompleted.Result);
Contract.Assert(signaledTaskIndex >= );
}
}
GC.KeepAlive(tasks);
return signaledTaskIndex;
} //Performs a blocking WaitAll on the vetted list of tasks.true if all of the tasks completed; otherwise, false.
private static bool WaitAllBlockingCore(List<Task> tasks, int millisecondsTimeout, CancellationToken cancellationToken)
{
Contract.Assert(tasks != null, "Expected a non-null list of tasks");
Contract.Assert(tasks.Count > , "Expected at least one task");
bool waitCompleted = false;
var mres = new SetOnCountdownMres(tasks.Count);
try
{
foreach (var task in tasks)
{
task.AddCompletionAction(mres, addBeforeOthers: true);
}
waitCompleted = mres.Wait(millisecondsTimeout, cancellationToken);
}
finally
{
if (!waitCompleted)
{
foreach (var task in tasks)
{
if (!task.IsCompleted) task.RemoveContinuation(mres);
}
}
}
return waitCompleted;
} private sealed class SetOnCountdownMres : ManualResetEventSlim, ITaskCompletionAction
{
private int _count; internal SetOnCountdownMres(int count)
{
Contract.Assert(count > , "Expected count > 0");
_count = count;
} public void Invoke(Task completingTask)
{
if (Interlocked.Decrement(ref _count) == 0) Set();
Contract.Assert(_count >= , "Count should never go below 0");
}
}
}
我们首先看看WaitAll的方法，检查Task数组中每个Task实例，检查Task是否已经完成，如果没有完成就把Task添加到waitedOnTaskList集合中，如果waitedOnTaskList集合有元素那么，我们就调用WaitAllBlockingCore来实现真正的等待，当代完毕后我们需要检查notificationTasks集合是否有元素，如果有则依次调用Task的NotifyDebuggerOfWaitCompletionIfNecessary方法。WaitAllBlockingCore实现阻塞是依靠SetOnCountdownMres实例的【和CountdownEvent思路一样，每次调用Invoke的时候，就把计数器_count减1，当_count==0时就调用Set方法】，在WaitAllBlockingCore方法退出前，需要检查Task是否都完成，如果有没有完成的需要移除相应task的SetOnCountdownMres实例【if (!task.IsCompleted) task.RemoveContinuation(mres);】，SetOnCountdownMres的Invoke方法是在Task的FinishContinuations方法调用的【 ITaskCompletionAction singleTaskCompletionAction = continuationObject as ITaskCompletionAction; singleTaskCompletionAction.Invoke(this);注意FinishContinuations方法是在FinishStageThree中调用】注意里面的GC.KeepAlive(tasks)。
现在我们来看看WaitAny方法的实现，首先我们需要循环Task[]，检查里面是否有Task已经完成，如果有则直接返回，否者我们调用Task<Task> firstCompleted = TaskFactory.CommonCWAnyLogic(tasks);返回一个Task，然后调用该Task的Wait方法【bool waitCompleted = firstCompleted.Wait(millisecondsTimeout, cancellationToken);】，让我们来看看CommonCWAnyLogic的实现：
public class TaskFactory
{
internal static Task<Task> CommonCWAnyLogic(IList<Task> tasks)
{
Contract.Requires(tasks != null);
var promise = new CompleteOnInvokePromise(tasks);
bool checkArgsOnly = false;
int numTasks = tasks.Count;
for(int i=; i<numTasks; i++)
{
var task = tasks[i];
if (task == null) throw new ArgumentException(Environment.GetResourceString("Task_MultiTaskContinuation_NullTask"), "tasks"); if (checkArgsOnly) continue; // If the promise has already completed, don't bother with checking any more tasks.
if (promise.IsCompleted)
{
checkArgsOnly = true;
}
// If a task has already completed, complete the promise.
else if (task.IsCompleted)
{
promise.Invoke(task);
checkArgsOnly = true;
}
// Otherwise, add the completion action and keep going.
else task.AddCompletionAction(promise);
}
return promise;
}
internal sealed class CompleteOnInvokePromise : Task<Task>, ITaskCompletionAction
{
private IList<Task> _tasks; // must track this for cleanup
private int m_firstTaskAlreadyCompleted; public CompleteOnInvokePromise(IList<Task> tasks) : base()
{
Contract.Requires(tasks != null, "Expected non-null collection of tasks");
_tasks = tasks; if (AsyncCausalityTracer.LoggingOn)
AsyncCausalityTracer.TraceOperationCreation(CausalityTraceLevel.Required, this.Id, "TaskFactory.ContinueWhenAny", ); if (Task.s_asyncDebuggingEnabled)
{
AddToActiveTasks(this);
}
} public void Invoke(Task completingTask)
{
if (Interlocked.CompareExchange(ref m_firstTaskAlreadyCompleted, , ) == )
{
if (AsyncCausalityTracer.LoggingOn)
{
AsyncCausalityTracer.TraceOperationRelation(CausalityTraceLevel.Important, this.Id, CausalityRelation.Choice);
AsyncCausalityTracer.TraceOperationCompletion(CausalityTraceLevel.Required, this.Id, AsyncCausalityStatus.Completed);
} if (Task.s_asyncDebuggingEnabled)
{
RemoveFromActiveTasks(this.Id);
} bool success = TrySetResult(completingTask);
Contract.Assert(success, "Only one task should have gotten to this point, and thus this must be successful."); var tasks = _tasks;
int numTasks = tasks.Count;
for (int i = ; i < numTasks; i++)
{
var task = tasks[i];
if (task != null && // if an element was erroneously nulled out concurrently, just skip it; worst case is we don't remove a continuation
!task.IsCompleted) task.RemoveContinuation(this);
}
_tasks = null; }
}
}
}
CommonCWAnyLogic首先实例化CompleteOnInvokePromise【var promise = new CompleteOnInvokePromise(tasks)】，检查promise 是否完成，检查每个Task是否完成，否者就把promise作为Task的Continue Task【这里可以理解为每个Task都有一个相同Continue Task】，而CompleteOnInvokePromise自己的wait是在WaitAny中的firstCompleted.Wait(millisecondsTimeout, cancellationToken)方法，当其中其中一个Task完成后，在Task的FinishContinuations方法调用的CompleteOnInvokePromise的Invoke【一旦触发后就需要移调其他task上的CompleteOnInvokePromise，如这里的task.RemoveContinuation(this)】。在CompleteOnInvokePromise的Invoke方法我们调用TrySetResult(completingTask)方法，期实现如下：
public class Task<TResult> : Task
{
internal bool TrySetResult(TResult result)
{
if (IsCompleted) return false;
Contract.Assert(m_action == null, "Task<T>.TrySetResult(): non-null m_action");
if (AtomicStateUpdate(TASK_STATE_COMPLETION_RESERVED,
TASK_STATE_COMPLETION_RESERVED | TASK_STATE_RAN_TO_COMPLETION | TASK_STATE_FAULTED | TASK_STATE_CANCELED))
{
m_result = result;
Interlocked.Exchange(ref m_stateFlags, m_stateFlags | TASK_STATE_RAN_TO_COMPLETION); var cp = m_contingentProperties;
if (cp != null) cp.SetCompleted();
FinishStageThree();
return true;
}
return false;
}
}
这里的TrySetResult方法里面调用FinishStageThree方法，以保证Task后面的Continue Task的执行。