동시성과 비동기 프로그래밍

동시성과 비동기 프로그래밍 (Concurrency)

대부분의 현대 프로그래밍 언어처럼 Raku도 병렬성(parallelism), 비동기성(asynchronicity), 동시성(concurrency)을 지원하도록 설계되어 있어요. 병렬성은 여러 일을 동시에 하는 것이고요. 비동기 프로그래밍은 프로그램 어딘가에서 촉발된 이벤트로 인한 프로그램 흐름의 변화를 지원하는 것을 뜻해서, 이벤트 주도(event driven) 또는 반응형(reactive) 프로그래밍이라고도 불러요. 마지막으로 동시성은 어떤 공유 자원에 대한 접근과 수정을 조정하는 것에 관한 것입니다.

Raku 동시성 설계의 목표는, 가상 머신이 특정 운영체제에서 어떻게 구현하든, 아래에서 설명하는 기능의 계층을 통해 고수준의, 조합 가능하고(composable) 일관된 인터페이스를 제공하는 거예요.

게다가 어떤 Raku 기능은 암시적으로 비동기 방식으로 동작할 수 있어요. 그래서 이런 기능들과의 예측 가능한 상호운용을 보장하려면, 사용자 코드는 가능한 한 저수준 동시성 API(예: Thread, Scheduler)를 피하고 고수준 인터페이스를 쓰는 게 좋습니다.

고수준 API (High-level APIs)

Promises

Promise는(다른 프로그래밍 환경에서는 future라고도 불러요) promise를 얻는 시점에 아직 완료되지 않았거나 시작조차 안 됐을 수 있는 계산의 결과를 감쌉니다. Promise는 Planned 상태에서 시작해서, 성공적으로 완료됐다는 뜻의 Kept 상태나 실패했다는 뜻의 Broken 상태로 귀결될 수 있어요. 보통 이것만으로도 사용자 코드가 동시적·비동기적으로 동작하는 데 필요한 기능의 상당 부분을 차지합니다.

my $p1 = Promise.new;
say $p1.status;         # OUTPUT: «Planned␤»
$p1.keep('Result');
say $p1.status;         # OUTPUT: «Kept␤»
say $p1.result;         # OUTPUT: «Result␤»
                        # (since it has been kept, a result is available!)

my $p2 = Promise.new;
$p2.break('oh no');
say $p2.status;         # OUTPUT: «Broken␤»
say $p2.result;         # dies, because the promise has been broken
CATCH { default { say .^name, ': ', .Str } };
# OUTPUT: «X::AdHoc+{X::Promise::Broken}: oh no␤»

Promise의 힘의 상당 부분은 조합 가능하다는 데서 나와요. 예를 들어 보통 then 메서드로 연결(chaining)할 수 있죠.

my $promise1 = Promise.new();
my $promise2 = $promise1.then(
    -> $v { say $v.result; "Second Result" }
);
$promise1.keep("First Result");
say $promise2.result;   # OUTPUT: «First Result␤Second Result␤»

여기서 then 메서드는 첫 Promise가 kept되거나 broken될 때 실행될 코드를 예약하는데, 그 자신도 새 Promise를 반환해요. 그 새 promise는 코드가 실행되면(성공하면) 그 결과로 kept되거나(실패하면) broken됩니다. keep는 promise 상태를 Kept로 바꾸고 위치 인자를 결과로 설정해요. result는 promise가 kept·broken될 때까지 현재 실행 스레드를 블록하는데, kept됐다면 결과를 반환하고(keep에 넘긴 값) 아니라면 break에 넘긴 값에 기반한 예외를 던집니다. 후자의 동작은 이렇게 설명돼요.

my $promise1 = Promise.new();
my $promise2 = $promise1.then(-> $v { say "Handled but : "; say $v.result});
$promise1.break("First Result");
try $promise2.result;
say $promise2.cause;        # OUTPUT: «Handled but : ␤First Result␤»

