UniTask 스트림은 당길 때만 값을 준다

var ct = this.GetCancellationTokenOnDestroy();
await foreach (var _ in button.OnClickAsAsyncEnumerable().WithCancellation(ct))
{
    await SaveAsync(ct);   // 3초. 그동안 누른 클릭은 받지 않는다
}

onClick.AddListener 로 짜면 저장 중 연타를 막으려고 isSaving 같은 플래그를 세워야 합니다. 이벤트는 받는 쪽 사정을 모르고 밀어 넣기 때문입니다.

UniTask 스트림은 소비자가 당길 때만 값을 줍니다. 당기지 않는 동안 온 값은 쌓이지 않고 버려지므로 플래그가 필요 없습니다.

소비자가 당길 때만 값이 간다

await foreach 는 while (await e.MoveNextAsync()) 로 바뀝니다. 본문이 도는 동안에는 MoveNextAsync() 가 불리지 않습니다. 이것이 “당길 때만” 의 실체입니다.

EveryUpdate 열거자로 봅니다. PlayerLoop 가 매 프레임 MoveNext() 로 값을 밀고, 소비자가 MoveNextAsync() 로 당깁니다. 안의 완료 원천은 한 세대에 한 번만 완료되고, 새 세대는 소비자가 당길 때만 열립니다.

// PlayerLoop 가 매 프레임 부른다
public bool MoveNext()
{
    if (disposed)
    {
        completionSource.TrySetResult(false);
        return false;                          // PlayerLoop 에서 빠진다
    }

    completionSource.TrySetResult(true);       // 이번 세대가 이미 완료됐으면 false 를 돌려주고 값은 버려진다
    return true;                               // 스트림은 끝나지 않으므로 루프에 남는다
}

// await foreach 가 값을 당길 때 부른다
public UniTask<bool> MoveNextAsync()
{
    completionSource.Reset();                  // 세대를 올리고 완료 상태를 푼다. 받을 준비가 됐다는 표시다
    return new UniTask<bool>(this, completionSource.Version);
}

소비자가 프레임 10 의 값을 받고 두 프레임짜리 작업을 하면 이렇게 됩니다.

소비자가 본문에서 break 하면 await foreach 의 finally 가 DisposeAsync 를 불러 disposed 가 켜집니다. 다음 프레임의 MoveNext() 가 그것을 보고 false 를 돌려주고, 그 순간 PlayerLoop 에서 빠집니다.

바쁜 동안의 값은 이미 완료된 세대에 들어가려다 실패하고 버려집니다. 버튼도 같습니다. 열거자는 세대 번호만 올려 재사용하므로 값마다 만드는 것이 없고, 매 프레임 할당은 0 B 였습니다.

속도가 이득은 아닙니다. 구독자 하나가 매 프레임 더하는 시간입니다.

얻는 것은 상태를 지역 변수로 들고 가고 바쁜 동안의 값을 저절로 버리는 코드 모양입니다.

이렇게 쓴다

처리 중에 온 값을 버려도 되는지로 고릅니다.

// 버려도 된다: 저장 중에 누른 클릭은 무시한다. 플래그가 필요 없다
await foreach (var _ in button.OnClickAsAsyncEnumerable().WithCancellation(ct))
{
    await SaveAsync(ct);
}

// 하나도 버리면 안 된다: Queue() 를 끼운다
await foreach (var message in messages.Queue().WithCancellation(ct))
{
    await HandleAsync(message, ct);   // 처리 중에 온 메시지는 쌓였다가 차례로 온다
}

Queue() 는 원본을 따로 끝까지 당겨 채널에 쌓아 두고, 소비자는 채널에서 꺼냅니다. 버퍼에 상한이 없어서, 생산이 계속 처리보다 빠르면 쌓이기만 합니다.

스트림은 await foreach 로 읽습니다. break 나 예외로 빠져나가도 DisposeAsync 가 불려 열거자가 PlayerLoop 에서 빠집니다. 열거자를 손으로 들고 MoveNextAsync() 를 부르다가 DisposeAsync 를 빠뜨리면, EveryUpdate 같은 열거자는 PlayerLoop 에 남아 매 프레임 불립니다.

트리거는 유니티 메시지를 받을 컴포넌트를 붙인다

Update, OnCollisionEnter 같은 메시지를 기다리려면 그 메시지를 받는 컴포넌트가 있어야 합니다. GetAsyncUpdateTrigger() 는 AsyncUpdateTrigger 를 찾고, 없으면 AddComponent 합니다. 오브젝트당 하나이고, 첫 호출에 64 B 가 들었습니다(Mono·IL2CPP Development 플레이어).

var trigger = this.GetAsyncUpdateTrigger();
await trigger.UpdateAsync();                        // 다음 Update 한 번
await foreach (var _ in trigger) { /* 매 Update */ } // 스트림

MonoBehaviour 에서 부른 GetCancellationTokenOnDestroy() 는 유니티 내장 destroyCancellationToken 을 돌려주고 컴포넌트를 붙이지 않습니다. 오브젝트가 파괴되면 await foreach 는 정상 종료하고, UpdateAsync() 는 OperationCanceledException 으로 끝납니다.

이렇게 쓴다

trigger.UpdateAsync() 는 부를 때마다 핸들러를 새로 만들어 목록에 넣고 뺍니다. 한 번 기다릴 때마다 120 B 였습니다. 루프에서 되풀이해 기다릴 때는 핸들러를 한 번 받아 두고 다시 씁니다. 이때는 0 B 였습니다.

