기초Channels

Channels ​

Future는 정확히 하나의 값을 약속합니다. 하지만 값이 여러 개이고, 그것들이 조금씩 도착한다면 어떨까요?

첫 장의 CSV 파일을 떠올려 봅시다. 이제 과제가 더 심각해졌습니다. 임포트하는 동안 모든 사용자의 주소를 GeoDirectory와 대조해서 확인해야 합니다. 우리는 이미 코루틴을 쓰는 법을 알고 있으니, 직접적인 방법을 시도해 봅시다.

php
while (($row = fgetcsv($handle)) !== false) {
    spawn(checkAddress(...), $row[$addressIndex]);
}

파일에는 십만 개의 행이 있습니다. 이 코드는 십만 개의 코루틴을 만들고, 그 모두가 병행으로 GeoDirectory로의 연결을 엽니다. 코루틴은 저렴하지만, 연결은 그렇지 않습니다. 서비스는 숨이 막힐 것이고, 우리의 임포트도 함께 막힐 것입니다.

우리가 원하는 것은, 이를테면 열 개의 코루틴이 확인 작업을 처리하고 나머지 주소들은 자기 차례를 기다리는 것입니다. 일부 코루틴이 작업을 꺼내 가고 다른 코루틴이 작업을 추가하는 큐가 필요합니다.

그런 큐를 채널이라고 부릅니다.

Channel ​

채널은 파이프처럼 코루틴을 직접 연결합니다.

php
use Async\Channel;
use function Async\spawn;

$channel = new Channel(5);

spawn(function () use ($channel) {
    $channel->send('hello');
});

echo $channel->recv(); // hello
  • send — 값을 채널에 넣습니다.
  • recv — 채널에서 값을 꺼냅니다.

두 연산 모두 블록되며, 바로 그것이 핵심입니다. 채널이 비어 있으면 recv는 데이터가 나타날 때까지 코루틴을 중단시킵니다. 채널이 가득 차 있으면 대신 send가 중단됩니다. 생성자의 5는 버퍼 크기를 설정합니다. 아무도 값을 집어 가기 전에 채널이 얼마나 많은 값을 보관할 수 있는지를 나타냅니다.

랑데부 ​

버퍼가 0인 채널을 만들면 어떻게 될까요? 값을 저장할 곳이 없으므로, 다른 코루틴이 recv를 호출할 때까지 send는 완료되지 않습니다.

php
$ch = new Channel(0);

spawn(function () use ($ch) {
    echo "before send\n";
    $ch->send('hello');
    echo "after send\n"; // recv 이후에만 실행됨
});

spawn(function () use ($ch) {
    echo "before recv\n";
    echo $ch->recv() . "\n";
});

이런 채널을 랑데부라고 부릅니다. 송신자와 수신자가 같은 시점에 만나도록 강제되며, 그래서 이런 이름이 붙었습니다.

버퍼 채널과의 차이는 보이는 것보다 더 미묘합니다. 버퍼는 "보냈다"는 의미일 뿐 "받았다"는 의미가 아닙니다. send는 값을 떨궈 놓고 넘어가며, 송신자는 그것이 집혔는지, 언제 집혔는지 전혀 알지 못합니다. 랑데부는 전달을 보장합니다. send가 반환되는 순간 그 값은 이미 수신자의 손에 들어가 있습니다. 이것은 작업을 큐에 그냥 방치할 수 없을 때 중요합니다. 예를 들어, 지금 당장 비어 있는 워커에게만 작업을 넘겨줄 때가 그렇습니다.

여기서 데이터는 거의 중요하지 않습니다. 랑데부는 순수한 동기화입니다. 설령 넘긴 것이 null뿐이라 하더라도 두 코루틴은 같은 순간에 만나는 것이 보장됩니다.

워커 풀 ​

임포트를 위한 해법을 조립해 봅시다. 열 개의 워커 코루틴이 채널에서 주소를 꺼내고, 메인 흐름은 파일을 읽으며 채널에 공급합니다.

php
use Async\Channel;
use Async\ChannelException;
use function Async\spawn;

$queue = new Channel(100);

for ($i = 0; $i < 10; $i++) {
    spawn(function () use ($queue) {
        foreach ($queue as $address) {
                checkAddress($address);
        }
    });
}

while (($row = fgetcsv($handle)) !== false) {
    $queue->send($row[$addressIndex]);
}

$queue->close();

채널에서 나오는 각 값은, 열 개가 모두 recv에서 대기 중이더라도, 정확히 하나의 워커에게만 갑니다. 값이 절대 중복되지 않으므로 워커들은 서로 밟지 않습니다. 채널이 알아서 비어 있는 워커들 사이에 작업을 분배합니다.

파일에 행이 아무리 많아도 GeoDirectory로의 연결은 결코 열 개를 넘지 않습니다. 워커의 수가 정확히 동시성 한계이며, 루프의 상수 하나로 설정됩니다.