여기서 break는 then의 코드 블록이, 인자로 넘겨진 원래 promise에 대해 result 메서드를 호출할 때 예외를 던지게 만들고, 그게 두 번째 promise를 broken하게 만들며, 그 결과를 가져올 때 차례로 예외를 일으켜요. 실제 Exception 객체는 cause에서 얻을 수 있습니다. 만약 promise가 broken되지 않았다면 cause는 X::Promise::CauseOnlyValidOnBroken 예외를 일으켜요.

Promise는 미래 시점에 자동으로 kept되도록 예약할 수도 있어요.

my $promise1 = Promise.in(5);
my $promise2 = $promise1.then(-> $v { say $v.status; 'Second Result' });
say $promise2.result;

in 메서드는 새 promise를 만들고, 주어진 초수 이상 지난 후에 그 위에서 keep을 호출하는 새 작업을 예약해서, 새 Promise 객체를 반환해요.

promise의 매우 흔한 용도는 코드 한 조각을 실행하고, 성공적으로 반환하면 promise를 kept하고, 코드가 죽으면 broken하는 것이에요. start 메서드가 그 단축키를 제공합니다.

my $promise = Promise.start(
    { my $i = 0; for 1 .. 10 { $i += $_ }; $i}
);
say $promise.result;    # OUTPUT: «55␤»

여기서 반환된 promise의 result는 코드의 반환값이에요. 마찬가지로 코드가 실패하면(그래서 promise가 broken되면), cause는 던져진 Exception 객체가 됩니다.

my $promise = Promise.start({ die "Broken Promise" });
try $promise.result;
say $promise.cause;

이 패턴은 너무 흔해서 키워드로도 제공됩니다.

my $promise = start {
    my $i = 0;
    for 1 .. 10 {
        $i += $_
    }
    $i
}
my $result = await $promise;
say $result;

await 서브루틴은 start가 반환한 promise 객체에 result를 호출하는 것과 거의 동등하지만, promise의 리스트도 받아 각각의 결과를 반환합니다.

my $p1 = start {
    my $i = 0;
    for 1 .. 10 {
        $i += $_
    }
    $i
};
my $p2 = start {
    my $i = 0;
    for 1 .. 10 {
        $i -= $_
    }
    $i
};
my @result = await $p1, $p2;
say @result;            # OUTPUT: «[55 -55]␤»

await 외에도, 두 개의 클래스 메서드가 여러 Promise 객체를 하나의 새 promise로 결합합니다. allof는 원래 promise들이 모두 kept·broken될 때 kept되는 promise를 반환해요.

my $promise = Promise.allof(
    Promise.in(2),
    Promise.in(3)
);

await $promise;
say "All done"; # Should be not much more than three seconds later

그리고 anyof는 원래 promise들 중 어떤 것이든 kept·broken되면 kept될 새 promise를 반환해요.

my $promise = Promise.anyof(
    Promise.in(3),
    Promise.in(8600)
);

await $promise;
say "All done"; # Should be about 3 seconds later

await와 달리 원래 kept된 promise들의 결과는 원래 promise를 참조하지 않고서는 얻을 수 없어요. 그래서 이 둘은, 작업의 완료 여부 자체가 실제 결과보다 소비자에게 더 중요할 때나, 결과를 다른 수단으로 모았을 때가 더 유용합니다. 예를 들어 각 원래 promise를 검사하는 의존 promise를 만들고 싶을 수 있죠.

my @promises;
for 1..5 -> $t {
    push @promises, start {
        sleep $t;
        Bool.pick;
    };
}
say await Promise.allof(@promises).then({ so all(@promises>>.result) });

이것은 모든 promise가 True로 kept되면 True를, 그렇지 않으면 False를 줍니다.

직접 keep하거나 break하려는 promise를 만들고 있다면, 그 promise를 받을 수 있는 어떤 코드도 당신보다 먼저 그 promise를 실수로(혹은 의도적으로) keep·break하지 않길 원할 거예요. 이를 위해 vow 메서드가 있어요. 이 메서드는 Vow 객체를 반환하는데, 그 객체가 promise를 keep·break할 수 있는 유일한 메커니즘이 됩니다. vow 객체를 private하게만 유지하면, promise를 직접 keep·break하려는 시도가 X::Promise::Vowed 예외를 던지므로 promise의 상태는 안전해요.

