From 6baaff6f103514f1768c1977eb4beddf921baa08 Mon Sep 17 00:00:00 2001 From: Zhexuan Yang Date: Wed, 25 Feb 2026 00:11:54 +0800 Subject: [PATCH 1/4] docs: update README and ARCHITECTURE to reflect current codebase - Add codecov badge - Update workspace structure (7 crates: added executor, media, pipeline) - Replace worker/scanner commands with unified command - Add development infrastructure section (Docker Compose) - Update ARCHITECTURE.md with new crates and CLI commands - Sync Chinese README with English version --- ARCHITECTURE.md | 109 ++++++++++++++++++++++++++++++++---------------- README.md | 67 ++++++++++++++++++++--------- README_zh.md | 67 ++++++++++++++++++++--------- 3 files changed, 168 insertions(+), 75 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 8abb853d..d7486c74 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -27,10 +27,11 @@ Roboflow is a distributed data transformation pipeline that converts robotics ba |-------|---------|-----------| | `roboflow-core` | Foundation types, error handling, registry | `RoboflowError`, `CodecValue`, `TypeRegistry` | | `roboflow-storage` | Storage abstraction layer | `Storage`, `LocalStorage`, `S3Storage`, `StorageFactory` | -| `roboflow-dataset` | Dataset format writers | `LerobotWriter`, `DatasetWriter`, `ImageData` | -| `roboflow-distributed` | Distributed coordination via TiKV | `TiKVClient`, `BatchController`, `Worker`, `Catalog` | -| `roboflow-sources` | Data source implementations | `BagSource`, `McapSource`, `RrdSource` | -| `roboflow-sinks` | Data sink implementations | `LerobotSink`, `ZarrSink`, `DatasetFrame` | +| `roboflow-executor` | Stage-based task executor | `StageExecutor`, `Pipeline`, `ExecutionPolicy`, `SlotPool` | +| `roboflow-media` | Image and video encoding/decoding | `ImageData`, `VideoEncoder`, `ConcurrentVideoEncoder` | +| `roboflow-dataset` | Dataset format writers and sources | `LerobotWriter`, `DatasetWriter`, `Source`, `BagSource`, `McapSource` | +| `roboflow-pipeline` | Pipeline execution and stages | `DatasetPipelineExecutor`, `DiscoverStage`, `ConvertStage`, `MergeStage` | +| `roboflow-distributed` | Distributed coordination via TiKV | `TiKVClient`, `BatchController`, `Worker`, `Scanner`, `Finalizer` | ## Core Abstractions @@ -55,7 +56,7 @@ trait SeekableStorage: Storage { - **S3**: AWS S3-compatible storage - **OSS**: Alibaba Cloud Object Storage -### Pipeline Stages +### Source/Sink Pattern ```rust trait Source: Send + Sync { @@ -63,13 +64,25 @@ trait Source: Send + Sync { async fn read_batch(&mut self, size: usize) -> SourceResult>>; async fn finalize(&mut self) -> SourceResult; } +``` + +**Supported sources:** +- **MCAP**: Streaming and memory-mapped reads +- **ROS1 Bag**: Legacy bag format support +- **RRD**: Rerun data format + +### Pipeline Stages -trait Sink: Send + Sync { - async fn initialize(&mut self, config: &SinkConfig) -> SinkResult<()>; - async fn write_frame(&mut self, frame: DatasetFrame) -> SinkResult<()>; - async fn flush(&mut self) -> SinkResult<()>; - async fn finalize(&mut self) -> SinkResult; - fn supports_checkpointing(&self) -> bool; +```rust +// Stage-based execution inspired by Spark +pub struct DiscoverStage; +pub struct ConvertStage; +pub struct MergeStage; + +// Pipeline executor +pub struct DatasetPipelineExecutor { + writer: Box, + config: DatasetPipelineConfig, } ``` @@ -117,6 +130,7 @@ The distributed system uses a Kubernetes-inspired design with TiKV as the contro | kubelet heartbeat | HeartbeatManager | Worker liveness | | Finalizers | Finalizer controller | Cleanup handling | | Job/CronJob | BatchSpec, WorkUnit | Work scheduling | +| Scheduler | Scanner | File discovery and job creation | ### Batch State Machine @@ -124,11 +138,11 @@ The distributed system uses a Kubernetes-inspired design with TiKV as the contro ┌──────────┐ ┌─────────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ Pending │───▶│ Discovering │───▶│ Running │───▶│ Merging │───▶│ Complete │ └──────────┘ └─────────────┘ └──────────┘ └──────────┘ └──────────┘ - │ - ▼ - ┌──────────┐ - │ Failed │ - └──────────┘ + │ + ▼ + ┌──────────┐ + │ Failed │ + └──────────┘ ``` ### TiKV Key Structure @@ -142,6 +156,27 @@ roboflow/worker/{pod_id}/lock → LockRecord roboflow/worker/{pod_id}/checkpoint→ CheckpointState ``` +## CLI Commands + +The unified `roboflow` binary provides all operations: + +```bash +# Run unified service (default: all roles) +roboflow run + +# Run specific roles +roboflow run --role worker +roboflow run --role finalizer + +# Job management +roboflow submit --input s3://bucket/file.bag --output s3://bucket/out/ +roboflow jobs list +roboflow batch list + +# Health check +roboflow health +``` + ## Dataset Writing ### LeRobot Format @@ -151,23 +186,30 @@ struct LerobotConfig { pub dataset: DatasetConfig, pub mappings: Vec, pub video: VideoConfig, - pub flushing: FlushingConfig, // Incremental flushing + pub flushing: FlushingConfig, + pub streaming: StreamingConfig, } -struct FlushingConfig { - pub max_frames_per_chunk: usize, // Default: 1000 - pub max_memory_bytes: usize, // Default: 2GB - pub incremental_video_encoding: bool, +struct StreamingConfig { + pub finalize_metadata_in_coordinator: bool, } ``` -### Incremental Flushing +### Video Encoding -To prevent OOM on long recordings, the writer processes data in chunks: +```rust +// Concurrent video encoder for parallel chunk encoding +pub struct ConcurrentVideoEncoder { + config: ConcurrentEncoderConfig, +} -1. **Frame-based**: Flush after N frames (configurable, default 1000) -2. **Memory-based**: Flush when memory exceeds threshold (default 2GB) -3. **Output structure**: `data/chunk-000/`, `data/chunk-001/`, etc. +pub struct ConcurrentEncoderConfig { + pub storage: Arc, + pub key_prefix: String, + pub codec: VideoCodec, + pub crf: u8, +} +``` ### Upload Coordinator @@ -176,7 +218,6 @@ struct EpisodeUploadCoordinator { pub storage: Arc, pub config: UploadConfig, pub progress: Option, - // Worker pool for parallel uploads } struct UploadConfig { @@ -213,7 +254,7 @@ let data = arena.alloc_vec::(size); ```toml [source] -type = "mcap" # or "bag", "rrd", "hdf5" +type = "mcap" # or "bag", "rrd" path = "s3://bucket/path/to/data.mcap" # Optional: topic filtering @@ -325,17 +366,15 @@ enum CircuitState { | Flag | Purpose | |------|---------| -| `distributed` | TiKV distributed coordination (always enabled) | -| `dataset-hdf5` | HDF5 dataset format support | -| `dataset-parquet` | Parquet dataset format support | -| `cloud-storage` | S3/OSS cloud storage support | -| `gpu` | GPU compression (Linux only) | | `jemalloc` | jemalloc allocator (Linux only) | | `cli` | CLI support for binaries | +| `profiling` | Profiling support for profiler binary | +| `cpuid` | CPU-aware WindowLog detection (x86_64 only) | +| `io-uring-io` | io_uring support for Linux 5.6+ | ## See Also - `CLAUDE.md` - Developer guidelines and conventions -- `tests/s3_pipeline_tests.rs` - Integration tests -- `crates/roboflow-dataset/src/lerobot/` - Dataset writer implementation +- `tests/` - Integration and E2E tests +- `crates/roboflow-dataset/src/` - Dataset writer and source implementations - `crates/roboflow-distributed/src/` - Distributed coordination diff --git a/README.md b/README.md index d072afa8..c5a00ec6 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,7 @@ [![License: MulanPSL-2.0](https://img.shields.io/badge/License-MulanPSL--2.0-blue.svg)](http://license.coscl.org.cn/MulanPSL2) [![Rust](https://img.shields.io/badge/rust-1.80%2B-orange.svg)](https://www.rust-lang.org) +[![codecov](https://codecov.io/gh/archebase/roboflow/branch/main/graph/badge.svg)](https://codecov.io/gh/archebase/roboflow) [English](README.md) | [简体中文](README_zh.md) @@ -73,46 +74,55 @@ Roboflow uses a **Kubernetes-inspired distributed control plane** for fault-tole |-------|---------| | `roboflow-core` | Error types, registry, values | | `roboflow-storage` | S3, OSS, Local storage (always available) | -| `roboflow-dataset` | KPS, LeRobot, streaming converters | -| `roboflow-distributed` | TiKV client, catalog, circuit breaker | -| `roboflow-hdf5` | Optional HDF5 format support | +| `roboflow-executor` | Stage-based task executor for distributed pipelines | +| `roboflow-media` | Image and video encoding/decoding for robotics datasets | +| `roboflow-dataset` | KPS, LeRobot, streaming converters, data sources | +| `roboflow-pipeline` | Pipeline execution and stages for dataset processing | +| `roboflow-distributed` | TiKV client, catalog, circuit breaker, worker coordination | ## Quick Start -### Submit a Conversion Job - -```bash -roboflow submit \ - --input s3://bucket/input.bag \ - --output s3://bucket/output/ \ - --config lerobot_config.toml -``` - -### Run a Worker +### Run the Unified Service ```bash +# Set environment variables export TIKV_PD_ENDPOINTS="127.0.0.1:2379" export AWS_ACCESS_KEY_ID="your-key" export AWS_SECRET_ACCESS_KEY="your-secret" -roboflow worker +# Run unified service (scanner + worker + finalizer + reaper) +roboflow run ``` -### Run a Scanner +### Run Specific Roles ```bash -export SCANNER_INPUT_PREFIX="s3://bucket/input/" -export SCANNER_OUTPUT_PREFIX="s3://bucket/jobs/" +# Worker only - processes work units +roboflow run --role worker + +# Finalizer only - merges completed batches +roboflow run --role finalizer -roboflow scanner +# With custom pod ID +roboflow run --pod-id my-pod-1 ``` -### List Jobs +### Submit a Conversion Job + +```bash +roboflow submit \ + --input s3://bucket/input.bag \ + --output s3://bucket/output/ \ + --config lerobot_config.toml +``` + +### Manage Jobs ```bash roboflow jobs list roboflow jobs get -roboflow jobs retry +roboflow batch list +roboflow batch get ``` ## Installation @@ -166,6 +176,7 @@ encoding = "cdr" | `WORKER_POLL_INTERVAL_SECS` | Job poll interval | `5` | | `WORKER_MAX_CONCURRENT_JOBS` | Max concurrent jobs | `1` | | `SCANNER_SCAN_INTERVAL_SECS` | Scan interval | `60` | +| `FINALIZER_POLL_INTERVAL_SECS` | Finalizer poll interval | `30` | ## Development @@ -188,6 +199,22 @@ cargo fmt cargo clippy --all-targets -- -D warnings ``` +### Development Infrastructure + +Start required services with Docker Compose: + +```bash +docker compose up -d # Start all services (MinIO, TiKV, PD) +docker compose down # Stop all services +``` + +**Services:** +| Service | Purpose | Ports | +|---------|---------|-------| +| MinIO | S3-compatible object storage | 9000 (API), 9001 (Console) | +| TiKV | Distributed KV storage | 20160 | +| PD | TiKV placement driver | 2379, 2380 | + ## Contributing See [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and guidelines. diff --git a/README_zh.md b/README_zh.md index 59d7b0bb..20047589 100644 --- a/README_zh.md +++ b/README_zh.md @@ -2,6 +2,7 @@ [![License: MulanPSL-2.0](https://img.shields.io/badge/License-MulanPSL--2.0-blue.svg)](http://license.coscl.org.cn/MulanPSL2) [![Rust](https://img.shields.io/badge/rust-1.80%2B-orange.svg)](https://www.rust-lang.org) +[![codecov](https://codecov.io/gh/archebase/roboflow/branch/main/graph/badge.svg)](https://codecov.io/gh/archebase/roboflow) [English](README.md) | [简体中文](README_zh.md) @@ -73,46 +74,55 @@ Roboflow 采用**受 Kubernetes 启发的分布式控制平面**，实现容错 |-------|------| | `roboflow-core` | 错误类型、注册表、值类型 | | `roboflow-storage` | S3、OSS、本地存储（始终可用） | -| `roboflow-dataset` | KPS、LeRobot、流式转换器 | -| `roboflow-distributed` | TiKV 客户端、目录、控制器 | -| `roboflow-hdf5` | 可选的 HDF5 格式支持 | +| `roboflow-executor` | 分布式流水线的阶段任务执行器 | +| `roboflow-media` | 机器人数据集的图像和视频编解码 | +| `roboflow-dataset` | KPS、LeRobot、流式转换器、数据源 | +| `roboflow-pipeline` | 数据集处理的流水线执行和阶段 | +| `roboflow-distributed` | TiKV 客户端、目录、熔断器、Worker 协调 | ## 快速开始 -### 提交转换任务 - -```bash -roboflow submit \ - --input s3://bucket/input.bag \ - --output s3://bucket/output/ \ - --config lerobot_config.toml -``` - -### 运行 Worker +### 运行统一服务 ```bash +# 设置环境变量 export TIKV_PD_ENDPOINTS="127.0.0.1:2379" export AWS_ACCESS_KEY_ID="your-key" export AWS_SECRET_ACCESS_KEY="your-secret" -roboflow worker +# 运行统一服务 (scanner + worker + finalizer + reaper) +roboflow run ``` -### 运行 Scanner +### 运行特定角色 ```bash -export SCANNER_INPUT_PREFIX="s3://bucket/input/" -export SCANNER_OUTPUT_PREFIX="s3://bucket/jobs/" +# 仅 Worker - 处理工作单元 +roboflow run --role worker + +# 仅 Finalizer - 合并已完成的批次 +roboflow run --role finalizer -roboflow scanner +# 使用自定义 Pod ID +roboflow run --pod-id my-pod-1 ``` -### 列出任务 +### 提交转换任务 + +```bash +roboflow submit \ + --input s3://bucket/input.bag \ + --output s3://bucket/output/ \ + --config lerobot_config.toml +``` + +### 管理任务 ```bash roboflow jobs list roboflow jobs get -roboflow jobs retry +roboflow batch list +roboflow batch get ``` ## 安装 @@ -166,6 +176,7 @@ encoding = "cdr" | `WORKER_POLL_INTERVAL_SECS` | 任务轮询间隔 | `5` | | `WORKER_MAX_CONCURRENT_JOBS` | 最大并发任务数 | `1` | | `SCANNER_SCAN_INTERVAL_SECS` | 扫描间隔 | `60` | +| `FINALIZER_POLL_INTERVAL_SECS` | Finalizer 轮询间隔 | `30` | ## 开发 @@ -188,6 +199,22 @@ cargo fmt cargo clippy --all-targets -- -D warnings ``` +### 开发基础设施 + +使用 Docker Compose 启动所需服务： + +```bash +docker compose up -d # 启动所有服务 (MinIO, TiKV, PD) +docker compose down # 停止所有服务 +``` + +**服务：** +| 服务 | 用途 | 端口 | +|------|------|------| +| MinIO | S3 兼容对象存储 | 9000 (API), 9001 (控制台) | +| TiKV | 分布式 KV 存储 | 20160 | +| PD | TiKV 调度器 | 2379, 2380 | + ## 贡献 详见 [CONTRIBUTING.md](CONTRIBUTING.md)。 From 889ef2482f7ad672dd327a8d6a67e7cf176f0c22 Mon Sep 17 00:00:00 2001 From: Zhexuan Yang Date: Wed, 25 Feb 2026 00:16:29 +0800 Subject: [PATCH 2/4] Update ARCHITECTURE.md Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com> --- ARCHITECTURE.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d7486c74..2deff689 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -169,7 +169,7 @@ roboflow run --role worker roboflow run --role finalizer # Job management -roboflow submit --input s3://bucket/file.bag --output s3://bucket/out/ +roboflow submit s3://bucket/file.bag --output s3://bucket/out/ roboflow jobs list roboflow batch list From c75553e9385c7447a4007c66d4239ec74836e215 Mon Sep 17 00:00:00 2001 From: Zhexuan Yang Date: Wed, 25 Feb 2026 00:18:45 +0800 Subject: [PATCH 3/4] docs: update CLAUDE.md workspace structure to 7 crates --- CLAUDE.md | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 01dab6d7..f68a91e1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -14,15 +14,17 @@ Roboflow: Distributed data transformation pipeline converting robotics bag/MCAP ## Workspace Structure -The project uses a Cargo workspace with 5 crates: +The project uses a Cargo workspace with 7 crates: | Crate | Purpose | |-------|---------| | `roboflow-core` | Error types, registry, values | | `roboflow-storage` | S3, OSS, Local storage (always available) | -| `roboflow-dataset` | KPS, LeRobot, streaming converters | -| `roboflow-distributed` | TiKV client, catalog, circuit breaker | -| `roboflow-hdf5` | Optional HDF5 format support | +| `roboflow-executor` | Stage-based task executor for distributed pipelines | +| `roboflow-media` | Image and video encoding/decoding | +| `roboflow-dataset` | Dataset writers, sources (MCAP, bag), streaming converters | +| `roboflow-pipeline` | Pipeline execution and stages for dataset processing | +| `roboflow-distributed` | TiKV client, catalog, circuit breaker, worker coordination | **Import patterns:** - Use facade re-exports from `roboflow`: `use roboflow::{Robocodec, DatasetWriter, ...}` From 9a83aad0ce5514b2608aa92dfe22d241ec85ee25 Mon Sep 17 00:00:00 2001 From: Zhexuan Yang Date: Wed, 25 Feb 2026 00:20:22 +0800 Subject: [PATCH 4/4] fix: correct submit command syntax (input is positional, not --input flag) --- README.md | 5 +---- README_zh.md | 5 +---- 2 files changed, 2 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index c5a00ec6..70865e49 100644 --- a/README.md +++ b/README.md @@ -110,10 +110,7 @@ roboflow run --pod-id my-pod-1 ### Submit a Conversion Job ```bash -roboflow submit \ - --input s3://bucket/input.bag \ - --output s3://bucket/output/ \ - --config lerobot_config.toml +roboflow submit s3://bucket/input.bag --output s3://bucket/output/ ``` ### Manage Jobs diff --git a/README_zh.md b/README_zh.md index 20047589..08a80fa6 100644 --- a/README_zh.md +++ b/README_zh.md @@ -110,10 +110,7 @@ roboflow run --pod-id my-pod-1 ### 提交转换任务 ```bash -roboflow submit \ - --input s3://bucket/input.bag \ - --output s3://bucket/output/ \ - --config lerobot_config.toml +roboflow submit s3://bucket/input.bag --output s3://bucket/output/ ``` ### 管理任务