rokdocs
v0.1

2.18. task - M:N 스케줄러 task API

rok::core::task 모듈은 M:N 스케줄러 위에서 동작하는 협력적 task API 를 제공합니다. 이 모듈은 Rust Tokio 와 Go goroutine 으로부터 영감을 받았으며, work-stealing 스케줄러 위에서 task 가 stackless coroutine 으로 구현됩니다.

use rok::core::**; 가 묵시적으로 적용되므로 task::spawn, task::race, Task<T> 등은 별도의 import 없이도 사용할 수 있습니다. 다만 모듈 prefix task:: 는 그대로 보존됩니다. 한편 명시적으로 비동기 코드임을 드러내고 싶은 자리에서는 use rok::core::task::*; 를 권장하는데, 이는 시각적 분리를 통해 비동기 흐름을 명확히 하기 위함입니다.

2.18.1. 핵심 API

핵심 API 의 시그니처는 다음과 같습니다.

fn block_on<F: Future>(f: F) -> F::Output;              // sync 진입점 - executor 에 붙어 f 를 끝까지 실행
fn spawn<F: Future + Send>(f: F) -> Task<F::Output> where F::Output: Send;
fn spawn_local<F: Future>(f: F) -> Task<F::Output>;     // non-Send future - 현재 worker 에 고정
fn spawn_blocking<T: Send>(f: fn() -> T + Send) -> Blocking<T>;   // blocking 작업을 executor 밖 OS 스레드에서
fn workers() -> usize;                                  // executor 안에서는 pool 크기, 밖에서는 1
fn cpu_count() -> usize;                                // 온라인 CPU 수 - pool 크기
async fn race<T>(tasks: List<Task<T>>) -> T;            // 첫 완료 task 결과
fn select<F: Future>(futs: List<F>) -> SelectFuture<F>; // await - (index, 결과), spawn 하지 않음
fn id() -> TaskId;                                      // 현재 task 의 id - task 밖이면 panic
fn try_id() -> TaskId | none;
async fn join<T>(tasks: List<Task<T>>) -> List<T>;      // 모두 완료 후
async fn timeout<T>(dur: Duration, task: Task<T>) throws (TimeoutError) -> T;   // try await 에서 던짐
async fn sleep(dur: Duration);
async fn sleep_until(when: Instant);
async fn yield_now();                                   // await - 다른 ready task 에 한 번 양보
fn registrations() -> usize;                            // 진단용 - 실행 중 pool 의 park 한 waker + 대기 timer 수 (구현이 정함)
fn readable(fd: i32) -> IoReady;                       // await - fd 가 읽기 가능해질 때까지 (refs-lang 7.7.2)
fn writable(fd: i32) -> IoReady;
fn wake_io(fd: i32);                                   // fd 의 readable / writable 대기자를 깨움 (close / shutdown 뒤에 씀; 저수준)

sleep 과 timeout 은 첫 poll 부터 시간을 잽니다.

사용 예

실제 사용 예시는 다음과 같습니다.

use rok::core::task::*;

let t = task::spawn(async { compute_heavy() });
let result = await t;

let results = await task::join([
    task::spawn(async { fetch("a") }),
    task::spawn(async { fetch("b") }),
    task::spawn(async { fetch("c") }),
]);

let first = await task::race([
    task::spawn(async { source_a() }),
    task::spawn(async { source_b() }),
]);

await task::sleep(Duration::ms(100));
await task::yield_now();

spawn 은 클로저가 아니라 future 를 받으므로 task::spawn(async { ... }) 처럼 async 블록이나 async fn 호출 결과를 넘깁니다. Send 가 아닌 future 는 spawn_local 로 현재 스레드에서 실행합니다. join / race / timeout 은 await 해서 결과를 받습니다.

spawn_blocking - blocking 작업

struct Blocking<T> { /* 결과 슬롯 */ }
impl<T> Future for Blocking<T> { type Output = T; ... }