sub get_promise {
    my $promise = Promise.new;
    my $vow = $promise.vow;
    Promise.in(10).then({$vow.keep});
    $promise;
}

my $promise = get_promise();

# Will throw an exception
# "Access denied to keep/break this Promise; already vowed"
$promise.keep;
CATCH { default { say .^name, ': ', .Str } };
# OUTPUT: «X::Promise::Vowed: Access denied to keep/break this Promise; already vowed␤»

in이나 start처럼 자동으로 kept·broken될 promise를 반환하는 메서드들은 이 vow를 이미 처리하므로, 그것들을 위해 이걸 할 필요는 없어요.

Supplies

Supply는 하나 이상의 소비자가 동시에, 다른 언어의 "이벤트"와 비슷한 방식으로 소비할 수 있는 비동기 데이터 스트리밍 메커니즘이에요. 이벤트 주도나 반응형 설계를 가능하게 한다고 볼 수 있습니다.

가장 단순하게는, Supply는 tap 메서드로 여러 구독자를 만들 수 있는 메시지 스트림이고, 데이터 항목은 emit으로 넣을 수 있어요.

Supply에는 live와 on demand 두 종류가 있어요. live supply는 TV 방송과 같아서, 나중에 채널을 맞춘 사람은 이전에 방출된 값을 받지 못해요. on-demand 방송은 비디오 스트리밍 서비스와 같아서, 영화를 켜기 시작하는 모든 사람(즉 Supply를 tap하는 사람)은 지금 몇 명이 보고 있든 관계없이 항상 처음부터 시작합니다(모든 값을 받아요).

on-demand supply에는 히스토리가 보관되지 않는다는 점을 기억하세요. 대신 supply가 tap될 때마다 supply 블록이 실행됩니다.

live Supply는 Supplier 팩토리로 만들어져요. 방출된 각 값은 활성 tapper들이 추가될 때 그 모두에게 전달됩니다.

my $supplier = Supplier.new;
my $supply   = $supplier.Supply;

$supply.tap( -> $v { say $v });

for 1 .. 10 {
    $supplier.emit($_);
}

tap은 Supplier가 만든 Supply 객체에 호출되고, 새 값은 Supplier에서 방출된다는 점을 기억하세요.

on-demand Supply는 supply 키워드로 만들어요.

my $supply = supply {
    for 1 .. 10 {
        emit($_);
    }
}
$supply.tap( -> $v { say $v });

이 경우 supply 블록의 코드는 supply가 반환한 Supply가 tap될 때마다 실행됩니다. 이것은 다음처럼 증명돼요.

my $supply = supply {
    for 1 .. 10 {
        emit($_);
    }
}
$supply.tap( -> $v { say "First : $v" });
$supply.tap( -> $v { say "Second : $v" });

tap 메서드는 Tap 객체를 반환하는데, 이 객체로 tap에 대한 정보를 얻을 수 있고, 더 이상 이벤트에 관심이 없을 때 tap을 끄는 데도 씁니다.

my $supplier = Supplier.new;
my $supply   = $supplier.Supply;

my $tap = $supply.tap( -> $v { say $v });

$supplier.emit("OK");
$tap.close;
$supplier.emit("Won't trigger the tap");

supply 객체에서 done을 호출하면, tap에 대해 지정된 done 콜백을 호출하지만, 스트림에 더 이상 이벤트가 방출되거나 tap이 그것을 받는 것은 막지 않아요.

interval 메서드는 지정된 간격으로 주기적으로 새 이벤트를 방출하는 새 on-demand supply를 반환합니다. 방출되는 데이터는 0에서 시작해서 이벤트마다 증가하는 정수예요. 다음 코드는 0 .. 5를 출력합니다.

