diff --git a/README.md b/README.md index 3f42b01..e5e6e99 100644 --- a/README.md +++ b/README.md @@ -680,6 +680,18 @@ conversion can differ slightly; validate a short Decord/PyNvVideoCodec pair before extracting every session, and do not mix decoder backends across train/test splits. +For CPU Decord, the direct reader intentionally indexes each clip and releases +that temporary reader, then keeps an LRU of at most two live readers during +extraction. Each reader defaults to one decode thread. These limits avoid one +worker retaining a thread pool for every one-minute EasyCom clip and can be +overridden with `--decoder-threads` and `--max-open-readers`. The recommended +full-run settings are: + +```text +--decoder decord --decoder-threads 1 --max-open-readers 2 +--frame-cache-size 64 +``` + Matching `states.shape[0]` to the target is necessary but not sufficient. The direct reader also validates EasyCom's 20-Hz source inventory, source PTS monotonicity, dense duration, crop bounds, audio sample count, and causal diff --git a/profile_streaming_bottleneck_gpu7.py b/profile_streaming_bottleneck_gpu7.py new file mode 100644 index 0000000..39c6bf1 --- /dev/null +++ b/profile_streaming_bottleneck_gpu7.py @@ -0,0 +1,547 @@ +#!/usr/bin/env python3 +"""Profile direct EasyCom Qwen2.5-Omni extraction on one physical GPU. + +This is a diagnostic entry point for ``streaming_turn_model_directly.py``. +It accepts the same extraction arguments plus three profiler-only options: + +``--gpu-id`` + Physical host GPU ID. It defaults to 7 and is exposed to the extraction + process as ``cuda:0``. +``--profile-log-dir`` + Directory for the console log, GPU telemetry CSV, and JSON summary. +``--gpu-sample-seconds`` + ``nvidia-smi`` sampling period in seconds. + +The profiler synchronizes CUDA around individual model stages so their times +are attributable. That synchronization perturbs throughput; use this script +to find bottlenecks, not to produce the final runtime measurement. +""" + +from __future__ import annotations + +import csv +import functools +import json +import os +import re +import resource +import statistics +import subprocess +import sys +import time +import types +from collections import defaultdict +from datetime import datetime +from pathlib import Path +from typing import Any, Callable, TextIO, TypeVar + + +def _pop_cli_option(name: str, default: str) -> str: + """Remove one profiler-only option before the normal parser sees it.""" + + for index, argument in enumerate(sys.argv[1:], start=1): + if argument == name: + if index + 1 >= len(sys.argv): + raise SystemExit(f"{name} requires a value") + value = sys.argv[index + 1] + del sys.argv[index : index + 2] + return value + prefix = f"{name}=" + if argument.startswith(prefix): + value = argument[len(prefix) :] + del sys.argv[index] + return value + return default + + +# CUDA visibility must be fixed before importing torch or the Qwen modules. +PHYSICAL_GPU_ID = int(_pop_cli_option("--gpu-id", "7")) +PROFILE_LOG_DIR = Path( + _pop_cli_option("--profile-log-dir", "logs/bottleneck_gpu7") +).expanduser() +GPU_SAMPLE_SECONDS = float(_pop_cli_option("--gpu-sample-seconds", "1")) +if PHYSICAL_GPU_ID < 0: + raise SystemExit("--gpu-id must be >= 0") +if GPU_SAMPLE_SECONDS <= 0: + raise SystemExit("--gpu-sample-seconds must be > 0") +os.environ["CUDA_VISIBLE_DEVICES"] = str(PHYSICAL_GPU_ID) + +import torch # noqa: E402 (must follow CUDA_VISIBLE_DEVICES) + +import streaming_turn_model as core # noqa: E402 +import streaming_turn_model_directly as direct # noqa: E402 + + +T = TypeVar("T") + + +class TeeStream: + """Write stdout/stderr to both the terminal and a durable log.""" + + def __init__(self, terminal: TextIO, log_file: TextIO) -> None: + self.terminal = terminal + self.log_file = log_file + self.encoding = getattr(terminal, "encoding", "utf-8") + + def write(self, data: str) -> int: + terminal_count = self.terminal.write(data) + self.log_file.write(data) + self.flush() + return terminal_count + + def flush(self) -> None: + self.terminal.flush() + self.log_file.flush() + + def isatty(self) -> bool: + return self.terminal.isatty() + + +def _percentile(values: list[float], fraction: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + index = round((len(ordered) - 1) * fraction) + return float(ordered[index]) + + +def _sync_cuda() -> None: + if torch.cuda.is_available() and torch.cuda.is_initialized(): + torch.cuda.synchronize() + + +class StageProfiler: + """Accumulate inclusive wall times for named extraction stages.""" + + def __init__(self) -> None: + self.durations: dict[str, list[float]] = defaultdict(list) + self.metadata: dict[str, Any] = {} + + def measure( + self, + stage: str, + function: Callable[..., T], + *args: Any, + cuda_sync: bool = False, + **kwargs: Any, + ) -> T: + if cuda_sync: + _sync_cuda() + started = time.perf_counter() + try: + return function(*args, **kwargs) + finally: + if cuda_sync: + _sync_cuda() + self.durations[stage].append(time.perf_counter() - started) + + def rows(self) -> list[dict[str, Any]]: + extract_time = sum(self.durations.get("extract_total", [])) + rows = [] + for stage, values in self.durations.items(): + total = sum(values) + rows.append( + { + "stage": stage, + "calls": len(values), + "total_seconds": total, + "average_ms": total / len(values) * 1000.0, + "p50_ms": statistics.median(values) * 1000.0, + "p95_ms": _percentile(values, 0.95) * 1000.0, + "maximum_ms": max(values) * 1000.0, + "percent_of_extract": ( + total / extract_time * 100.0 if extract_time else None + ), + } + ) + return sorted(rows, key=lambda row: row["total_seconds"], reverse=True) + + +PROFILER = StageProfiler() + + +def _patch_function( + owner: Any, + name: str, + stage: str, + *, + cuda_sync: bool = False, +) -> Callable[..., Any]: + original = getattr(owner, name) + + @functools.wraps(original) + def wrapped(*args: Any, **kwargs: Any) -> Any: + return PROFILER.measure( + stage, + original, + *args, + cuda_sync=cuda_sync, + **kwargs, + ) + + setattr(owner, name, wrapped) + return original + + +def _patch_bound_method( + owner: Any, + name: str, + stage: str, + *, + cuda_sync: bool, +) -> None: + original = getattr(owner, name) + + @functools.wraps(original) + def wrapped(_self: Any, *args: Any, **kwargs: Any) -> Any: + return PROFILER.measure( + stage, + original, + *args, + cuda_sync=cuda_sync, + **kwargs, + ) + + setattr(owner, name, types.MethodType(wrapped, owner)) + + +def _install_profiling_hooks() -> None: + # One-time input setup. + _patch_function(direct, "_decode_audio_exact", "audio_decode") + _patch_function(direct, "_timestamp_array", "video_timestamp_index") + _patch_function(direct.CausalSourceAV, "__init__", "media_reader_init") + + # Per-state CPU media and processor work. + _patch_function(direct.CausalSourceAV, "frames", "media_frames") + _patch_function(direct.CausalSourceAV, "audio_slice", "media_audio_slice") + _patch_function(core, "_prepare_conversation", "processor") + + # Per-state GPU/cache work. Leaf GPU calls synchronize for attribution. + _patch_function(core, "clone_cache", "cache_clone", cuda_sync=True) + _patch_function(core, "_multimodal_embeddings", "multimodal_merge_total") + _patch_function(core, "_append_embeddings", "append_total") + _patch_function(core, "_read_open_state", "readout_total") + _patch_function( + core.OpenStreamReplayExtractor, + "_prepare_pieces", + "prepare_piece_total", + ) + _patch_function( + core.OpenStreamReplayExtractor, + "_cache_from_intervals", + "rolling_cache_rebuild", + ) + _patch_function(torch, "save", "output_save") + + original_extract = core.OpenStreamReplayExtractor.extract + + @functools.wraps(original_extract) + def extract_wrapped(self: Any, *args: Any, **kwargs: Any) -> Any: + result = PROFILER.measure( + "extract_total", + original_extract, + self, + *args, + cuda_sync=True, + **kwargs, + ) + PROFILER.metadata["num_states"] = int(len(result.times)) + return result + + core.OpenStreamReplayExtractor.extract = extract_wrapped + + original_load = core.load_qwen_omni + + @functools.wraps(original_load) + def load_wrapped(*args: Any, **kwargs: Any) -> Any: + model, processor = PROFILER.measure( + "model_load", + original_load, + *args, + cuda_sync=True, + **kwargs, + ) + + # These are the mutually exclusive heavy GPU leaves inside one state. + _patch_bound_method( + model, "get_audio_features", "audio_tower", cuda_sync=True + ) + _patch_bound_method( + model, "get_video_features", "video_tower", cuda_sync=True + ) + _patch_bound_method(model, "get_rope_index", "tmrope", cuda_sync=True) + _patch_bound_method( + model.model, "forward", "thinker_forward", cuda_sync=True + ) + torch.cuda.reset_peak_memory_stats() + return model, processor + + core.load_qwen_omni = load_wrapped + + +def _start_gpu_monitor(path: Path) -> tuple[subprocess.Popen[str] | None, TextIO | None]: + command = [ + "nvidia-smi", + "-i", + str(PHYSICAL_GPU_ID), + "--query-gpu=timestamp,index,name,utilization.gpu,utilization.memory," + "memory.used,memory.total,power.draw,temperature.gpu", + "--format=csv,nounits", + "-l", + f"{GPU_SAMPLE_SECONDS:g}", + ] + try: + output = path.open("w", encoding="utf-8", buffering=1) + process = subprocess.Popen( + command, + stdout=output, + stderr=subprocess.STDOUT, + text=True, + ) + return process, output + except (FileNotFoundError, OSError) as exc: + print(f"WARNING: could not start nvidia-smi monitor: {exc}", flush=True) + return None, None + + +def _stop_gpu_monitor( + process: subprocess.Popen[str] | None, + output: TextIO | None, +) -> None: + if process is not None and process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + if output is not None: + output.close() + + +def _numeric(value: str | None) -> float | None: + if not value: + return None + match = re.search(r"[-+]?[0-9]*\.?[0-9]+", value) + return float(match.group(0)) if match else None + + +def _gpu_telemetry_summary(path: Path) -> dict[str, Any]: + if not path.is_file(): + return {"samples": 0} + metrics: dict[str, list[float]] = defaultdict(list) + with path.open("r", encoding="utf-8", errors="replace") as handle: + reader = csv.DictReader(handle) + for row in reader: + for key, value in row.items(): + normalized = key.strip().lower() + number = _numeric(value) + if number is None: + continue + if normalized.startswith("utilization.gpu"): + metrics["gpu_util_percent"].append(number) + elif normalized.startswith("utilization.memory"): + metrics["memory_util_percent"].append(number) + elif normalized.startswith("memory.used"): + metrics["memory_used_mib"].append(number) + elif normalized.startswith("power.draw"): + metrics["power_watts"].append(number) + elif normalized.startswith("temperature.gpu"): + metrics["temperature_c"].append(number) + result: dict[str, Any] = { + "samples": max((len(values) for values in metrics.values()), default=0) + } + for name, values in metrics.items(): + result[name] = { + "mean": statistics.fmean(values), + "p50": statistics.median(values), + "p95": _percentile(values, 0.95), + "max": max(values), + } + return result + + +def _print_summary( + *, + rows: list[dict[str, Any]], + gpu_summary: dict[str, Any], + overall_seconds: float, +) -> None: + extract_seconds = sum(PROFILER.durations.get("extract_total", [])) + num_states = int(PROFILER.metadata.get("num_states", 0)) + print("\n=== Qwen Omni bottleneck profile (inclusive timings) ===") + print( + "Nested rows intentionally overlap; compare leaf rows such as " + "processor/audio_tower/video_tower/thinker_forward/cache_clone." + ) + print( + f"overall={overall_seconds:.3f}s extract={extract_seconds:.3f}s " + f"states={num_states} " + f"throughput={(num_states / extract_seconds if extract_seconds else 0):.3f} state/s" + ) + print( + f"{'stage':<27} {'calls':>7} {'total_s':>10} {'avg_ms':>10} " + f"{'p95_ms':>10} {'extract%':>10}" + ) + for row in rows: + percent = row["percent_of_extract"] + percent_text = "-" if percent is None else f"{percent:.1f}" + print( + f"{row['stage']:<27} {row['calls']:>7d} " + f"{row['total_seconds']:>10.3f} {row['average_ms']:>10.3f} " + f"{row['p95_ms']:>10.3f} {percent_text:>10}" + ) + + gpu_util = gpu_summary.get("gpu_util_percent") + memory_used = gpu_summary.get("memory_used_mib") + if gpu_util: + print( + "GPU telemetry: " + f"samples={gpu_summary['samples']} " + f"util mean/p95/max={gpu_util['mean']:.1f}/" + f"{gpu_util['p95']:.1f}/{gpu_util['max']:.1f}%" + ) + if memory_used: + print( + "GPU memory MiB mean/p95/max=" + f"{memory_used['mean']:.0f}/{memory_used['p95']:.0f}/" + f"{memory_used['max']:.0f}" + ) + + leaf_stages = ( + "media_frames", + "processor", + "audio_tower", + "video_tower", + "tmrope", + "thinker_forward", + "cache_clone", + ) + totals = { + stage: sum(PROFILER.durations.get(stage, [])) for stage in leaf_stages + } + bottleneck = max(totals, key=totals.get) + advice = { + "media_frames": "bulk-decode/cache the selected 2-FPS resized frames", + "processor": "cache processor-ready AV tensors for every causal prefix", + "audio_tower": "microbatch the ten 0.2-s prefixes in each 2-s block", + "video_tower": "cache unique 2-FPS visual inputs and batch equal shapes", + "tmrope": "cache position IDs for the ten repeated prefix shapes", + "thinker_forward": "microbatch independent streams/prefix branches", + "cache_clone": "replace full KV clones with append-and-crop branch reuse", + } + print( + f"Primary measured leaf bottleneck: {bottleneck} " + f"({totals[bottleneck]:.3f}s). Recommended next optimization: " + f"{advice[bottleneck]}." + ) + + +def main() -> None: + PROFILE_LOG_DIR.mkdir(parents=True, exist_ok=True) + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + console_log = PROFILE_LOG_DIR / f"profile_gpu{PHYSICAL_GPU_ID}_{stamp}.log" + gpu_log = PROFILE_LOG_DIR / f"gpu_gpu{PHYSICAL_GPU_ID}_{stamp}.csv" + json_log = PROFILE_LOG_DIR / f"profile_gpu{PHYSICAL_GPU_ID}_{stamp}.json" + + original_stdout = sys.stdout + original_stderr = sys.stderr + status = "ok" + error_text: str | None = None + started = time.perf_counter() + usage_started = resource.getrusage(resource.RUSAGE_SELF) + + with console_log.open("w", encoding="utf-8", buffering=1) as log_handle: + sys.stdout = TeeStream(original_stdout, log_handle) + sys.stderr = TeeStream(original_stderr, log_handle) + monitor: subprocess.Popen[str] | None = None + monitor_output: TextIO | None = None + try: + print(f"Physical GPU: {PHYSICAL_GPU_ID} -> process cuda:0") + print(f"CUDA_VISIBLE_DEVICES={os.environ['CUDA_VISIBLE_DEVICES']}") + print(f"Console log: {console_log.resolve()}") + print(f"GPU log: {gpu_log.resolve()}") + print(f"JSON summary: {json_log.resolve()}") + print("Extraction argv:", " ".join(sys.argv[1:])) + print( + "NOTE: CUDA synchronization is enabled for attribution; " + "this is not a production throughput benchmark." + ) + if torch.cuda.is_available(): + print(f"Visible GPU: {torch.cuda.get_device_name(0)}") + print(f"PyTorch: {torch.__version__}, CUDA: {torch.version.cuda}") + else: + raise RuntimeError("CUDA is not available after GPU selection") + + monitor, monitor_output = _start_gpu_monitor(gpu_log) + _install_profiling_hooks() + direct.main() + except BaseException as exc: + status = "error" + error_text = f"{type(exc).__name__}: {exc}" + raise + finally: + _sync_cuda() + _stop_gpu_monitor(monitor, monitor_output) + overall_seconds = time.perf_counter() - started + usage_finished = resource.getrusage(resource.RUSAGE_SELF) + gpu_summary = _gpu_telemetry_summary(gpu_log) + rows = PROFILER.rows() + PROFILER.metadata.update( + { + "status": status, + "error": error_text, + "physical_gpu_id": PHYSICAL_GPU_ID, + "visible_cuda_device": 0, + "gpu_name": torch.cuda.get_device_name(0), + "torch_version": torch.__version__, + "torch_cuda_version": torch.version.cuda, + "overall_seconds": overall_seconds, + "cpu_user_seconds": ( + usage_finished.ru_utime - usage_started.ru_utime + ), + "cpu_system_seconds": ( + usage_finished.ru_stime - usage_started.ru_stime + ), + "max_rss_kib": usage_finished.ru_maxrss, + "peak_cuda_allocated_mib": ( + torch.cuda.max_memory_allocated() / 1024**2 + ), + "peak_cuda_reserved_mib": ( + torch.cuda.max_memory_reserved() / 1024**2 + ), + "console_log": str(console_log.resolve()), + "gpu_log": str(gpu_log.resolve()), + } + ) + _print_summary( + rows=rows, + gpu_summary=gpu_summary, + overall_seconds=overall_seconds, + ) + print( + "Process resources: " + f"user_cpu={PROFILER.metadata['cpu_user_seconds']:.2f}s " + f"system_cpu={PROFILER.metadata['cpu_system_seconds']:.2f}s " + f"max_rss={PROFILER.metadata['max_rss_kib'] / 1024:.1f}MiB " + f"peak_cuda_allocated=" + f"{PROFILER.metadata['peak_cuda_allocated_mib']:.1f}MiB" + ) + with json_log.open("w", encoding="utf-8") as handle: + json.dump( + { + "metadata": PROFILER.metadata, + "gpu_telemetry": gpu_summary, + "stages": rows, + "argv": sys.argv[1:], + }, + handle, + indent=2, + ) + print(f"Saved profile summary to {json_log.resolve()}") + sys.stdout = original_stdout + sys.stderr = original_stderr + + +if __name__ == "__main__": + main() diff --git a/streaming_turn_model_directly.py b/streaming_turn_model_directly.py index c108417..6463a53 100644 --- a/streaming_turn_model_directly.py +++ b/streaming_turn_model_directly.py @@ -230,6 +230,8 @@ def _decode_audio_exact(path: Path, duration: float) -> np.ndarray: "-nostdin", "-v", "error", + "-threads", + "1", "-i", str(path), "-map", @@ -275,6 +277,8 @@ class CausalSourceAV: frame_cache_size: int = 8, decoder_backend: str = "decord", decoder_gpu_id: int = 0, + decoder_threads: int = 1, + max_open_readers: int = 2, ) -> None: if not specs: raise ValueError("At least one source clip is required") @@ -288,10 +292,15 @@ class CausalSourceAV: raise ValueError(f"Unsupported decoder backend: {decoder_backend}") if decoder_gpu_id < 0: raise ValueError("decoder_gpu_id must be non-negative") + if decoder_threads <= 0: + raise ValueError("decoder_threads must be positive") + if max_open_readers <= 0: + raise ValueError("max_open_readers must be positive") if shutil.which("ffmpeg") is None: raise RuntimeError("ffmpeg is required to decode source audio") decord = None + self._decord = None self._nvc = None if decoder_backend == "decord": try: @@ -300,6 +309,7 @@ class CausalSourceAV: raise RuntimeError( "The decord backend requires qwen-omni-utils[decord]" ) from exc + self._decord = decord else: try: import PyNvVideoCodec as nvc @@ -315,6 +325,11 @@ class CausalSourceAV: self.frame_cache_size = int(frame_cache_size) self.decoder_backend = decoder_backend self.decoder_gpu_id = int(decoder_gpu_id) + self.decoder_threads = int(decoder_threads) + self.max_open_readers = int(max_open_readers) + self._decord_readers: OrderedDict[int, Any] = OrderedDict() + self._decord_reader_opens = 0 + self._decord_reader_evictions = 0 self._nvc_decoder = None self._nvc_source_path: Path | None = None self._nvc_decoder_reconfigurations = 0 @@ -340,17 +355,23 @@ class CausalSourceAV: ) if decoder_backend == "decord": assert decord is not None - reader = decord.VideoReader(str(path)) - if not len(reader): + index_reader = decord.VideoReader( + str(path), num_threads=self.decoder_threads + ) + if not len(index_reader): raise RuntimeError(f"No video frames decoded from {path}") - starts, ends = _timestamp_array(reader) - source_fps = float(reader.get_avg_fps()) - first_shape = tuple(int(value) for value in reader[0].shape) + starts, ends = _timestamp_array(index_reader) + source_fps = float(index_reader.get_avg_fps()) + first_shape = tuple(int(value) for value in index_reader[0].shape) if len(first_shape) != 3: raise RuntimeError( f"Unexpected decoded frame shape {first_shape}: {path}" ) height, width = first_shape[:2] + # VideoReader.__del__ releases its native handle. Do not retain + # one reader/thread pool for every one-minute EasyCom clip. + del index_reader + reader = None else: reader = None starts, ends, source_fps, width, height = _ffprobe_video_index(path) @@ -487,9 +508,8 @@ class CausalSourceAV: clip = self.clips[clip_index] if self.decoder_backend == "decord": - if clip.reader is None: - raise RuntimeError(f"Missing decord reader for {clip.path}") - array = clip.reader[frame_index].asnumpy() + reader = self._decord_reader_for_clip(clip_index) + array = reader[frame_index].asnumpy() else: decoder = self._pynv_decoder_for_clip(clip) decoded_frame = decoder[frame_index] @@ -516,6 +536,35 @@ class CausalSourceAV: self._frame_cache.popitem(last=False) return image.copy() + def _decord_reader_for_clip(self, clip_index: int) -> Any: + """Return an LRU-managed CPU reader instead of retaining every clip.""" + + reader = self._decord_readers.get(clip_index) + if reader is not None: + self._decord_readers.move_to_end(clip_index) + return reader + if self._decord is None: + raise RuntimeError("Decord was not initialized") + + clip = self.clips[clip_index] + reader = self._decord.VideoReader( + str(clip.path), num_threads=self.decoder_threads + ) + if len(reader) != clip.num_frames: + raise RuntimeError( + f"Decord frame count changed for {clip.path}: " + f"runtime={len(reader)}, indexed={clip.num_frames}" + ) + self._decord_readers[clip_index] = reader + self._decord_readers.move_to_end(clip_index) + self._decord_reader_opens += 1 + + while len(self._decord_readers) > self.max_open_readers: + _, evicted_reader = self._decord_readers.popitem(last=False) + del evicted_reader + self._decord_reader_evictions += 1 + return reader + def _pynv_decoder_for_clip(self, clip: IndexedSourceClip) -> Any: """Return one reusable NVDEC decoder configured for ``clip``.""" @@ -613,6 +662,12 @@ class CausalSourceAV: "decoder_gpu_id": ( self.decoder_gpu_id if self.decoder_backend == "pynvcodec" else None ), + "decoder_threads": ( + self.decoder_threads if self.decoder_backend == "decord" else None + ), + "max_open_readers": ( + self.max_open_readers if self.decoder_backend == "decord" else None + ), "decoder_version": ( getattr(self._nvc, "__version__", None) if self.decoder_backend == "pynvcodec" @@ -650,6 +705,8 @@ class CausalSourceAV: "short_interval_fallbacks": self._short_interval_fallbacks, "tail_frame_reuses": self._tail_frame_reuses, "decoder_reconfigurations": self._nvc_decoder_reconfigurations, + "decord_reader_opens": self._decord_reader_opens, + "decord_reader_evictions": self._decord_reader_evictions, }, } @@ -788,6 +845,18 @@ def parse_args() -> argparse.Namespace: default=0, help="GPU visible to this process used by PyNvVideoCodec", ) + parser.add_argument( + "--decoder-threads", + type=int, + default=1, + help="CPU decode threads per Decord VideoReader", + ) + parser.add_argument( + "--max-open-readers", + type=int, + default=2, + help="Maximum simultaneously open Decord VideoReaders per process", + ) parser.add_argument("--verify-every", type=int, default=0) parser.add_argument("--verify-continuous-every", type=int, default=0) parser.add_argument("--no-progress", action="store_true") @@ -923,6 +992,8 @@ def main() -> None: frame_cache_size=args.frame_cache_size, decoder_backend=args.decoder, decoder_gpu_id=args.decoder_gpu_id, + decoder_threads=args.decoder_threads, + max_open_readers=args.max_open_readers, ) is_easycom = ( args.easycom_session_dir is not None