spawn_blocking(f) 는 f 를 executor 의 worker 가 아닌 blocking pool 의 OS 스레드에서 실행하고, 그 결과로 끝나는 future 를 돌려줍니다. readiness 로 기다릴 수 없는 blocking syscall (일반 파일 I/O 등, stdlib 3.1.4a) 을 task 안에서 기다릴 때 씁니다.

  • f 는 첫 poll 이 아니라 호출하는 즉시 시작합니다 (pool 이 가득 차면 먼저 큐에 들어감).
  • blocking pool 은 프로세스 전역 하나이며, 필요할 때 스레드를 늘리고 (상한 있음) 한동안 일이 없는 스레드는 끝냅니다. 작업은 들어온 순서 (FIFO) 로 처리합니다. 상한은 스레드 512 개이며, 상한에 닿으면 작업을 큐에 쌓습니다. 10 초 동안 일이 없는 스레드는 끝납니다. 작업 하나는 쉬고 있는 스레드 하나에 넘깁니다.
  • await 하던 task 가 취소되어도 f 는 끝까지 실행되며, 결과만 버려집니다. 
let n = await task::spawn_blocking(|| heavy_checksum(path));

spawn 의 Send 검사

task::spawn 은 결과 타입도 Send 여야 하며 (where F::Output: Send), frame 이 Send 가 아닌 future 를 거부합니다 (2.13.9). future 가 어디서 왔든 같습니다: async 블록이나 async fn 호출뿐 아니라 매개변수, struct / 튜플 / 합의 필드, helper 함수의 결과, 리스트의 원소, 클로저가 들고 있는 future, 다른 코루틴 frame 안의 future 도 검사합니다. future 의 frame 은 다음입니다.

  • 캡처한 값.
  • await 를 넘어 살아 있는 지역 변수와 임시값.
  • frame 이 들고 있는 다른 future (그 frame 도 재귀로). 
let c = Cell::new(0);                      // Cell: !Send
task::spawn(async { c.set(1); });          // ERROR - !Send 캡처
task::spawn_local(async { c.set(1); });    // OK - 현재 worker 에 고정

fn make() -> AsyncFuture<()> { let c = Cell::new(0); async { c.set(1); } }
task::spawn(make());                       // ERROR - helper 가 만든 future 의 frame 도 검사

같은 검사가 다른 스레드로 클로저를 보내는 자리에도 적용됩니다: thread::spawn, thread::Builder::spawn, par 의 Scope::spawn, task::spawn_blocking. 클로저가 Send 가 아닌 frame 의 future 를 들고 있으면 `X` would move a future to another thread: ... 에러입니다.

이 검사는 보수적으로 근사해도 됩니다. 구현은 실제로는 Send 인 future 를 거부할 수 있지만, Send 가 아닌 값을 다른 스레드로 넘기는 future 를 받아서는 안 됩니다. 다음 값은 따라 들어가지 않습니다.

  • 타입이 함수 값을 담을 수 없는 값 (future 의 frame 을 숨길 수 없음).
  • unsafe impl Send 를 명시한 타입의 값 (예: Task<T>). 그 타입이 안전성을 책임집니다.