my $supply = Supply.interval(2);
$supply.tap(-> $v { say $v });
sleep 10;

interval에 두 번째 인자를 줄 수 있는데, 첫 이벤트가 발화되기 전의 지연 시간을 초 단위로 지정해요. interval이 만든 supply를 tap할 때마다 0부터 시작하는 자신만의 시퀀스를 갖습니다. 다음 예시가 그걸 보여줘요.

my $supply = Supply.interval(2);
$supply.tap(-> $v { say "First $v" });
sleep 6;
$supply.tap(-> $v { say "Second $v"});
sleep 10;

첫 tap까지 값을 보관하는 live Supply는 Supplier::Preserving으로 만들 수 있어요.

whenever

whenever 키워드는 supply 블록이나 react 블록 안에서 쓸 수 있어요. 6.d 버전부터는 그들의 렉시컬 스코프 안에서 사용해야 합니다. 이 키워드는 자신이 지정한 비동기 이벤트에 의해 촉발될 때 실행될 코드 블록을 도입하는데, 그 이벤트는 Supply, Channel, Promise, 또는 Iterable일 수 있어요.

whenever 안의 코드는 가능한 한 작게 유지해야 한다는 점을 명심하세요. 어느 순간에 실행되는 whenever 블록은 하나뿐이니까요. 더 오래 걸리는 코드를 실행하려면 whenever 블록 안에서 start 블록을 쓸 수 있습니다.

이 예시에서는 supply 두 개를 지켜보고 있어요.

my $bread-supplier = Supplier.new;
my $vegetable-supplier = Supplier.new;

my $supply = supply {
    whenever $bread-supplier.Supply {
        emit("We've got bread: " ~ $_);
    };
    whenever $vegetable-supplier.Supply {
        emit("We've got a vegetable: " ~ $_);
    };
}
$supply.tap( -> $v { say "$v" });

$vegetable-supplier.emit("Radish");   # OUTPUT: «We've got a vegetable: Radish␤»
$bread-supplier.emit("Thick sliced"); # OUTPUT: «We've got bread: Thick sliced␤»
$vegetable-supplier.emit("Lettuce");  # OUTPUT: «We've got a vegetable: Lettuce␤»

react

react 키워드는 비동기 이벤트를 지켜볼 whenever 키워드를 하나 이상 담은 코드 블록을 도입해요. supply 블록과 react 블록의 주된 차이는, react 블록의 코드는 코드 흐름에서 자기 위치에서 실행되는 반면, supply 블록은 아무것도 하기 전에 tap되어야 한다는 점이에요.

또 다른 차이는, supply 블록은 whenever 키워드 없이도 쓸 수 있지만, react 블록은 실질적으로 쓰이려면 whenever가 최소 하나 필요하다는 점입니다.

react {
    whenever Supply.interval(2) -> $v {
        say $v;
        done() if $v == 4;
    }
}

여기서 whenever 키워드는 제공된 블록에서 tap을 만들기 위해 .act을 사용해요. tap 중 하나에서 done()이 호출되면 react 블록이 종료됩니다.

on-demand Supply는 차례로 방출될 값들의 리스트로도 만들 수 있어요. 그래서 첫 on-demand 예시는 이렇게 쓸 수 있습니다.

react {
    whenever Supply.from-list(1..10) -> $v {
        say $v;
    }
}

supply 변환하기 (Transforming supplies)

기존 supply 객체는 grep과 map 메서드로 각각 필터링하거나 변환해서 새 supply를 만들 수 있어요. 이름이 비슷한 리스트 메서드처럼 동작합니다. grep은 소스 스트림에서 방출된 이벤트 중 grep 조건이 참인 것만 두 번째 supply에서 방출되게 해요.

my $supplier = Supplier.new;
my $supply = $supplier.Supply;
$supply.tap(-> $v { say "Original : $v" });
my $odd_supply = $supply.grep({ $_ % 2 });
$odd_supply.tap(-> $v { say "Odd : $v" });
my $even_supply = $supply.grep({ not $_ % 2 });
$even_supply.tap(-> $v { say "Even : $v" });
for 0 .. 10 {
    $supplier.emit($_);
}

