時間制限のある並列データ読み込み

public sealed class AsyncDataLoader<T> { /// <summary> ///   . /// </summary> public AsyncDataLoader() { EmergencyPeriod = TimeSpan.Zero; MinResultsCount = 0; } /// <summary> ///   . /// </summary> /// <param name="dataLoaders">,  .</param> /// <param name="loadDataPeriod">,       .</param> public AsyncDataLoader(IEnumerable<Func<T>> dataLoaders, TimeSpan loadDataPeriod) : this(dataLoaders, loadDataPeriod, 0, TimeSpan.Zero) { } /// <summary> ///   . /// </summary> /// <param name="dataLoaders">,  .</param> /// <param name="loadDataPeriod">,       .</param> /// <param name="minimalResultsCount">    .</param> /// <param name="emergencyPeriod">,       .</param> public AsyncDataLoader(IEnumerable<Func<T>> dataLoaders, TimeSpan loadDataPeriod, int minimalResultsCount, TimeSpan emergencyPeriod) { DataLoaders = dataLoaders; LoadDataPeriod = loadDataPeriod; EmergencyPeriod = emergencyPeriod; MinResultsCount = minimalResultsCount; } /// <summary> ///    ,        ,     . /// </summary> public TimeSpan EmergencyPeriod { get; set; } /// <summary> ///       . /// </summary> public int MinResultsCount { get; set; } /// <summary> ///    ,  . /// </summary> public IEnumerable<Func<T>> DataLoaders { get; set; } /// <summary> ///    ,       . /// </summary> public TimeSpan LoadDataPeriod { get; set; } /// <summary> ///       . /// </summary> public bool SkipDefaultResults { get; set; } /// <summary> ///      . /// </summary> /// <returns> .</returns> public async Task<T[]> GetResultsAsync() { BlockingCollection<T> results = new BlockingCollection<T>(); List<Task> tasks = new List<Task>(); tasks.AddRange(DataLoaders.Select(handler => Task.Factory.StartNew(() => { T result = handler.Invoke(); results.Add(result); }, CancellationToken.None, TaskCreationOptions.LongRunning | TaskCreationOptions.PreferFairness, TaskScheduler.Default))); bool isAllCompleted = true; try { CancellationTokenSource source = new CancellationTokenSource(LoadDataPeriod); CancellationToken token = source.Token; #if DEBUG token = CancellationToken.None; //    #endif await Task.Factory.ContinueWhenAll(tasks.ToArray(), (t) => { }, token); } catch (TaskCanceledException ex) //  ?  . { isAllCompleted = false; } if (!isAllCompleted && EmergencyPeriod > TimeSpan.Zero) //     { Func<bool> isReadyHandler = () => results.Count >= MinResultsCount; //,  ,     . await WaitWhenReady(isReadyHandler, EmergencyPeriod); } if (SkipDefaultResults) return results.Where(r => !object.Equals(r, default(T))).ToArray(); return results.ToArray(); } /// <summary> ///    . /// </summary> /// <param name="isReadyValidator">, ,   .</param> /// <param name="totalDelay"> .</param> /// <param name="iterationsCount">   .</param> private async Task WaitWhenReady(Func<bool> isReadyValidator, TimeSpan totalDelay, int iterationsCount = 7) { if (isReadyValidator()) return; double milliseconds = totalDelay.TotalMilliseconds / iterationsCount; TimeSpan delay = TimeSpan.FromMilliseconds(milliseconds); for (int i = 0; i < iterationsCount; i++) { if (isReadyValidator()) return; await Task.Delay(delay); } } }