async fn 의 결과 타입 AsyncFuture<T> 는 T: Send 일 때, throwing 판 TryAsyncFuture<T, E> 는 T 와 E 가 모두 Send 일 때 Send 입니다. 캡처와 frame 은 위 검사가 따로 봅니다. 캡처가 Send 이지만 Sync 가 아닌 frame (async fn 이 받는 리스트, async 블록이 캡처하는 리스트나 future) 은 받아들입니다: task::spawn 은 채널 send 처럼 future 를 넘깁니다 (2.13.9). task 는 future 가 캡처한 것의 복사본을 담은 자기 frame 을 받는데, 넘김 한 번으로 만들므로 캡처들이 공유하던 것 (한 리스트 두 번, 리스트와 그 위의 future) 은 그대로 공유합니다. spawn 뒤에 그 future 나 그것이 캡처한 것을 쓰면 send 뒤와 같이 컴파일 에러 (E202) 입니다. 아무것도 poll 하지 않은 future 만 넘길 수 있습니다: 시작한 것 (select, 사용자의 Future, 직접 poll 한 것) 이나 끝난 것, 또는 frame 이나 캡처한 future 의 frame 에 [ref x] 캡처가 있는 것을 넘기면 panic 하며, 아무것도 공유하지 않습니다. 채널로 보내거나 Mutex 나 Frozen 에 넣는 코루틴 future 도 같이 넘겨지고, 그때마다 자기 frame 을 받습니다. Task 가 넘겨진 task (Frozen::get, 채널, 잠금, par 항목) 를 await 할 때마다 결과의 자기 복사본을 받으며, 한 스레드 안의 Task 복사본들은 그것을 공유합니다.

select - 먼저 끝나는 future

fn select<F: Future>(futs: List<F>) -> SelectFuture<F>;   // Output = (usize, F::Output)

let (i, v) = await task::select([fetch_a(), fetch_b()]);

race 와 달리 future 들을 spawn 하지 않고, 기다리는 task 안에서 실행합니다.

  • 목록 순서대로 poll 하며, 같은 poll 에서 둘 이상이 끝나면 낮은 index 가 이깁니다.
  • 하나가 끝나면, 이미 시작한 나머지는 모두 취소됩니다. 취소된 상태로 한 번 더 poll 되어 finally / using 이 실행됩니다 (2.18.4). 한 번도 poll 되지 않은 future 는 실행되지 않습니다.
  • 진 future 들이 만든 깨우기 등록 (timer, parking) 은 모두 풀립니다.
  • 기다리는 task 가 취소되면 자식 future 들도 취소됩니다.
  • 빈 목록이면 panic 입니다.
  • future 들은 한 타입이어야 합니다. 결과 타입이 다르면 감쌉니다: async { Msg::A(await a) }.

select 와 sleep 으로 timeout 을 만들 수도 있습니다.

enum Ev { Line(str), Timeout }

let (_, ev) = await task::select([
    async { Ev::Line(await read_line()) },
    async { await task::sleep(Duration::sec(5)); Ev::Timeout },
]);

TaskId

task::id() 는 실행 중인 task 의 id 이며, task 밖에서 부르면 panic 합니다 (try_id() 는 none). Task::id() 는 그 task 의 id 입니다. id 는 프로세스 전역 카운터에서 나오며 다시 쓰이지 않습니다.

2.18.2. Task<T> 타입

Task<T> 타입은 다음과 같이 정의됩니다.

struct Task<T> {
    // 내부: GC heap 의 coroutine frame + 결과 슬롯
}

impl<T> Task<T> {
    fn cancel(self);                      // 취소 요청 - 다음 await 에서 끝남 (2.18.4)
    fn is_cancelled(self) -> bool;
    fn is_finished(self) -> bool;
    fn join(self) -> Join<T>;             // try await t.join() - 결과 없이 취소됐으면 CancellationError
    fn id(self) -> TaskId;
}

struct TaskId { /* u64 */ }
impl TaskId {
    fn as_u64(self) -> u64;
}
// TaskId: PartialEq / Eq / Hash / Display / Debug

struct CancellationError {}

impl<T> Future for Task<T> {              // `await t` - 완료까지 suspend
    type Output = T;
    ...
}

IntoFuture 통합

Task<T> 는 Future<Output = T> 를 구현하므로 await t 로 기다릴 수 있고, 2.20 future 의 IntoFuture blanket impl 에도 포함되어 Thread 와 동일한 방식으로 await 될 수 있습니다. IntoFuture 트레잇과 lang slot future_trait 의 권위 있는 정의는 2.20 future 에 있으며, 본 절은 그 정의를 그대로 따릅니다.

2.18.3. 스케줄러 모델 - work-stealing