map은 새 supply를 반환하는데, 원래 supply에 방출된 각 항목에 대해 map에 넘긴 표현식의 결과인 새 항목을 방출해요.

my $supplier = Supplier.new;
my $supply = $supplier.Supply;
$supply.tap(-> $v { say "Original : $v" });
my $half_supply = $supply.map({ $_ / 2 });
$half_supply.tap(-> $v { say "Half : $v" });
for 0 .. 10 {
    $supplier.emit($_);
}

supply 끝내기 (Ending a supply)

supply가 끝날 때 실행할 동작이 필요하다면, tap 호출에서 done과 quit 옵션을 설정하면 돼요.

$supply.tap: { ... },
    done => { say 'Job is done.' },
    quit => {
        when X::MyApp::Error { say "App Error: ", $_.message }
    };

quit 블록은 CATCH와 매우 비슷하게 동작해요. 예외가 when이나 default 블록으로 "seen"으로 표시되면, 그 예외는 잡혀 처리됩니다. 그렇지 않으면 예외는 계속 콜 트리를 올라가요(즉 quit이 설정되지 않았을 때와 같은 동작).

supply나 react 블록의 페이저 (Phasers in a supply or react block)

whenever와 함께 react나 supply 블록 문법을 사용한다면, whenever 블록 안에 페이저(phaser)를 추가해서 tap된 supply의 done과 quit 메시지를 처리할 수 있어요.

react {
    whenever $supply {
        ...; # your usual supply tap code here
        LAST { say 'Job is done.' }
        QUIT { when X::MyApp::Error { say "App Error: ", $_.message } }
    }
}

여기서의 동작은 tap에 done과 quit을 설정하는 것과 같습니다.

Channels

Channel은 스레드 안전한 대기열(queue)로, 여러 읽는 사람과 쓰는 사람을 가질 수 있어요. 동작이 "fifo"나 named pipe와 비슷하다고 볼 수 있지만 프로세스 간 통신은 지원하지 않아요. 진짜 대기열이기 때문에 Channel에 보낸 각 값은 먼저 읽는 사람이 먼저 얻는(first read, first served) 방식으로 단일 읽는 사람만 가질 수 있다는 점을 기억해야 해요. 여러 읽는 사람이 보낸 모든 항목을 받게 하려면 Supply를 고려해 보는 게 좋습니다.

항목은 send 메서드로 Channel에 대기열에 넣고, receive 메서드는 대기열에서 항목을 제거해 반환합니다. 대기열이 비어 있으면 새 항목이 보내질 때까지 블록돼요.

my $channel = Channel.new;
$channel.send('Channel One');
say $channel.receive;  # OUTPUT: «Channel One␤»

채널이 close 메서드로 닫혔다면, 어떤 send도 X::Channel::SendOnClosed 예외를 일으키고, receive는 X::Channel::ReceiveOnClosed를 던집니다.

list 메서드는 Channel의 모든 항목을 반환하고, 추가 항목이 대기열에 쌓일 때까지 블록합니다(채널이 닫히지 않았다면요).

my $channel = Channel.new;
await (^10).map: -> $r {
    start {
        sleep $r;
        $channel.send($r);
    }
}
$channel.close;
for $channel.list -> $r {
    say $r;
}

비블로킹 poll 메서드도 있어요. 채널에서 사용 가능한 항목을 반환하고, 항목이 없거나 채널이 닫혀 있으면 Nil을 반환합니다. 이는 채널이 닫혔는지 판단하기 위해 채널을 확인해야 한다는 뜻이에요.

my $c = Channel.new;

# Start three Promises that sleep for 1..3 seconds, and then
# send a value to our Channel
^3 .map: -> $v {
    start {
        sleep 3 - $v;
        $c.send: "$v from thread {$*THREAD.id}";
    }
}

# Wait 3 seconds before closing the channel
Promise.in(3).then: { $c.close }

