멈춰 버린 읽기에 시간 제한 걸기

우열 없음 async

요구사항

프로브 값 네 개를 차례로 읽습니다; 세 번째 읽기는 500 ms 동안 멈춥니다. 모든 읽기에 150 ms 예산을 주세요: 제때 도착한 측정값들을 출력하고, 멈춘 읽기에는 reading timed out을 출력한 뒤 중단합니다 — 네 번째 읽기는 보고되면 안 됩니다. 멈춤은 코드에 결정적으로 주입되어 있으며, 두 버전 모두 예상 출력 아래에 표시된 줄들을 출력해야 합니다.

예상 출력

reading: 21.5
reading: 21.7
reading timed out

나란히 보기

RxDart

import 'package:rxdart/rxdart.dart';

/// Four probe reads; the third stalls far past the 150 ms budget.
const readMs = [30, 40, 500, 30];
const probeValues = [21.5, 21.7, 22.4, 21.9];

Future<double> readProbe(int i) async {
  await Future<void>.delayed(Duration(milliseconds: readMs[i]));
  return probeValues[i];
}

Future<void> main() async {
  final lines = await Stream.fromIterable([0, 1, 2, 3])
      // The budget is per read, so bound the Future itself. (The stream-
      // level .timeout operator measures a different thing: gaps between
      // events on the push side.)
      .asyncMap((i) =>
          readProbe(i).timeout(const Duration(milliseconds: 150)))
      .map((v) => 'reading: ${v.toStringAsFixed(1)}')
      // Replace the timeout error with a report line…
      .onErrorReturnWith((_, _) => 'reading timed out')
      // …and stop there: the paused source would otherwise resume pushing
      // readings once the stalled read finally lands.
      .takeWhileInclusive((line) => line != 'reading timed out')
      .toList();

lines.forEach(print);
}

FxDart

차이가 나는 이유

읽기별 예산 자체는 양쪽 다 쉽습니다 — RxDart 패널은 asyncMap 안의 Future에 제한을 걸고, FxDart의 timeout은 풀에 제한을 겁니다 — 그래서 흥미로운 차이는 각 모델의 스트림 수준 동명 연산자가 무엇을 재는가입니다. Stream.timeout은 이벤트 사이의 간격을 지켜봅니다: 값이 언제 도착할지는 생산자가 정하므로, "너무 느리다"는 "요즘 아무것도 도착하지 않았다"라는 뜻일 수밖에 없습니다. FxDart의 timeout은 요구부터 항목까지의 시간에 제한을 겁니다: 소비자가 묻고, 시계는 물음에서 답까지 돕니다. 이 유한하고 순차적인 과제에서는 둘이 일치하겠지만 — 이들은 진짜로 다른 양입니다: 요구가 없는 풀 파이프라인에는 잴 간격이 없고, 푸시 스트림은 누구의 물음에도 답을 빚지지 않습니다.

그다음 양쪽 모두 "그리고 중단" 조항을 맞추기 위한 진짜 굴곡이 하나씩 필요합니다. 푸시 쪽에서는 멈춘 소스가 여전히 저 바깥에 있고, 느린 읽기가 마침내 도착하면 다시 측정값을 밀어 넣기 시작할 것입니다 — 그래서 onErrorReturnWith가 오류를 보고 줄로 바꾼 뒤, takeWhileInclusive가 스트림을 끝내고 구독을 취소합니다. 풀 쪽에서 중단은 공짜입니다 — TimeoutException이 그냥 루프를 빠져나가고 아무것도 다시 풀지 않습니다 — 다만 멈춤 이전의 측정값들을 지키려면, 오류를 던지는 순간 그것들을 버렸을 toList 대신 each로 수집해야 합니다.

무승부입니다: 양쪽 다 연산자 하나에 굴곡 하나씩이고, 그 굴곡들은 각 모델의 본성을 거울처럼 비춥니다.

벤치마크

Apple M1 Max, RAM 32 GB · Dart 3.12.2 (AOT 컴파일) · 2026-08-18

비동기 예제입니다. 대표 스케일이 1,000,000이 아니라 N = 10,000인 이유 — 원소 하나마다 양쪽 모두 이벤트 루프를 한 바퀴 돌아야 하므로, 실제 await 백만 번은 파이프라인이 아니라 Dart 이벤트 루프를 몇 분씩 재게 됩니다. 지연은 0으로 두고 예제의 동시성 제한은 그대로 유지합니다. 막대가 비교하는 것은 파이프라인 기계 장치 자체입니다.

N = 100

시간 무승부

RxDart 174 µs

FxDart 145 µs

최대 메모리 무승부

RxDart 16.6 MB

FxDart 17.1 MB

N = 10,000

시간 FxDart 승

RxDart 14.3 ms

FxDart 12.9 ms

최대 메모리 무승부

RxDart 27.4 MB

FxDart 27.0 MB

막대는 사이드별로 새 프로세스에서 반복 측정한 중앙값입니다(작은 N은 타이머 해상도를 위해 배치 처리). 두 사이드가 서로 5% 이내이거나 — 사람이 지각할 수 없는 차이인 0.6ms 이내이면 — 무승부로 칩니다. 상대 차이가 근소한 경우는 최대 5회까지 다시 측정합니다. 앱에서는 어느 막대가 짧든 몇 밀리초 이하의 차이는 사용자에게 보이지 않습니다. 메모리는 프로세스 최대 RSS입니다. Dart VM과 데이터셋은 양쪽이 동일하므로, 두 막대의 차이가 곧 파이프라인 자체가 붙들고 있는 양입니다.