프로세스에는 executor 가 하나 있습니다. 처음 쓸 때 시작하며, CPU 수 (cpu_count()) 만큼의 worker thread, blocking lane, I/O driver 로 이루어집니다. block_on 은 부른 스레드를 guest worker 로 이 executor 에 붙여 f 를 끝까지 실행합니다. block_on 이 돌아올 때 실행 중인 task 는 계속 실행됩니다. 다만 block_on 은 그 스레드에 고정된 spawn_local task 들도 끝난 뒤에야 돌아옵니다. 그 task 들은 다른 스레드로 옮겨 갈 수 없기 때문입니다. par::* 와 spawn_blocking 도 같은 executor 를 씁니다.

spawn 한 task 는 어느 worker 에서든 실행될 수 있으므로 Future + Send 이고 결과도 Send 여야 합니다. spawn_local 의 task 는 그것을 만든 worker 에 고정됩니다. 다만 block_on 스레드가 기다리는 동안 맡아 poll 한, 자기 것이 아닌 task (spawn 한 task) 에서 부른 spawn_local 은 pool worker 에 고정됩니다. 그래서 block_on 은 자기 task 들의 local task 만 기다립니다. 이런 poll 한 번에서 만든 local task 들은 모두 같은 pool worker 로 가고, 그 poll 이 끝난 뒤에야 시작합니다. 원래 그 task 를 실행했을 worker 에서와 같습니다.

스케줄러 골격은 다음과 같이 구성됩니다.

Worker 0 ─┐      Worker 1 ─┐    ...    Worker N-1 ─┐
 Local     │      Local     │            Local      │
 deque     │      deque     │            deque      │
 (Chase-Lev)      (Chase-Lev)            (Chase-Lev)│
           └─ Global Injector Queue ──────┘
           └─ I/O Driver (포터블)   ──────┘
           └─ Timer (heap)  ───────────────┘
           └─ Parked Worker Stack   ──────┘

이 구조의 주요 특성을 정리하면, 우선 로컬 큐는 lock-free Chase-Lev deque 이며, 소유 worker 와 도둑 모두 가장 오래된 task 부터 꺼냅니다 (스스로를 다시 깨우는 task 가 다른 task 를 굶기지 않도록, refs-lang 7.6.2). 바쁜 worker 는 task 32 개마다 기한이 지난 timer 를 처리합니다. I/O Driver 는 프로세스당 reactor thread 하나가 epoll / kqueue 를 edge-triggered 로 기다리며 (io_uring / IOCP 는 향후, refs-lang 7.7), Timer 구조는 구현이 정합니다 (binary heap 또는 timing wheel, refs-lang 7.7.4).

표준 라이브러리 측의 권위는 API 시그니처에 한정되며, 알고리듬 본문은 spec 외부의 runtime 및 표준 라이브러리 body 영역에 속한다는 점에 유의해야 합니다.

2.18.4. 협력적 취소 - Task::cancel

협력적 취소는 다음과 같이 사용합니다.

let t = task::spawn(async { long_running() });
// ... 다른 일 ...
t.cancel();              // 협력적 신호 - task 가 cancellation point 에서 종료