# Continuously loop and poll the channel, until it's closed
my $is-closed = $c.closed;
loop {
    if $c.poll -> $item {
        say "$item received after {now - INIT now} seconds";
    }
    elsif $is-closed {
        last;
    }

    say 'Doing some unrelated things...';
    sleep .6;
}

# Doing some unrelated things...
# Doing some unrelated things...
# 2 from thread 5 received after 1.2063182 seconds
# Doing some unrelated things...
# Doing some unrelated things...
# 1 from thread 4 received after 2.41117376 seconds
# Doing some unrelated things...
# 0 from thread 3 received after 3.01364461 seconds
# Doing some unrelated things...

closed 메서드는 채널이 닫히면 kept되는(그래서 Boolean 컨텍스트에서 True로 평가되는) Promise를 반환합니다.

.poll 메서드는 .receive 메서드와 조합해서 캐싱 메커니즘으로 쓸 수 있어요. .poll이 값을 안 돌려주는 게 더 많은 값을 fetch해서 채널에 채워야 한다는 신호로 쓰이는 거죠.

sub get-value {
    return $c.poll // do { start replenish-cache; $c.receive };
}

sub replenish-cache {
    for ^20 {
        $c.send: $_ for slowly-fetch-a-thing();
    }
}

Channel은 앞서 설명한 react 블록의 whenever에서 Supply 대신 쓸 수 있어요.

my $channel = Channel.new;
my $p = start {
    react {
        whenever $channel {
            say $_;
        }
    }
}

await (^10).map: -> $r {
    start {
        sleep $r;
        $channel.send($r);
    }
}

$channel.close;
await $p;

Supply에서 Channel 메서드를 사용해 Channel을 얻을 수도 있어요. 이 메서드는 Supply에 대한 tap이 공급하는 Channel을 반환합니다.

my $supplier = Supplier.new;
my $supply   = $supplier.Supply;
my $channel = $supply.Channel;

my $p = start {
    react  {
        whenever $channel -> $item {
            say "via Channel: $item";
        }
    }
}

await (^10).map: -> $r {
    start {
        sleep $r;
        $supplier.emit($r);
    }
}

$supplier.done;
await $p;

Channel은 호출될 때마다 같은 데이터를 공급하는 서로 다른 Channel을 반환해요. 예를 들어 프로그램에서 서로 다른 인터페이스를 제공하기 위해 Supply를 하나 이상의 Channel로 팬아웃(fan-out)하는 데 쓸 수 있습니다.

Proc::Async

Proc::Async는 앞서 설명한 기능 위에 쌓여, 외부 프로그램을 비동기적으로 실행하고 상호작용합니다.

my $proc = Proc::Async.new('echo', 'foo', 'bar');

$proc.stdout.tap(-> $v { print "Output: $v" });
$proc.stderr.tap(-> $v { print "Error:  $v" });

say "Starting...";
my $promise = $proc.start;

await $promise;
say "Done.";

# Output:
# Starting...
# Output: foo bar
# Done.

명령의 경로와 명령에 대한 인자들은 생성자에 제공돼요. 명령은 start가 호출될 때까지 실행되지 않는데, start는 프로그램이 종료되면 kept될 Promise를 반환합니다. 프로그램의 표준 출력과 표준 오류는 각각 stdout과 stderr 메서드에서 Supply 객체로 얻을 수 있고, 필요에 따라 tap할 수 있어요.

프로그램의 표준 입력에 쓰고 싶다면 생성자에 :w 부사(adverb)를 제공하고, 프로그램이 시작된 뒤 열린 파이프에 write, print, say 메서드를 사용해 쓸 수 있어요.

my $proc = Proc::Async.new(:w, 'grep', 'foo');

$proc.stdout.tap(-> $v { print "Output: $v" });

say "Starting...";
my $promise = $proc.start;

$proc.say("this line has foo");
$proc.say("this one doesn't");

$proc.close-stdin;
await $promise;
say "Done.";

