using System.Diagnostics;
namespace MeloongCore;
public static class TaskUtils {
///
/// 并行执行所有任务。
///
public static Task WhenAll(IEnumerable> tasks, ProgressProvider? progress = null) {
var taskList = tasks.ToList();
if (!taskList.Any()) {
progress?.Skip();
return Task.CompletedTask;
}
// 实际分割
var progressEach = 1d / taskList.Count;
return Task.WhenAll(taskList.Select(async (task, _) => {
var cp = progress?.SplitBy(progressEach)[0];
await task(cp).NoCapture();
cp?.Finish();
}));
}
///
/// 并行执行所有任务并返回结果。
///
public static Task WhenAll(IEnumerable>> tasks, ProgressProvider? progress = null) {
var taskList = tasks.ToList();
if (!taskList.Any()) {
progress?.Skip();
return Task.FromResult(Array.Empty());
}
// 实际分割
var progressEach = 1d / taskList.Count;
return Task.WhenAll(taskList.Select(async (task, _) => {
var cp = progress?.SplitBy(progressEach)[0];
var result = await task(cp).NoCapture();
cp?.Finish();
return result;
}));
}
///
/// 并行执行所有任务。
///
public static Task WhenAll(IEnumerable tasks, ProgressProvider? progress = null)
=> WhenAll(tasks.Select(task => new Func(_ => task)), progress);
///
/// 并行执行所有任务并返回结果。
///
public static Task WhenAll(IEnumerable> tasks, ProgressProvider? progress = null)
=> WhenAll(tasks.Select(task => new Func>(_ => task)), progress);
public static void ForEach(IEnumerable source, Action body)
=> Parallel.ForEach(source, body);
public static async Task ForEachAsync(IEnumerable source, int maxDegreeOfParallelism, Func body,
CancellationToken cancellationToken = default, ProgressProvider? progress = null) {
using var linkedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
var token = linkedCancellationTokenSource.Token;
var enumeratorLock = new object();
using var enumerator = source.GetEnumerator();
async Task WorkerAsync(ProgressProvider? cp) {
try {
while (true) {
T item;
lock (enumeratorLock) {
token.ThrowIfCancellationRequested();
if (!enumerator.MoveNext()) return;
item = enumerator.Current;
}
await body(item, cp).NoCapture();
}
} catch {
try {
linkedCancellationTokenSource.Cancel(); // 让尚未开始的 worker 尽快停止
} catch {
}
throw;
}
}
await TaskUtils.WhenAll(Enumerable.Repeat(WorkerAsync, maxDegreeOfParallelism), progress).NoCapture();
}
///
/// 静默运行程序并等待其结束,返回其输出和退出码。
/// 支持 notepad、git 等命令。
/// 超时或启动失败会抛出异常。
///
public static async Task<(string Output, int ExitCode)> RunProgramAsync(string file, string arguments = "", int? timeoutMs = null, Encoding? encoding = null, CancellationToken cancellationToken = default) {
cancellationToken.ThrowIfCancellationRequested();
var info = new ProcessStartInfo {
Arguments = arguments,
FileName = PathUtils.ToShortPath(file),
UseShellExecute = false,
CreateNoWindow = true,
RedirectStandardOutput = true, StandardOutputEncoding = encoding,
RedirectStandardError = true, StandardErrorEncoding = encoding
};
using var program = new Process();
program.StartInfo = info;
program.EnableRaisingEvents = true;
if (!program.Start()) throw new InvalidOperationException($"运行程序时出现意外错误:{file} {arguments}");
bool hasTimeout = timeoutMs is > 0;
Logger.Info($"运行程序,并返回其输出:{file} {arguments}{(hasTimeout ? $",最长可等待 {timeoutMs}ms" : "")}");
// 输出和错误流
var outputTask = program.StandardOutput.ReadToEndAsync();
var errorTask = program.StandardError.ReadToEndAsync();
var completionSource = new TaskCompletionSource