cancel() 은 취소를 요청 합니다. 규칙은 다음과 같습니다 (refs-lang 7.7.9).

  • 모든 await 가 cancellation point 입니다. 취소된 task 는 다음 await 에서 return 처럼 본문을 떠나며, 지나는 finally 와 using 정리가 안쪽부터 실행됩니다.
  • 취소는 task 안에서 catch 할 수 있는 에러가 아닙니다. task 안으로 CancellationError 가 던져지지 않습니다.
  • blocking syscall 은 cancellation point 가 아닙니다. 그 호출이 끝난 뒤 다음 await 에서 떠납니다.
  • 소켓 I/O (*_async), readable / writable, sleep / timeout, lock_async, recv_async / send_async 에서 기다리던 task 는 취소되면 깨어나고 다음 await 에서 떠납니다.
  • task 가 끝나면 그 task 의 깨우기 등록 (I/O readiness, timer, park 한 대기) 은 모두 풀립니다.
  • spawn_blocking 과 그 위의 fs::*_async 를 기다리던 중 취소되면, 이미 시작된 작업은 백그라운드에서 끝까지 실행되고 결과만 버려집니다. 쓰기 같은 부수 효과는 일어납니다. 그 사이에 끝난 결과도 버립니다. 취소된 poll 은 Pending 을 돌려줍니다.
  • 기다리던 future 가 값을 넘겨준 (메시지, 잠금 guard) poll 은 그 값을 지킵니다. 취소 검사는 "기다린 코루틴이 취소 때문에 떠났는가" 를 보며, 값은 남고 다음 await 에서 떠납니다 (refs-lang 7.7.9).
  • 마지막 cancellation point 를 지난 뒤 취소된 task 는 정상적으로 끝나고 결과를 가집니다. await t / join 은 그 결과를 돌려줍니다.
  • CancellationError 는 결과 없이 끝난 task 를 기다리는 쪽이 받습니다. 그런 task 를 try await t.join() 으로 기다리면 CancellationError 가 던져지고, await t 로 기다리면 panic 입니다. 
let t = task::spawn(async {
    using log = open_log();                      // 취소되면 여기서 drop
    loop {
        await task::sleep(Duration::sec(1));     // cancellation point
        tick(log);
    }
});
t.cancel();

let r = try await t.join() catch _: CancellationError { fallback() };

2.18.5. thread::* - 1:1 OS thread

별도 모듈인 thread 는 M:N 모델이 아니라 OS thread 를 직접 spawn 합니다.

use rok::std::thread;

let t = thread::spawn { os_level_work() };
let r = t.join();              // OS-level 동기 대기

두 모델 사이의 차이를 비교하면 다음 표와 같습니다.

측면task::spawnthread::spawn
모델M:N (사용자 공간 스케줄러)1:1 (OS thread)
비용가벼움 (~1KB frame)무거움 (~MB stack)
협력적 취소✓✗ (OS-level kill 필요)
async 통합✓△ (joiner 만 await)
사용처일반 동시성 (default)CPU-bound, blocking I/O

기본적으로는 task::spawn 을 사용하면 되고, thread::spawn 은 명시적으로 OS thread 가 요구되는 상황에서만 선택하면 됩니다.

2.18.6. Duration / Instant

시간 표현을 위한 Duration 과 Instant 는 다음과 같이 정의됩니다.

struct Duration { /* ns 단위 i64 */ }
struct Instant { /* monotonic clock */ }

impl Duration {
    fn ns(n: i64) -> Duration;
    fn us(n: i64) -> Duration;
    fn ms(n: i64) -> Duration;
    fn sec(n: i64) -> Duration;
    fn min(n: i64) -> Duration;
    fn hours(n: i64) -> Duration;
}

impl Instant {
    fn now() -> Instant;
    fn elapsed(self) -> Duration;
}

이를 활용하여 await task::sleep(Duration::ms(100)) 또는 await task::sleep_until(t) 와 같이 호출하면 정확한 monotonic time 을 보장받을 수 있습니다. 두 타입은 rok::core::task::time 에 있으며 (2.18.7), rok::std::time 이 이를 다시 내보냅니다.

2.18.7. 모듈 매핑

각 항목이 자리 잡는 모듈은 다음 표와 같습니다.

항목모듈동작
task::spawnrok::core::task + scheduler runtimescheduler push
task::racerok::core::taskfirst-completed selector
task::joinrok::core::taskjoin all
task::timeoutrok::core::taskrace + timer
Task<T>rok::core::taskcoroutine frame ref
Duration/Instantrok::core::task::timei64 ns + monotonic clock
Future (lang slot)rok::core::futurepoll-based
IntoFuturerok::core::futureawait 진입점

GitHub에서 이 페이지 고치기