# Output:
# Starting...
# Output: this line has foo
# Done.

어떤 프로그램(이 예시에서 파일 인자 없는 grep 같은)은 표준 입력이 닫힐 때까지 종료하지 않아요. 그래서 close-stdin을 다 썼을 때 호출해서 start가 반환한 Promise가 kept될 수 있게 합니다.

저수준 API (Low-level APIs)

Threads

동시성을 위한 가장 낮은 수준의 인터페이스는 Thread가 제공해요. 스레드는 언젠가 프로세서에서 실행될 수 있는 코드 조각으로 생각할 수 있는데, 그 배치는 거의 전적으로 가상 머신과/또는 운영체제가 정합니다. 스레드는 사실상 크게 관리되지 않는다고 여겨야 하며, 사용자 코드에서의 직접 사용은 피해야 해요.

스레드는 만들어서 나중에 실제로 실행할 수 있고,

my $thread = Thread.new(code => { for  1 .. 10  -> $v { say $v }});
# ...
$thread.run;

단일 호출로 만들고 실행할 수도 있어요.

my $thread = Thread.start({ for  1 .. 10  -> $v { say $v }});

두 경우 모두 Thread 객체가 감싼 코드의 완료를 finish 메서드로 기다릴 수 있는데, 스레드가 완료될 때까지 블록합니다.

$thread.finish;

그 외에는 동기화나 자원 공유를 위한 기능이 더 없어요. 그래서 스레드가 사용자 코드에서 직접 유용할 가능성이 낮다는 점을 강조하는 이유입니다.

Schedulers

동시성 API의 다음 수준은 Scheduler 역할이 정의한 인터페이스를 구현하는 클래스들이 제공해요. 스케줄러 인터페이스의 의도는 특정 작업을 실행할 자원과 시점을 결정하는 메커니즘을 제공하는 것입니다. 고수준 동시성 API 대부분은 스케줄러 위에 구축되어 있어서 사용자 코드가 아예 쓸 필요가 없을 수 있어요. 다만 Proc::Async, Promise, Supply의 어떤 메서드들은 스케줄러를 명시적으로 제공할 수 있게 해 줍니다.

현재 기본 전역 스케줄러는 변수 $*SCHEDULER에서 사용 가능해요.

스케줄러의 주 인터페이스(실제로 Scheduler 인터페이스가 요구하는 유일한 메서드)는 cue 메서드입니다.

method cue(:&code, Instant :$at, :$in, :$every, :$times = 1; :&catch)

이 메서드는 &code의 Callable을 부사들이 결정한 방식대로(Scheduler에 문서화된 대로) 스케줄러가 구현한 실행 방식을 사용해 실행하도록 예약해요. 예를 들어,

my $i = 0;
my $cancellation = $*SCHEDULER.cue({ say $i++}, every => 2 );
sleep 20;

$*SCHEDULER가 기본값에서 바뀌지 않았다고 가정하면, 이 코드는 숫자 0부터 10까지를 대략(즉 운영체제 스케줄링 허용 범위 내에서) 2초마다 출력합니다. 이 경우 코드는 프로그램이 정상 종료될 때까지 실행되도록 예약되지만, 메서드는 Cancellation 객체를 반환하는데, 이를 사용해 정상 완료 전에 예약된 실행을 취소할 수 있어요.

my $i = 0;
my $cancellation = $*SCHEDULER.cue({ say $i++}, every => 2 );
sleep 10;
$cancellation.cancel;
sleep 10;

이것은 0부터 5까지만 출력해야 합니다.

Scheduler 인터페이스가 Thread보다 분명한 이점을 제공함에도 불구하고, 전체 기능이 고수준 인터페이스로 이용 가능하므로 스케줄러를 직접 쓸 필요는 없어요. 다만 위에서 언급한 것처럼 특정 메서드에 스케줄러를 명시적으로 제공해야 하는 경우는 예외일 수 있어요.

