Fix stream write slowdown, benchmark crashes and small tool bugs

benchmarks/charts.py

@@ -39,6 +39,13 @@
 CHARTS = BENCHMARKS / 'charts'
 
 
+def _ratio(value: float, baseline: float, *, lower_is_better: bool) -> float:
+    """How many times better than the baseline the value is; NaN when either is 0."""
+    if value == 0 or baseline == 0:
+        return math.nan
+    return baseline / value if lower_is_better else value / baseline
+
+
 def _relative(figure: Figure) -> tuple[list[str], list[str], list[float]]:
     """The categories, the libraries and the values relative to the baseline's, as the columns of a long table."""
     categories: list[str] = []
@@ -48,7 +55,7 @@ def _relative(figure: Figure) -> tuple[list[str], list[str], list[float]]:
     for category, values in figure.values.items():
         baseline = values.get(BASELINE, math.nan)
         for library, value in values.items():
-            ratio = baseline / value if figure.lower_is_better else value / baseline
+            ratio = _ratio(value, baseline, lower_is_better=figure.lower_is_better)
             if math.isfinite(ratio):
                 categories.append(category)
                 libraries.append(library)

benchmarks/suites/memory.py

@@ -327,14 +327,20 @@ def _per_connection_row(name: str, data: dict[str, Results], libraries: dict[str
     ]
 
 
-def _levels_rows(name: str, result: Results, libraries: dict[str, Hello]) -> list[list[str]]:
+def _median_total_at(runs: list[StateRun], connections: int) -> float:
+    """The median RSS of the runs that reached a count of connections."""
+    totals = [_total(reading) for run in runs for reading in run['levels'] if reading['connections'] == connections]
+    return median(totals)
+
+
+def _levels_rows(name: str, result: Results, libraries: dict[str, Hello], *, levels: list[int]) -> list[list[str]]:
     rows: list[list[str]] = []
     for state in STATES:
         runs = _runs(result, state)
         if len(runs) == 0:
             continue
-        levels = len(runs[0]['levels'])
-        totals = [median([_total(run['levels'][level]) for run in runs]) for level in range(levels)]
+
+        totals = [_median_total_at(runs, level) for level in levels]
         rows.append([markdown.library(name, libraries), state, *(markdown.number(total, 0) for total in totals)])
     return rows
 
@@ -355,8 +361,9 @@ def _growth_row(name: str, result: Results, libraries: dict[str, Hello]) -> list
 
 
 def _levels(data: dict[str, Results]) -> list[int]:
+    """Every count of connections a run reached, as runs stop early at the memory limit."""
     runs = [run for result in data.values() for state in STATES for run in _runs(result, state)]
-    return [] if len(runs) == 0 else [reading['connections'] for reading in runs[0]['levels']]
+    return sorted({reading['connections'] for run in runs for reading in run['levels']})
 
 
 def _per_connection_table(data: dict[str, Results], libraries: dict[str, Hello]) -> list[str]:
@@ -389,8 +396,9 @@ def _retained_table(data: dict[str, Results], libraries: dict[str, Hello]) -> li
 
 
 def _levels_table(data: dict[str, Results], libraries: dict[str, Hello]) -> list[str]:
-    header = ['Library', 'State', *(str(level) for level in _levels(data))]
-    rows = [row for name, result in data.items() for row in _levels_rows(name, result, libraries)]
+    levels = _levels(data)
+    header = ['Library', 'State', *(str(level) for level in levels)]
+    rows = [row for name, result in data.items() for row in _levels_rows(name, result, libraries, levels=levels)]
     return ['### RSS of both processes by connections (MB)', '', *markdown.table(header, rows)]
 
 

examples/openai_live.py

@@ -356,7 +356,8 @@ async def _create_session(self, sdp: str) -> str:
                     json={'session': session, 'transport': transport},
                 )
         except httpx.HTTPError as e:
-            msg = f'Could not reach the OpenAI API: {e or type(e).__name__}'
+            reason = str(e) if str(e) != '' else type(e).__name__
+            msg = f'Could not reach the OpenAI API: {reason}'
             raise CallError(msg) from None
         if response.is_error:
             raise CallError(_describe_http_error(response))

python-webrtc/python/webrtc/streams.py

@@ -1151,6 +1151,7 @@ def __init__(self, stream: WritableStream[_W_contra], sink: object, strategy: _S
         self._strategy = strategy
         # (chunk, future, size) of the writes, then (_CLOSE, future, 0)
         self._queue: collections.deque[tuple[object, asyncio.Future[None], float]] = collections.deque()
+        self._queue_total_size = 0.0
         self._started = False
         self._in_flight = False
 
@@ -1172,7 +1173,16 @@ def error(self, e: object = None) -> None:
             self._stream._start_erroring(_error_of(e))
 
     def _desired_size(self) -> float:
-        return self._strategy.high_water_mark - sum(size for _, _, size in self._queue)
+        return self._strategy.high_water_mark - self._queue_total_size
+
+    def _enqueue(self, chunk: object, future: asyncio.Future[None], size: float) -> None:
+        self._queue.append((chunk, future, size))
+        self._queue_total_size += size
+
+    def _dequeue(self) -> None:
+        _, _, size = self._queue.popleft()
+        # rounding errors could leave it below 0
+        self._queue_total_size = max(0.0, self._queue_total_size - size)
 
     def _start(self) -> None:
         def started() -> None:
@@ -1190,17 +1200,17 @@ def _write(self, chunk: _W_contra, future: asyncio.Future[None]) -> None:
             size = _chunk_size(self._strategy, chunk)
         except Exception as e:
             # queued to be rejected once the stream errors
-            self._queue.append((chunk, future, 0))
+            self._enqueue(chunk, future, 0)
             self.error(e)
             return
 
-        self._queue.append((chunk, future, size))
+        self._enqueue(chunk, future, size)
         # advance first: a sink done right away leaves no backpressure to signal
         self._advance()
         self._stream._update_backpressure()
 
     def _close(self, future: asyncio.Future[None]) -> None:
-        self._queue.append((_CLOSE, future, 0))
+        self._enqueue(_CLOSE, future, 0)
         self._advance()
 
     def _advance(self) -> None:
@@ -1227,7 +1237,7 @@ def failed(error: BaseException) -> None:
 
     def _settle_in_flight(self, future: asyncio.Future[None], error: BaseException | None = None) -> None:
         self._in_flight = False
-        _ = self._queue.popleft()
+        self._dequeue()
         _settle(future, None, error)
 
     def _written(self, future: asyncio.Future[None]) -> None:
@@ -1248,8 +1258,9 @@ def _reject_queue(self, error: BaseException) -> None:
         for _, future, _ in self._queue:
             _fail(future, error)
         self._queue.clear()
+        self._queue_total_size = 0.0
         if in_flight is not None:
-            self._queue.append(in_flight)
+            self._enqueue(*in_flight)
 
 
 class WritableStream(Generic[_W_contra]):

scripts/debug/hunt.py

@@ -839,7 +839,7 @@ def _build_env(build: Build) -> dict[str, str]:
 def build(jobs: Sequence[Job], results: Results) -> None:
     """Builds the editable install and the sanitizer builds the jobs need, once."""
     steps: list[tuple[list[str], dict[str, str]]] = [([sys.executable, '-c', 'import webrtc'], {})]
-    builds = sorted({job.build for job in jobs if job.build is not None}, key=lambda build: build.env.items())
+    builds = sorted({job.build for job in jobs if job.build is not None}, key=lambda build: sorted(build.env.items()))
     steps += [([SANITIZERS, '--build-only'], _build_env(build)) for build in builds]
 
     build_log = results.paths.build_log