Add UniTask.WhenEach

v2.5.11
neuecc 2024-09-30 20:42:17 +09:00
parent 87e164e275
commit 0826b7e976
8 changed files with 310 additions and 21 deletions

View File

@ -160,7 +160,7 @@ UniTask provides three pattern of extension methods.
> Note: AssetBundleRequest has `asset` and `allAssets`, default await returns `asset`. If you want to get `allAssets`, you can use `AwaitForAllAssets()` method.
The type of `UniTask` can use utilities like `UniTask.WhenAll`, `UniTask.WhenAny`. They are like `Task.WhenAll`/`Task.WhenAny` but the return type is more useful. They return value tuples so you can deconstruct each result and pass multiple types.
The type of `UniTask` can use utilities like `UniTask.WhenAll`, `UniTask.WhenAny`, `UniTask.WhenEach`. They are like `Task.WhenAll`/`Task.WhenAny` but the return type is more useful. They return value tuples so you can deconstruct each result and pass multiple types.
```csharp
public async UniTaskVoid LoadManyAsync()
@ -716,6 +716,19 @@ await UniTaskAsyncEnumerable.EveryUpdate().ForEachAsync(_ =>
}, token);
```
`UniTask.WhenEach` that is similar to .NET 9's `Task.WhenEach` can consume new way for await multiple tasks.
```csharp
await foreach (var result in UniTask.WhenEach(task1, task2, task3))
{
// The result is of type WhenEachResult<T>.
// It contains either `T Result` or `Exception Exception`.
// You can check `IsCompletedSuccessfully` or `IsFaulted` to determine whether to access `.Result` or `.Exception`.
// If you want to throw an exception when `IsFaulted` and retrieve the result when successful, use `GetResult()`.
Debug.Log(result.GetResult());
}
```
UniTaskAsyncEnumerable implements asynchronous LINQ, similar to LINQ in `IEnumerable<T>` or Rx in `IObservable<T>`. All standard LINQ query operators can be applied to asynchronous streams. For example, the following code shows how to apply a Where filter to a button-click asynchronous stream that runs once every two clicks.
```csharp
@ -1026,6 +1039,7 @@ Use UniTask type.
| `Task.Run` | `UniTask.RunOnThreadPool` |
| `Task.WhenAll` | `UniTask.WhenAll` |
| `Task.WhenAny` | `UniTask.WhenAny` |
| `Task.WhenEach` | `UniTask.WhenEach` |
| `Task.CompletedTask` | `UniTask.CompletedTask` |
| `Task.FromException` | `UniTask.FromException` |
| `Task.FromResult` | `UniTask.FromResult` |

View File

@ -0,0 +1,69 @@
using Cysharp.Threading.Tasks;
using FluentAssertions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Xunit;
namespace NetCoreTests
{
public class WhenEachTest
{
[Fact]
public async Task Each()
{
var a = Delay(1, 3000);
var b = Delay(2, 1000);
var c = Delay(3, 2000);
var l = new List<int>();
await foreach (var item in UniTask.WhenEach(a, b, c))
{
l.Add(item.Result);
}
l.Should().Equal(2, 3, 1);
}
[Fact]
public async Task Error()
{
var a = Delay2(1, 3000);
var b = Delay2(2, 1000);
var c = Delay2(3, 2000);
var l = new List<WhenEachResult<int>>();
await foreach (var item in UniTask.WhenEach(a, b, c))
{
l.Add(item);
}
l[0].IsCompletedSuccessfully.Should().BeTrue();
l[0].IsFaulted.Should().BeFalse();
l[0].Result.Should().Be(2);
l[1].IsCompletedSuccessfully.Should().BeFalse();
l[1].IsFaulted.Should().BeTrue();
l[1].Exception.Message.Should().Be("ERROR");
l[2].IsCompletedSuccessfully.Should().BeTrue();
l[2].IsFaulted.Should().BeFalse();
l[2].Result.Should().Be(1);
}
async UniTask<int> Delay(int id, int sleep)
{
await Task.Delay(sleep);
return id;
}
async UniTask<int> Delay2(int id, int sleep)
{
await Task.Delay(sleep);
if (id == 3) throw new Exception("ERROR");
return id;
}
}
}

View File

@ -0,0 +1,183 @@
using Cysharp.Threading.Tasks.Internal;
using System;
using System.Collections.Generic;
using System.Runtime.ExceptionServices;
using System.Threading;
namespace Cysharp.Threading.Tasks
{
public partial struct UniTask
{
public static IUniTaskAsyncEnumerable<WhenEachResult<T>> WhenEach<T>(IEnumerable<UniTask<T>> tasks)
{
return new WhenEachEnumerable<T>(tasks);
}
public static IUniTaskAsyncEnumerable<WhenEachResult<T>> WhenEach<T>(params UniTask<T>[] tasks)
{
return new WhenEachEnumerable<T>(tasks);
}
}
public readonly struct WhenEachResult<T>
{
public T Result { get; }
public Exception Exception { get; }
//[MemberNotNullWhen(false, nameof(Exception))]
public bool IsCompletedSuccessfully => Exception == null;
//[MemberNotNullWhen(true, nameof(Exception))]
public bool IsFaulted => Exception != null;
public WhenEachResult(T result)
{
this.Result = result;
this.Exception = null;
}
public WhenEachResult(Exception exception)
{
if (exception == null) throw new ArgumentNullException(nameof(exception));
this.Result = default!;
this.Exception = exception;
}
public void TryThrow()
{
if (IsFaulted)
{
ExceptionDispatchInfo.Capture(Exception).Throw();
}
}
public T GetResult()
{
if (IsFaulted)
{
ExceptionDispatchInfo.Capture(Exception).Throw();
}
return Result;
}
public override string ToString()
{
if (IsCompletedSuccessfully)
{
return Result?.ToString() ?? "";
}
else
{
return $"Exception{{{Exception.Message}}}";
}
}
}
internal enum WhenEachState : byte
{
NotRunning,
Running,
Completed
}
internal sealed class WhenEachEnumerable<T> : IUniTaskAsyncEnumerable<WhenEachResult<T>>
{
IEnumerable<UniTask<T>> source;
public WhenEachEnumerable(IEnumerable<UniTask<T>> source)
{
this.source = source;
}
public IUniTaskAsyncEnumerator<WhenEachResult<T>> GetAsyncEnumerator(CancellationToken cancellationToken = default)
{
return new Enumerator(source, cancellationToken);
}
sealed class Enumerator : IUniTaskAsyncEnumerator<WhenEachResult<T>>
{
readonly IEnumerable<UniTask<T>> source;
CancellationToken cancellationToken;
Channel<WhenEachResult<T>> channel;
IUniTaskAsyncEnumerator<WhenEachResult<T>> channelEnumerator;
int completeCount;
WhenEachState state;
public Enumerator(IEnumerable<UniTask<T>> source, CancellationToken cancellationToken)
{
this.source = source;
this.cancellationToken = cancellationToken;
}
public WhenEachResult<T> Current => channelEnumerator.Current;
public UniTask<bool> MoveNextAsync()
{
cancellationToken.ThrowIfCancellationRequested();
if (state == WhenEachState.NotRunning)
{
state = WhenEachState.Running;
channel = Channel.CreateSingleConsumerUnbounded<WhenEachResult<T>>();
channelEnumerator = channel.Reader.ReadAllAsync().GetAsyncEnumerator(cancellationToken);
if (source is UniTask<T>[] array)
{
ConsumeAll(this, array, array.Length);
}
else
{
using (var rentArray = ArrayPoolUtil.Materialize(source))
{
ConsumeAll(this, rentArray.Array, rentArray.Length);
}
}
}
return channelEnumerator.MoveNextAsync();
}
static void ConsumeAll(Enumerator self, UniTask<T>[] array, int length)
{
for (int i = 0; i < length; i++)
{
RunWhenEachTask(self, array[i], length).Forget();
}
static async UniTaskVoid RunWhenEachTask(Enumerator self, UniTask<T> task, int length)
{
try
{
var result = await task;
self.channel.Writer.TryWrite(new WhenEachResult<T>(result));
}
catch (Exception ex)
{
self.channel.Writer.TryWrite(new WhenEachResult<T>(ex));
}
if (Interlocked.Increment(ref self.completeCount) == length)
{
self.state = WhenEachState.Completed;
self.channel.Writer.TryComplete();
}
}
}
public async UniTask DisposeAsync()
{
if (channelEnumerator != null)
{
await channelEnumerator.DisposeAsync();
}
if (state != WhenEachState.Completed)
{
state = WhenEachState.Completed;
channel.Writer.TryComplete(new OperationCanceledException());
}
}
}
}
}

View File

@ -0,0 +1,2 @@
fileFormatVersion: 2
guid: 7cac24fdda5112047a1cd3dd66b542c4

View File

@ -119,7 +119,28 @@ public class AsyncMessageBroker<T> : IDisposable
connection.Dispose();
}
}
public class WhenEachTest
{
public async UniTask Each()
{
var a = Delay(1, 3000);
var b = Delay(2, 1000);
var c = Delay(3, 2000);
var l = new List<int>();
await foreach (var item in UniTask.WhenEach(a, b, c))
{
Debug.Log(item.Result);
}
}
async UniTask<int> Delay(int id, int sleep)
{
await UniTask.Delay(sleep);
return id;
}
}
public class SandboxMain : MonoBehaviour
{
@ -147,6 +168,18 @@ public class SandboxMain : MonoBehaviour
Debug.Log("Again");
// var foo = InstantiateAsync<SandboxMain>(this).ToUniTask();
// var tako = await foo;
return 10;
}
@ -557,6 +590,7 @@ public class SandboxMain : MonoBehaviour
async UniTaskVoid Start()
{
await new WhenEachTest().Each();
// UniTask.Delay(TimeSpan.FromSeconds(1)).TimeoutWithoutException

View File

@ -13,7 +13,7 @@ OcclusionCullingSettings:
--- !u!104 &2
RenderSettings:
m_ObjectHideFlags: 0
serializedVersion: 9
serializedVersion: 10
m_Fog: 0
m_FogColor: {r: 0.5, g: 0.5, b: 0.5, a: 1}
m_FogMode: 3
@ -43,7 +43,6 @@ RenderSettings:
LightmapSettings:
m_ObjectHideFlags: 0
serializedVersion: 12
m_GIWorkflowMode: 1
m_GISettings:
serializedVersion: 2
m_BounceScale: 1
@ -66,9 +65,6 @@ LightmapSettings:
m_LightmapParameters: {fileID: 0}
m_LightmapsBakeMode: 1
m_TextureCompression: 1
m_FinalGather: 0
m_FinalGatherFiltering: 1
m_FinalGatherRayCount: 256
m_ReflectionCompression: 2
m_MixedBakeMode: 2
m_BakeBackend: 0
@ -719,7 +715,6 @@ GameObject:
- component: {fileID: 1556045507}
- component: {fileID: 1556045506}
- component: {fileID: 1556045505}
- component: {fileID: 1556045509}
m_Layer: 0
m_Name: Canvas
m_TagString: Untagged
@ -812,18 +807,6 @@ RectTransform:
m_AnchoredPosition: {x: 0, y: 0}
m_SizeDelta: {x: 0, y: 0}
m_Pivot: {x: 0, y: 0}
--- !u!114 &1556045509
MonoBehaviour:
m_ObjectHideFlags: 0
m_CorrespondingSourceObject: {fileID: 0}
m_PrefabInstance: {fileID: 0}
m_PrefabAsset: {fileID: 0}
m_GameObject: {fileID: 1556045504}
m_Enabled: 1
m_EditorHideFlags: 0
m_Script: {fileID: 11500000, guid: a478e5f6126dc184ca902adfb35401b4, type: 3}
m_Name:
m_EditorClassIdentifier:
--- !u!1 &1584557231
GameObject:
m_ObjectHideFlags: 0

View File

@ -1,4 +1,6 @@
using Cysharp.Threading.Tasks;
#pragma warning disable CS0618
using Cysharp.Threading.Tasks;
using Cysharp.Threading.Tasks.Linq;
using FluentAssertions;
using NUnit.Framework;

View File

@ -1,4 +1,6 @@
#if !(UNITY_4_5 || UNITY_4_6 || UNITY_4_7 || UNITY_5_0 || UNITY_5_1 || UNITY_5_2)
#pragma warning disable CS0618
#if !(UNITY_4_5 || UNITY_4_6 || UNITY_4_7 || UNITY_5_0 || UNITY_5_1 || UNITY_5_2)
#pragma warning disable CS1591 // Missing XML comment for publicly visible type or member
using UnityEngine;