라이브러리는 특별한 요구사항이 있다면 대안 스케줄러 구현을 제공하고 싶을 수 있어요. 예를 들어 UI 라이브러리는 모든 코드가 단일 UI 스레드 안에서 실행되길 원하거나, 어떤 커스텀 우선순위 메커니즘이 필요할 수 있죠. 다만 표준으로 제공되고 아래에 설명되는 구현들이 대부분의 사용자 코드에는 충분할 거예요.

ThreadPoolScheduler

ThreadPoolScheduler가 기본 스케줄러예요. 필요에 따라 할당되는 스레드 풀을 유지하고, 필요하면 새로 만듭니다.

Rakudo는 프로그램 시작 시 환경 변수 RAKUDO_MAX_THREADS로 기본 스케줄러에서 허용되는 최대 스레드 수를 설정할 수 있게 해 줍니다.

최대값을 초과하면 cue가 스레드가 생길 때까지 코드를 대기시킬 수 있어요.

CurrentThreadScheduler

CurrentThreadScheduler는 매우 단순한 스케줄러로, 항상 코드를 현재 스레드에서 즉시 실행하도록 예약해요. 이 스케줄러의 cue는 코드 실행이 끝날 때까지 블록한다는 뜻이라, 테스팅 같은 특정 특수한 경우로 그 용도가 제한됩니다.

Locks

Lock 클래스는 동시성 환경에서 공유 데이터를 보호하는 저수준 메커니즘을 제공하며, 그래서 고수준 API에서 스레드 안전성을 지원하는 핵심이에요. 다른 프로그래밍 언어에서는 "Mutex"라고도 불러요. 고수준 클래스들(Promise, Supply, Channel)이 필요할 때 Lock을 사용하므로, 사용자 코드가 Lock을 직접 쓸 일은 거의 없을 거예요.

Lock의 주 인터페이스는 protect 메서드로, 코드 블록(보통 "임계 구간(critical section)"이라고 함)이 한 번에 한 스레드에서만 실행되도록 보장합니다.

my $lock = Lock.new;

my $a = 0;

await (^10).map: {
    start {
        $lock.protect({
            my $r = rand;
            sleep $r;
            $a++;
        });
    }
}

say $a; # OUTPUT: «10␤»

protect는 코드 블록이 반환하는 무엇이든 반환합니다.

protect는 임계 구간을 실행하려고 기다리는 스레드들을 블록하므로, 코드는 가능한 한 빨라야 해요.

Lock::Async

Lock::Async는 Lock과 비슷한 상호 배제(mutual exclusion) 메커니즘이지만, 기능을 Promise로 노출해서, 전체 스레드를 블록하는 대신 락을 기다릴 때 await를 사용할 수 있게 해 줍니다.

또 다른 차이는 재진입(re-entrant)이 아니라는 점이에요. 즉 그 락이 protected하는 코드에는 그 락이 이용 가능한 것으로 간주되지 않는다는 뜻입니다.

Lock::Async는 Lock보다 고수준이지만, 여전히 저수준 원시 장치(primitive)로 간주돼요. 임계 구간 안에서 공유 데이터를 변형하는 것보다는 고수준 원시 장치를 선호해야 합니다.

안전성 우려 (Safety concerns)

어떤 공유 데이터 동시성 문제는 다른 것보다 덜 명확해요. 이 주제에 대한 좋은 일반적인 글은 이 블로그 포스트를 보세요.

특히 주목할 문제 하나는 컨테이너 자동생성(autovivification)이나 확장이 일어날 때예요. Array나 Hash 항목이 처음 할당될 때 기본 구조가 변경되는데, 그 연산은 비동기 안전하지 않습니다. 예를 들어 이 코드에서,

my @array;
my $slot := @array[20];
$slot = 'foo';

세 번째 줄이 임계 구간이에요. 배열이 확장되는 때이기 때문이죠. 가장 간단한 수정은 Lock으로 임계 구간을 보호하는 거예요. 어쩌면 더 나은 수정은 컨테이너를 공유할 필요가 없도록 코드를 리팩터링하는 것입니다.