// 매번 120 B
while (!ct.IsCancellationRequested)
{
    await trigger.UpdateAsync();
    Move();
}

// 0 B. 핸들러를 루프 밖에서 한 번 받는다
var handler = trigger.GetUpdateAsyncHandler(ct);

while (!ct.IsCancellationRequested)
{
    await handler.UpdateAsync();
    Move();
}

토큰을 넘겨 두면 취소될 때 핸들러가 트리거의 목록에서 빠집니다.

트리거는 도는 중에 목록이 바뀌어도 한 번씩 건넨다

한 트리거를 여러 곳에서 기다릴 수 있어, 메시지가 오면 구독자 목록을 돌며 OnNext 를 부릅니다. 목록은 구독자가 Prev/Next 를 자기 몸에 든 연결 리스트이고, head.Prev 가 꼬리입니다.

문제는 OnNext 안에서 목록이 바뀐다는 것입니다. UpdateAsync() 를 기다리던 코드는 그 자리에서 이어 달리며 자기 핸들러를 빼고, 다시 기다리며 새로 넣습니다. A, B, C 가 있고 B 가 UpdateAsync() 를 기다리던 경우입니다.

var h = head;

while (h != null)
{
    iteratingNode = h;                  // 지금 부르는 노드를 적어 둔다
    h.OnNext(value);                    // 기다리던 코드가 여기서 이어 달린다. 목록이 바뀔 수 있다

    // 부르는 동안 h 가 빠졌으면 Remove 가 iteratingNode 에 다음 노드를 적어 둔다
    h = h == iteratingNode ? h.Next : iteratingNode;
}

iteratingNode = null;

// 도는 동안 새로 등록된 핸들러는 대기 목록에 있다. 순회가 끝난 뒤 꼬리에 붙인다
if (iteratingHead != null)
{
    Add(iteratingHead);
    iteratingHead = null;
}

B 의 OnNext 안에서 일어나는 일은 둘입니다.

1. B 가 빠집니다. Remove 는 빠지는 노드가 iteratingNode 이면 그 자리에 다음 노드 C 를 적어 둡니다. 그래서 h 는 끊어진 B.Next 가 아니라 C 로 갑니다. 목록은 A → C 가 됩니다.
2. B 가 곧바로 다시 기다립니다. while 루프가 await trigger.UpdateAsync() 를 다시 불러 새 핸들러 B2 를 등록합니다. 도는 중이라 본 목록이 아니라 대기 목록에 들어갑니다. 순회가 끝나면 A → C → B2 가 되고, B2 는 다음 Update 부터 받습니다.

빠지는 쪽은 다음 노드를 iteratingNode 에 적어 두고, 새로 들어오는 쪽은 대기 목록에 있다가 순회가 끝나면 붙습니다. 그래서 받을 쪽은 모두 한 번씩 받고, 같은 발행을 두 번 받는 쪽은 없습니다.

대기 목록에 둘 이상 쌓이면 이 약속이 깨집니다. B 와 C 가 대기 목록에 있으면 꼬리 C 는 B 의 Prev 에 적혀 있습니다. Add(iteratingHead) 는 B 를 노드 하나로 보고 head.Prev = B 로 적습니다. 다음 Add(D) 가 꼬리로 적힌 B 뒤에 D 를 붙이며 B.Next = D 로 덮어써서 C 가 떨어집니다.

실측도 같았습니다. A 의 발행 안에 B, C 를 걸고 같은 프레임에 D 를 걸자 5 프레임 동안 B 5번, D 5번, C 0번 받았습니다. C 는 예외도 경고도 없이 영영 깨어나지 않습니다.

이렇게 쓴다

조건은 “트리거가 구독자를 부르는 도중에 같은 트리거에 새 구독이 둘 이상 생기는 것” 입니다. 그 조건을 만들지 않습니다.

• 여러 곳이 같은 트리거를 기다려야 하면, 트리거 콜백 밖에서 핸들러를 미리 받아 둡니다. 위의 GetUpdateAsyncHandler 모양입니다.
• 콜백 안에서 이어서 기다려야 하면 await UniTask.Yield() 로 한 프레임 넘긴 뒤에 구독합니다.

await trigger.UpdateAsync();   // 여기서 깨어난 자리는 트리거가 구독자를 부르는 도중이다

await UniTask.Yield();         // 순회가 끝난 뒤로 넘긴다

RunB(trigger).Forget();        // 이제 둘을 새로 걸어도 대기 목록에 쌓이지 않는다
RunC(trigger).Forget();

정리

UniTask 스트림은 소비자가 당길 때만 값을 줍니다.

• 완료 원천이 세대마다 한 번만 완료되어, 바쁜 동안의 값은 버려집니다. 전부 받으려면 Queue() 입니다.
• 트리거는 메시지를 받을 컴포넌트를 붙이고, 도는 중에 목록이 바뀌어도 구독자마다 한 번씩 건넵니다.
• 한 발행 안에 둘 이상 구독하면 꼬리 표시가 틀어져 하나가 떨어집니다.

그래서 버리면 안 되는 값에는 Queue() 를 끼우고, 되풀이해 기다리는 루프는 핸들러를 받아 두고, 트리거 콜백 안에서는 새 구독을 한 프레임 미룹니다.