백프레셔 ​

왜 버퍼가 하필 100일까요? 워커들이 메인 흐름이 파일을 읽는 속도보다 느리게 주소를 확인한다고 상상해 봅시다. 한계가 없다면 큐는 십만 행짜리 파일 전체가 메모리에 앉을 때까지 계속 커질 것입니다.

버퍼가 100이면 다른 일이 벌어집니다. 채널에 백 개의 주소가 쌓이는 즉시 send가 메인 흐름을 중단시킵니다. 파일 읽기가 멈췄다가 워커들이 공간을 비우면 다시 재개됩니다. 생산자가 소비자의 속도에 자동으로 맞춰집니다.

이 메커니즘을 백프레셔라고 부릅니다. 우리가 직접 프로그래밍하지 않았다는 점에 주목하세요. 그것은 send가 블록되는 성질에서 저절로 나옵니다. 버퍼 크기만 고르면, 나머지는 전부 스스로 균형을 맞춥니다.

채널 닫기 ​

파일을 다 읽고 나면 워커들은 계속 recv에서 기다립니다. 익숙한 상황입니다. 취소를 다룬 장에서 진행률 코루틴도 영원히 돌기만 했었죠. 하지만 여기서는 모든 워커에 cancel()을 할 필요가 없습니다. 채널에는 더 정확한 도구가 있습니다.

php
$queue->close();

close는 새 값이 더 오지 않는다고 알립니다. 워커들은 먼저 버퍼에 남아 있는 것을 다 읽고 나서, recv가 ChannelException을 던져 while (true) 루프를 끝냅니다. 순서에 주목하세요. 닫기는 작업을 중간에 끊는 것이 아니라, 작업이 제대로 끝나도록 놔둡니다.

그리고 익숙한 세부 사항이 하나 더 있습니다. recv와 send는 타임아웃 장의 await와 같은 취소 토큰을 받습니다.

php
$address = $queue->recv(timeout(5000));

5초 안에 채널에 아무것도 나타나지 않으면 워커는 AsyncCancellation을 받고, 예를 들어 경고를 로그에 남길 수 있습니다.

데이터 전달인가, 동기화인가? ​

이 장에서 채널이 실제로 무엇을 하고 있었는지 더 자세히 들여다봅시다. 비어 있는 채널에서의 recv는 작업이 나타날 때까지 워커를 재웠습니다. 가득 찬 채널에서의 send는 워커들이 큐를 비울 때까지 파일 읽기를 재웠습니다. 랑데부는 두 코루틴을 같은 시점에 한데 모았습니다. 모든 블록 연산이 코루틴들이 서로에게 맞춰지는 지점으로 드러난 셈입니다.

그리고 여기서 단일 스레드 세계의 미묘한 점을 짚어둘 만합니다. 데이터를 전달하는 것 자체를 위해서라면 채널이 필수는 아닙니다. 코루틴은 공유 메모리 안에 살고 있으며, 객체는 그냥 use를 통해 그것들에게 넘겨줄 수 있습니다. 단일 스레드 안에서는 데이터 경쟁이 일어나지 않습니다. 두 코루틴이 정확히 같은 순간에 실행되는 일은 결코 없기 때문입니다. 채널은 다른 목적으로 쓰입니다. 경계가 있는 큐, 비어 있는 워커들 사이의 작업 분배, close를 통한 "더 이상 작업 없음" 신호가 그것입니다.

그래서 채널은 이렇게 생각하는 편이 더 정확합니다. 채널은 코루틴을 동기화하는 방식이며, 그 과정에서 데이터를 실어 나르는 것이지, 그 반대가 아닙니다. 하지만 코루틴이 서로 다른 스레드에 흩어져 있는 멀티스레드 코드에서는 채널이 두 역할 모두에서 없어서는 안 될 존재가 됩니다. 거기에는 공유 메모리가 없고, 채널이 유일한 안전한 다리로 남기 때문입니다.

이제 우리에게는 코루틴을 서로 연결하는 두 가지 방법이 있습니다. Future는 하나의 값을 약속합니다. 채널은 값의 스트림 전체를 실어 나르며, 그 과정에서 작업에 공유된 리듬을 부여합니다. 생산자는 누가 언제 데이터를 집어 갈지 모르고, 소비자는 그것이 어디에서 왔는지 모르지만, 아무도 다른 누구에게 작업을 쏟아붓지 않습니다.

한 가지 마음에 걸리는 생각이 남습니다. 우리는 열 개의 워커를 띄우고 그냥 넘어갔습니다. 임포트 도중에 그중 하나가 예외로 죽어 버리면 어떻게 될까요? 도대체 누가 이 모든 코루틴을 지켜보고 있는 걸까요? 다음 장에서 알아봅시다.