From b1bcf4a0d4c59c416d8a89c4823ee174696b63ed Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Tue, 8 Jul 2025 21:52:18 +0000
Subject: [PATCH 1/2] Initial plan


From c6c32b599d70fdf3a79d993194c914143584a9ca Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Tue, 8 Jul 2025 21:58:10 +0000
Subject: [PATCH 2/2] Complete ReplyBuffer analysis with class diagrams and
 sequence diagrams

Co-authored-by: zhshgmail <126103537+zhshgmail@users.noreply.github.com>
---
 BUFFER_ANALYSIS_SUMMARY.md           |  72 +++++++
 docs/README.md                       |  86 +++++++++
 docs/amend_batch_sequence.puml       |  34 ++++
 docs/buffer_analysis.md              | 276 +++++++++++++++++++++++++++
 docs/buffer_analysis_en.md           | 276 +++++++++++++++++++++++++++
 docs/buffer_class_diagram.puml       |  76 ++++++++
 docs/get_batch_for_rpc_sequence.puml |  51 +++++
 docs/put_batch_sequence.puml         |  34 ++++
 8 files changed, 905 insertions(+)
 create mode 100644 BUFFER_ANALYSIS_SUMMARY.md
 create mode 100644 docs/README.md
 create mode 100644 docs/amend_batch_sequence.puml
 create mode 100644 docs/buffer_analysis.md
 create mode 100644 docs/buffer_analysis_en.md
 create mode 100644 docs/buffer_class_diagram.puml
 create mode 100644 docs/get_batch_for_rpc_sequence.puml
 create mode 100644 docs/put_batch_sequence.puml

diff --git a/BUFFER_ANALYSIS_SUMMARY.md b/BUFFER_ANALYSIS_SUMMARY.md
new file mode 100644
index 0000000000..2aca387b30
--- /dev/null
+++ b/BUFFER_ANALYSIS_SUMMARY.md
@@ -0,0 +1,72 @@
+# ReplyBuffer Analysis Summary
+
+## 问题回答 (Answer to the Question)
+
+根据对 AReaL 中 ReplyBuffer 实现的分析，我已经创建了详细的类图和时序图来说明其架构和行为。
+
+Based on the analysis of the ReplyBuffer implementation in AReaL, I have created detailed class diagrams and sequence diagrams to illustrate its architecture and behavior.
+
+## 核心发现 (Key Findings)
+
+### 1. 架构设计 (Architecture Design)
+- **三层结构**: AsyncIOSequenceBuffer (公共接口) → _TensorDictSequenceBuffer (内部存储) → _ReplayEntry (数据条目)
+- **Three-tier structure**: AsyncIOSequenceBuffer (public interface) → _TensorDictSequenceBuffer (internal storage) → _ReplayEntry (data entries)
+
+### 2. 并发控制 (Concurrency Control)  
+- 使用 asyncio.Condition 实现异步同步
+- 通过状态数组管理并发访问
+- 支持多读者和多写者并发操作
+
+### 3. 状态管理 (State Management)
+- 五种互斥状态：being_put, being_amended, being_read, idle, empty
+- 使用 numpy 数组提供 O(1) 状态操作
+- 完整的状态一致性检查
+
+### 4. RPC 集成 (RPC Integration)
+- 智能的 RPC 就绪检测
+- 基于数据键的依赖解析
+- 支持多个 RPC 的并发执行
+
+## 文档结构 (Documentation Structure)
+
+```
+docs/
+├── README.md                           # 总览文档
+├── buffer_analysis.md                  # 中文详细分析  
+├── buffer_analysis_en.md              # 英文详细分析
+├── buffer_class_diagram.puml          # 类图 (PlantUML)
+├── put_batch_sequence.puml            # put_batch 时序图
+├── amend_batch_sequence.puml          # amend_batch 时序图
+└── get_batch_for_rpc_sequence.puml    # get_batch_for_rpc 时序图
+```
+
+## 主要操作流程 (Main Operation Flows)
+
+### 1. put_batch (数据写入)
+1. 获取锁并验证状态
+2. 找到空闲索引
+3. 设置写入状态
+4. 执行实际写入操作
+5. 更新就绪状态和通知等待者
+
+### 2. amend_batch (数据修改)
+1. 等待条目变为空闲或可修改状态
+2. 增加修改者计数
+3. 执行数据更新
+4. 更新状态并通知
+
+### 3. get_batch_for_rpc (数据读取)
+1. 等待 RPC 所需数据就绪
+2. 按时间顺序选择条目
+3. 设置读取状态
+4. 执行数据读取和重用计数
+5. 清理已耗尽的条目
+
+## 设计优势 (Design Advantages)
+
+1. **高并发性能**: 精心设计的锁策略和状态管理
+2. **内存效率**: 固定大小缓冲区和重用计数机制
+3. **灵活性**: 支持动态数据修改和多种 RPC 操作
+4. **可靠性**: 完善的状态检查和异常处理
+
+这个分析提供了对 AReaL ReplyBuffer 实现的全面理解，包括其设计原理、关键特性和操作流程。
\ No newline at end of file
diff --git a/docs/README.md b/docs/README.md
new file mode 100644
index 0000000000..13e4daf261
--- /dev/null
+++ b/docs/README.md
@@ -0,0 +1,86 @@
+# AReaL ReplyBuffer Documentation
+
+This directory contains the comprehensive analysis and documentation of the ReplyBuffer (AsyncIOSequenceBuffer) implementation in AReaL.
+
+## Files Overview
+
+### Analysis Documents
+- **`buffer_analysis.md`** - Detailed analysis in Chinese (中文分析文档)
+- **`buffer_analysis_en.md`** - Detailed analysis in English
+
+### Diagram Files
+
+#### PlantUML Diagrams
+- **`buffer_class_diagram.puml`** - Class diagram showing relationships between buffer components
+- **`put_batch_sequence.puml`** - Sequence diagram for put_batch operation
+- **`amend_batch_sequence.puml`** - Sequence diagram for amend_batch operation  
+- **`get_batch_for_rpc_sequence.puml`** - Sequence diagram for get_batch_for_rpc operation
+
+## Key Findings
+
+### Architecture Overview
+The AReaL ReplyBuffer system consists of three main components:
+
+1. **AsyncIOSequenceBuffer** - The main async buffer class that provides the public interface
+2. **_TensorDictSequenceBuffer** - Internal storage implementation with thread-unsafe operations
+3. **_ReplayEntry** - Data structure for individual buffer entries
+
+### Key Features
+- **Asynchronous Operations**: Full async/await support for non-blocking operations
+- **Concurrent Access**: Multiple readers and writers can access the buffer simultaneously
+- **State Management**: Sophisticated state tracking using numpy arrays for thread safety
+- **Memory Efficiency**: Reuse counting and fixed-size allocation for optimal memory usage
+- **RPC Integration**: Built-in support for multiple RPC operations with dependency resolution
+
+### Buffer States
+The buffer maintains mutually exclusive states for each entry:
+- `_is_being_put` - Entry is being written
+- `_is_being_amended` - Entry is being modified
+- `_is_being_read` - Entry is being read
+- `_is_idle` - Entry is available for operations
+- `_is_empty` - Entry slot is empty
+
+## Usage in AReaL Context
+
+The ReplyBuffer serves as the central data management component in AReaL's asynchronous reinforcement learning pipeline:
+
+1. **Data Ingestion**: Rollout workers put trajectory data into the buffer
+2. **Data Enhancement**: Reward services and other processors amend data with additional information
+3. **Training Data Delivery**: Trainer workers retrieve batches for model updates
+4. **Memory Management**: Automatic cleanup based on reuse counting
+
+## Viewing Diagrams
+
+### PlantUML
+To render PlantUML diagrams:
+1. Install PlantUML: `pip install plantuml`
+2. Render diagrams: `plantuml docs/*.puml`
+
+### Mermaid
+The Mermaid diagrams in the markdown files can be viewed directly on GitHub or using:
+1. Mermaid Live Editor: https://mermaid-js.github.io/mermaid-live-editor/
+2. VS Code with Mermaid extension
+3. Any Markdown viewer that supports Mermaid
+
+## Implementation Notes
+
+### Thread Safety
+- All public methods use asyncio.Condition for synchronization
+- State transitions are atomic within lock contexts
+- Concurrent readers/amenders are supported through reference counting
+
+### Performance Considerations
+- Fixed-size numpy arrays for O(1) state operations
+- FIFO ordering based on birth timestamps
+- Minimal memory allocations during normal operation
+
+### Error Handling
+- `BufferFull` exception when capacity is exceeded
+- Comprehensive state validation via `_assert_valid_indicator()`
+- Graceful handling of concurrent access scenarios
+
+## Related Code
+- Source: `/realhf/system/buffer.py`
+- Tests: `/tests/experiments/test_buffer_recover.py`
+- API: `/realhf/api/core/data_api.py` (SequenceSample)
+- RPC: `/realhf/api/core/dfg.py` (MFCDef)
\ No newline at end of file
diff --git a/docs/amend_batch_sequence.puml b/docs/amend_batch_sequence.puml
new file mode 100644
index 0000000000..4e105a1518
--- /dev/null
+++ b/docs/amend_batch_sequence.puml
@@ -0,0 +1,34 @@
+@startuml AmendBatchSequence
+
+participant Client
+participant "AsyncIOSequenceBuffer" as Buffer
+participant "_TensorDictSequenceBuffer" as Internal
+participant "asyncio.Condition" as Lock
+
+Client -> Buffer: amend_batch(indices, samples)
+Buffer -> Lock: acquire()
+Buffer -> Lock: wait_for(indices idle or being_amended)
+Buffer -> Buffer: _assert_valid_indicator()
+Buffer -> Buffer: set _is_idle[indices] = False
+Buffer -> Buffer: set _is_being_amended[indices] = True
+Buffer -> Buffer: increment _n_amenders[indices]
+Buffer -> Lock: release()
+
+Buffer -> Internal: amend_batch(indices, samples)
+Internal -> Internal: update sample data
+
+Buffer -> Lock: acquire()
+Buffer -> Internal: _update_has_keys(indices)
+Buffer -> Buffer: update _ready_for_rpcs
+Buffer -> Buffer: decrement _n_amenders[indices]
+Buffer -> Buffer: update _is_being_amended[indices]
+Buffer -> Buffer: update _is_idle[indices]
+
+alt any indices become idle
+    Buffer -> Lock: notify(n_rpcs)
+end
+
+Buffer -> Lock: release()
+Buffer --> Client: complete
+
+@enduml
\ No newline at end of file
diff --git a/docs/buffer_analysis.md b/docs/buffer_analysis.md
new file mode 100644
index 0000000000..acf1e848f7
--- /dev/null
+++ b/docs/buffer_analysis.md
@@ -0,0 +1,276 @@
+# ReplyBuffer (AsyncIOSequenceBuffer) 分析报告
+
+## 概述
+
+AReaL 中的 ReplyBuffer 实现主要基于 `AsyncIOSequenceBuffer` 类，这是一个用于异步强化学习系统的高性能缓冲区实现。该缓冲区支持并发的读者和修改者，并管理多个 RPC (Remote Procedure Call) 操作的数据流。
+
+## 核心组件
+
+### 1. 主要类结构
+
+#### AsyncIOSequenceBuffer
+- **作用**: 主要的异步序列缓冲区类
+- **特点**: 
+  - 支持异步操作和并发访问
+  - 使用 numpy 数组管理缓冲区状态
+  - 通过 asyncio.Condition 实现同步控制
+  - 支持多个 RPC 操作的数据管理
+
+#### _TensorDictSequenceBuffer  
+- **作用**: 内部存储实现
+- **特点**:
+  - 线程不安全的内部缓冲区实现
+  - 基于 Python 列表的固定大小存储
+  - 管理数据键的可用性状态
+
+#### _ReplayEntry
+- **作用**: 单个缓冲区条目的数据结构
+- **包含**: 重用次数、接收时间、序列样本数据
+
+## 类图
+
+```mermaid
+classDiagram
+    class AsyncIOSequenceBuffer {
+        +List~MFCDef~ rpcs
+        +asyncio.Condition _lock
+        +ndarray _is_being_put
+        +ndarray _is_being_amended  
+        +ndarray _is_being_read
+        +ndarray _is_idle
+        +ndarray _is_empty
+        +ndarray _n_amenders
+        +ndarray _n_readers
+        +ndarray _ready_for_rpcs
+        +ndarray _completed_rpc
+        +_TensorDictSequenceBuffer __buffer
+        
+        +__init__(rpcs, max_size)
+        +put_batch(samples, birth_times) async
+        +amend_batch(indices, samples) async
+        +get_batch_for_rpc(rpc) async
+        +_can_do_rpc(rpc) bool
+        +_assert_valid_indicator()
+        +put_batch_synced(samples)
+    }
+    
+    class _TensorDictSequenceBuffer {
+        +List~_ReplayEntry~ __storage
+        +ndarray __has_keys
+        +List~str~ __keys
+        +int __reuses
+        
+        +__init__(keys, max_size, reuses)
+        +put_batch(indices, xs)
+        +amend_batch(indices, xs)
+        +get_batch(indices) List~_ReplayEntry~
+        +inspect_batch(indices) List~_ReplayEntry~
+        +pop_batch(indices)
+        +_update_has_keys(indices)
+        +_get_has_keys(indices)
+    }
+    
+    class _ReplayEntry {
+        +int reuses_left
+        +float receive_time
+        +SequenceSample sample
+    }
+    
+    class SequenceSample {
+        +Dict data
+        +Dict seqlens
+        +Set keys
+        +update_(other)
+        +gather(samples, keys) SequenceSample
+    }
+    
+    class MFCDef {
+        +str name
+        +int n_seqs
+        +Tuple input_keys
+        +Tuple output_keys
+        +ModelInterfaceType interface_type
+    }
+    
+    class BufferFull {
+        <<Exception>>
+    }
+    
+    AsyncIOSequenceBuffer --> _TensorDictSequenceBuffer : contains
+    _TensorDictSequenceBuffer --> _ReplayEntry : stores
+    _ReplayEntry --> SequenceSample : contains
+    AsyncIOSequenceBuffer --> MFCDef : uses
+    AsyncIOSequenceBuffer ..> BufferFull : throws
+```
+
+## 状态管理
+
+AsyncIOSequenceBuffer 使用多个 numpy 数组来管理缓冲区的状态：
+
+- **_is_being_put**: 正在写入的条目
+- **_is_being_amended**: 正在修改的条目  
+- **_is_being_read**: 正在读取的条目
+- **_is_idle**: 空闲的条目
+- **_is_empty**: 空的条目
+
+这些状态是互斥的，确保数据的一致性和线程安全。
+
+## 主要操作时序图
+
+### 1. put_batch 操作时序图
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant AsyncIOSequenceBuffer as Buffer
+    participant _TensorDictSequenceBuffer as Internal
+    participant Lock as asyncio.Condition
+    
+    Client->>Buffer: put_batch(samples, birth_times)
+    Buffer->>Lock: acquire()
+    Buffer->>Buffer: _assert_valid_indicator()
+    Buffer->>Buffer: find empty indices
+    
+    alt insufficient space
+        Buffer-->>Client: raise BufferFull
+    else sufficient space
+        Buffer->>Buffer: set _is_empty[indices] = False
+        Buffer->>Buffer: set _is_being_put[indices] = True
+        Buffer->>Lock: release()
+        
+        Buffer->>Internal: put_batch(indices, samples)
+        Internal->>Internal: create _ReplayEntry for each sample
+        
+        Buffer->>Lock: acquire()
+        Buffer->>Internal: _update_has_keys(indices)
+        Buffer->>Buffer: update _ready_for_rpcs
+        Buffer->>Buffer: set _is_being_put[indices] = False
+        Buffer->>Buffer: set _is_idle[indices] = True
+        Buffer->>Buffer: increment _buf_size
+        Buffer->>Lock: notify(n_rpcs)
+        Buffer->>Lock: release()
+        Buffer-->>Client: return indices
+    end
+```
+
+### 2. amend_batch 操作时序图
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant AsyncIOSequenceBuffer as Buffer
+    participant _TensorDictSequenceBuffer as Internal
+    participant Lock as asyncio.Condition
+    
+    Client->>Buffer: amend_batch(indices, samples)
+    Buffer->>Lock: acquire()
+    Buffer->>Lock: wait_for(indices idle or being_amended)
+    Buffer->>Buffer: _assert_valid_indicator()
+    Buffer->>Buffer: set _is_idle[indices] = False
+    Buffer->>Buffer: set _is_being_amended[indices] = True
+    Buffer->>Buffer: increment _n_amenders[indices]
+    Buffer->>Lock: release()
+    
+    Buffer->>Internal: amend_batch(indices, samples)
+    Internal->>Internal: update sample data
+    
+    Buffer->>Lock: acquire()
+    Buffer->>Internal: _update_has_keys(indices)
+    Buffer->>Buffer: update _ready_for_rpcs
+    Buffer->>Buffer: decrement _n_amenders[indices]
+    Buffer->>Buffer: update _is_being_amended[indices]
+    Buffer->>Buffer: update _is_idle[indices]
+    
+    alt any indices become idle
+        Buffer->>Lock: notify(n_rpcs)
+    end
+    
+    Buffer->>Lock: release()
+    Buffer-->>Client: complete
+```
+
+### 3. get_batch_for_rpc 操作时序图
+
+```mermaid
+sequenceDiagram
+    participant RPC as RPC Client
+    participant AsyncIOSequenceBuffer as Buffer
+    participant _TensorDictSequenceBuffer as Internal
+    participant Lock as asyncio.Condition
+    
+    RPC->>Buffer: get_batch_for_rpc(rpc)
+    Buffer->>Lock: acquire()
+    
+    loop until can_do_rpc
+        Buffer->>Buffer: _can_do_rpc(rpc)
+        alt not ready
+            Buffer->>Lock: wait()
+        end
+    end
+    
+    Buffer->>Buffer: _assert_valid_indicator()
+    Buffer->>Buffer: find ready_indices for rpc
+    Buffer->>Buffer: sort by birth_time (FIFO)
+    Buffer->>Buffer: select n_seqs indices
+    Buffer->>Buffer: set _is_idle[indices] = False
+    Buffer->>Buffer: set _is_being_read[indices] = True
+    Buffer->>Buffer: increment _n_readers[indices]
+    Buffer->>Lock: release()
+    
+    Buffer->>Internal: get_batch(indices)
+    Internal->>Internal: decrement reuses_left
+    Internal->>Internal: identify entries with reuses_left = 0
+    
+    alt has entries to pop
+        Buffer->>Internal: pop_batch(pop_indices)
+        Internal->>Internal: clear storage and keys
+    end
+    
+    Buffer->>Lock: acquire()
+    Buffer->>Buffer: decrement _n_readers[indices]
+    Buffer->>Buffer: update _is_being_read[indices]
+    Buffer->>Buffer: update _is_idle[indices]
+    Buffer->>Buffer: set _completed_rpc[indices, rpc_idx] = True
+    Buffer->>Buffer: clean up popped indices
+    Buffer->>Buffer: decrement _buf_size
+    
+    alt any indices become idle
+        Buffer->>Lock: notify(n_rpcs)
+    end
+    
+    Buffer->>Lock: release()
+    Buffer-->>RPC: return (indices, SequenceSample)
+```
+
+## 关键特性
+
+### 1. 并发控制
+- 使用 `asyncio.Condition` 实现异步同步
+- 支持多个读者和修改者并发访问
+- 通过状态数组确保操作的原子性
+
+### 2. 内存管理
+- 基于重用计数的自动内存回收
+- 固定大小的缓冲区避免动态内存分配
+- 支持数据的就地更新
+
+### 3. RPC 支持
+- 智能的 RPC 就绪检测
+- 基于数据键的依赖解析
+- 支持多个 RPC 的并发执行
+
+### 4. 数据优先级
+- FIFO (First In, First Out) 数据处理策略
+- 基于出生时间的数据排序
+- 确保训练数据的时序一致性
+
+## 总结
+
+AsyncIOSequenceBuffer 是 AReaL 系统中的核心组件，通过精心设计的状态管理和并发控制，实现了高效的异步数据缓冲。其主要优势包括：
+
+1. **高并发**: 支持多个读者和写者的并发访问
+2. **内存效率**: 通过重用计数和固定大小缓冲区优化内存使用
+3. **灵活性**: 支持动态的数据修改和多种 RPC 操作
+4. **可靠性**: 完善的状态检查和异常处理机制
+
+这个设计使得 AReaL 能够在分布式强化学习环境中提供稳定、高效的数据管理服务。
\ No newline at end of file
diff --git a/docs/buffer_analysis_en.md b/docs/buffer_analysis_en.md
new file mode 100644
index 0000000000..ae6818a6af
--- /dev/null
+++ b/docs/buffer_analysis_en.md
@@ -0,0 +1,276 @@
+# ReplyBuffer (AsyncIOSequenceBuffer) Analysis Report
+
+## Overview
+
+The ReplyBuffer implementation in AReaL is primarily based on the `AsyncIOSequenceBuffer` class, which is a high-performance buffer implementation for asynchronous reinforcement learning systems. This buffer supports concurrent readers and modifiers, and manages data flow for multiple RPC (Remote Procedure Call) operations.
+
+## Core Components
+
+### 1. Main Class Structure
+
+#### AsyncIOSequenceBuffer
+- **Purpose**: Main asynchronous sequence buffer class
+- **Features**: 
+  - Supports asynchronous operations and concurrent access
+  - Uses numpy arrays for buffer state management
+  - Implements synchronization control via asyncio.Condition
+  - Supports data management for multiple RPC operations
+
+#### _TensorDictSequenceBuffer  
+- **Purpose**: Internal storage implementation
+- **Features**:
+  - Thread-unsafe internal buffer implementation
+  - Fixed-size storage based on Python lists
+  - Manages data key availability states
+
+#### _ReplayEntry
+- **Purpose**: Data structure for individual buffer entries
+- **Contains**: Reuse count, receive time, sequence sample data
+
+## Class Diagram
+
+```mermaid
+classDiagram
+    class AsyncIOSequenceBuffer {
+        +List~MFCDef~ rpcs
+        +asyncio.Condition _lock
+        +ndarray _is_being_put
+        +ndarray _is_being_amended  
+        +ndarray _is_being_read
+        +ndarray _is_idle
+        +ndarray _is_empty
+        +ndarray _n_amenders
+        +ndarray _n_readers
+        +ndarray _ready_for_rpcs
+        +ndarray _completed_rpc
+        +_TensorDictSequenceBuffer __buffer
+        
+        +__init__(rpcs, max_size)
+        +put_batch(samples, birth_times) async
+        +amend_batch(indices, samples) async
+        +get_batch_for_rpc(rpc) async
+        +_can_do_rpc(rpc) bool
+        +_assert_valid_indicator()
+        +put_batch_synced(samples)
+    }
+    
+    class _TensorDictSequenceBuffer {
+        +List~_ReplayEntry~ __storage
+        +ndarray __has_keys
+        +List~str~ __keys
+        +int __reuses
+        
+        +__init__(keys, max_size, reuses)
+        +put_batch(indices, xs)
+        +amend_batch(indices, xs)
+        +get_batch(indices) List~_ReplayEntry~
+        +inspect_batch(indices) List~_ReplayEntry~
+        +pop_batch(indices)
+        +_update_has_keys(indices)
+        +_get_has_keys(indices)
+    }
+    
+    class _ReplayEntry {
+        +int reuses_left
+        +float receive_time
+        +SequenceSample sample
+    }
+    
+    class SequenceSample {
+        +Dict data
+        +Dict seqlens
+        +Set keys
+        +update_(other)
+        +gather(samples, keys) SequenceSample
+    }
+    
+    class MFCDef {
+        +str name
+        +int n_seqs
+        +Tuple input_keys
+        +Tuple output_keys
+        +ModelInterfaceType interface_type
+    }
+    
+    class BufferFull {
+        <<Exception>>
+    }
+    
+    AsyncIOSequenceBuffer --> _TensorDictSequenceBuffer : contains
+    _TensorDictSequenceBuffer --> _ReplayEntry : stores
+    _ReplayEntry --> SequenceSample : contains
+    AsyncIOSequenceBuffer --> MFCDef : uses
+    AsyncIOSequenceBuffer ..> BufferFull : throws
+```
+
+## State Management
+
+AsyncIOSequenceBuffer uses multiple numpy arrays to manage buffer states:
+
+- **_is_being_put**: Entries being written
+- **_is_being_amended**: Entries being modified  
+- **_is_being_read**: Entries being read
+- **_is_idle**: Idle entries
+- **_is_empty**: Empty entries
+
+These states are mutually exclusive, ensuring data consistency and thread safety.
+
+## Key Operation Sequence Diagrams
+
+### 1. put_batch Operation Sequence Diagram
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant AsyncIOSequenceBuffer as Buffer
+    participant _TensorDictSequenceBuffer as Internal
+    participant Lock as asyncio.Condition
+    
+    Client->>Buffer: put_batch(samples, birth_times)
+    Buffer->>Lock: acquire()
+    Buffer->>Buffer: _assert_valid_indicator()
+    Buffer->>Buffer: find empty indices
+    
+    alt insufficient space
+        Buffer-->>Client: raise BufferFull
+    else sufficient space
+        Buffer->>Buffer: set _is_empty[indices] = False
+        Buffer->>Buffer: set _is_being_put[indices] = True
+        Buffer->>Lock: release()
+        
+        Buffer->>Internal: put_batch(indices, samples)
+        Internal->>Internal: create _ReplayEntry for each sample
+        
+        Buffer->>Lock: acquire()
+        Buffer->>Internal: _update_has_keys(indices)
+        Buffer->>Buffer: update _ready_for_rpcs
+        Buffer->>Buffer: set _is_being_put[indices] = False
+        Buffer->>Buffer: set _is_idle[indices] = True
+        Buffer->>Buffer: increment _buf_size
+        Buffer->>Lock: notify(n_rpcs)
+        Buffer->>Lock: release()
+        Buffer-->>Client: return indices
+    end
+```
+
+### 2. amend_batch Operation Sequence Diagram
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant AsyncIOSequenceBuffer as Buffer
+    participant _TensorDictSequenceBuffer as Internal
+    participant Lock as asyncio.Condition
+    
+    Client->>Buffer: amend_batch(indices, samples)
+    Buffer->>Lock: acquire()
+    Buffer->>Lock: wait_for(indices idle or being_amended)
+    Buffer->>Buffer: _assert_valid_indicator()
+    Buffer->>Buffer: set _is_idle[indices] = False
+    Buffer->>Buffer: set _is_being_amended[indices] = True
+    Buffer->>Buffer: increment _n_amenders[indices]
+    Buffer->>Lock: release()
+    
+    Buffer->>Internal: amend_batch(indices, samples)
+    Internal->>Internal: update sample data
+    
+    Buffer->>Lock: acquire()
+    Buffer->>Internal: _update_has_keys(indices)
+    Buffer->>Buffer: update _ready_for_rpcs
+    Buffer->>Buffer: decrement _n_amenders[indices]
+    Buffer->>Buffer: update _is_being_amended[indices]
+    Buffer->>Buffer: update _is_idle[indices]
+    
+    alt any indices become idle
+        Buffer->>Lock: notify(n_rpcs)
+    end
+    
+    Buffer->>Lock: release()
+    Buffer-->>Client: complete
+```
+
+### 3. get_batch_for_rpc Operation Sequence Diagram
+
+```mermaid
+sequenceDiagram
+    participant RPC as RPC Client
+    participant AsyncIOSequenceBuffer as Buffer
+    participant _TensorDictSequenceBuffer as Internal
+    participant Lock as asyncio.Condition
+    
+    RPC->>Buffer: get_batch_for_rpc(rpc)
+    Buffer->>Lock: acquire()
+    
+    loop until can_do_rpc
+        Buffer->>Buffer: _can_do_rpc(rpc)
+        alt not ready
+            Buffer->>Lock: wait()
+        end
+    end
+    
+    Buffer->>Buffer: _assert_valid_indicator()
+    Buffer->>Buffer: find ready_indices for rpc
+    Buffer->>Buffer: sort by birth_time (FIFO)
+    Buffer->>Buffer: select n_seqs indices
+    Buffer->>Buffer: set _is_idle[indices] = False
+    Buffer->>Buffer: set _is_being_read[indices] = True
+    Buffer->>Buffer: increment _n_readers[indices]
+    Buffer->>Lock: release()
+    
+    Buffer->>Internal: get_batch(indices)
+    Internal->>Internal: decrement reuses_left
+    Internal->>Internal: identify entries with reuses_left = 0
+    
+    alt has entries to pop
+        Buffer->>Internal: pop_batch(pop_indices)
+        Internal->>Internal: clear storage and keys
+    end
+    
+    Buffer->>Lock: acquire()
+    Buffer->>Buffer: decrement _n_readers[indices]
+    Buffer->>Buffer: update _is_being_read[indices]
+    Buffer->>Buffer: update _is_idle[indices]
+    Buffer->>Buffer: set _completed_rpc[indices, rpc_idx] = True
+    Buffer->>Buffer: clean up popped indices
+    Buffer->>Buffer: decrement _buf_size
+    
+    alt any indices become idle
+        Buffer->>Lock: notify(n_rpcs)
+    end
+    
+    Buffer->>Lock: release()
+    Buffer-->>RPC: return (indices, SequenceSample)
+```
+
+## Key Features
+
+### 1. Concurrency Control
+- Uses `asyncio.Condition` for asynchronous synchronization
+- Supports multiple concurrent readers and modifiers
+- Ensures atomic operations through state arrays
+
+### 2. Memory Management
+- Automatic memory reclamation based on reuse counting
+- Fixed-size buffer avoids dynamic memory allocation
+- Supports in-place data updates
+
+### 3. RPC Support
+- Intelligent RPC readiness detection
+- Dependency resolution based on data keys
+- Supports concurrent execution of multiple RPCs
+
+### 4. Data Prioritization
+- FIFO (First In, First Out) data processing strategy
+- Data sorting based on birth time
+- Ensures temporal consistency of training data
+
+## Summary
+
+AsyncIOSequenceBuffer is a core component in the AReaL system, achieving efficient asynchronous data buffering through carefully designed state management and concurrency control. Its main advantages include:
+
+1. **High Concurrency**: Supports concurrent access by multiple readers and writers
+2. **Memory Efficiency**: Optimizes memory usage through reuse counting and fixed-size buffers
+3. **Flexibility**: Supports dynamic data modification and various RPC operations
+4. **Reliability**: Comprehensive state checking and exception handling mechanisms
+
+This design enables AReaL to provide stable, efficient data management services in distributed reinforcement learning environments.
\ No newline at end of file
diff --git a/docs/buffer_class_diagram.puml b/docs/buffer_class_diagram.puml
new file mode 100644
index 0000000000..be468ff2e7
--- /dev/null
+++ b/docs/buffer_class_diagram.puml
@@ -0,0 +1,76 @@
+@startuml BufferClassDiagram
+
+!define RECTANGLE class
+
+class AsyncIOSequenceBuffer {
+    + rpcs: List<MFCDef>
+    + _lock: asyncio.Condition
+    + _is_being_put: ndarray
+    + _is_being_amended: ndarray  
+    + _is_being_read: ndarray
+    + _is_idle: ndarray
+    + _is_empty: ndarray
+    + _n_amenders: ndarray
+    + _n_readers: ndarray
+    + _ready_for_rpcs: ndarray
+    + _completed_rpc: ndarray
+    - __buffer: _TensorDictSequenceBuffer
+    
+    + __init__(rpcs, max_size)
+    + put_batch(samples, birth_times): async
+    + amend_batch(indices, samples): async
+    + get_batch_for_rpc(rpc): async
+    + _can_do_rpc(rpc): bool
+    + _assert_valid_indicator()
+    + put_batch_synced(samples)
+}
+
+class _TensorDictSequenceBuffer {
+    - __storage: List<_ReplayEntry>
+    - __has_keys: ndarray
+    - __keys: List<str>
+    - __reuses: int
+    
+    + __init__(keys, max_size, reuses)
+    + put_batch(indices, xs)
+    + amend_batch(indices, xs)
+    + get_batch(indices): List<_ReplayEntry>
+    + inspect_batch(indices): List<_ReplayEntry>
+    + pop_batch(indices)
+    + _update_has_keys(indices)
+    + _get_has_keys(indices)
+}
+
+class _ReplayEntry {
+    + reuses_left: int
+    + receive_time: float
+    + sample: SequenceSample
+}
+
+class SequenceSample {
+    + data: Dict
+    + seqlens: Dict
+    + keys: Set
+    + update_(other)
+    + gather(samples, keys): SequenceSample
+}
+
+class MFCDef {
+    + name: str
+    + n_seqs: int
+    + input_keys: Tuple
+    + output_keys: Tuple
+    + interface_type: ModelInterfaceType
+}
+
+class BufferFull {
+    <<Exception>>
+}
+
+AsyncIOSequenceBuffer *-- _TensorDictSequenceBuffer : contains
+_TensorDictSequenceBuffer o-- _ReplayEntry : stores
+_ReplayEntry *-- SequenceSample : contains
+AsyncIOSequenceBuffer --> MFCDef : uses
+AsyncIOSequenceBuffer ..> BufferFull : throws
+
+@enduml
\ No newline at end of file
diff --git a/docs/get_batch_for_rpc_sequence.puml b/docs/get_batch_for_rpc_sequence.puml
new file mode 100644
index 0000000000..65796efcd4
--- /dev/null
+++ b/docs/get_batch_for_rpc_sequence.puml
@@ -0,0 +1,51 @@
+@startuml GetBatchForRPCSequence
+
+participant "RPC Client" as RPC
+participant "AsyncIOSequenceBuffer" as Buffer
+participant "_TensorDictSequenceBuffer" as Internal
+participant "asyncio.Condition" as Lock
+
+RPC -> Buffer: get_batch_for_rpc(rpc)
+Buffer -> Lock: acquire()
+
+loop until can_do_rpc
+    Buffer -> Buffer: _can_do_rpc(rpc)
+    alt not ready
+        Buffer -> Lock: wait()
+    end
+end
+
+Buffer -> Buffer: _assert_valid_indicator()
+Buffer -> Buffer: find ready_indices for rpc
+Buffer -> Buffer: sort by birth_time (FIFO)
+Buffer -> Buffer: select n_seqs indices
+Buffer -> Buffer: set _is_idle[indices] = False
+Buffer -> Buffer: set _is_being_read[indices] = True
+Buffer -> Buffer: increment _n_readers[indices]
+Buffer -> Lock: release()
+
+Buffer -> Internal: get_batch(indices)
+Internal -> Internal: decrement reuses_left
+Internal -> Internal: identify entries with reuses_left = 0
+
+alt has entries to pop
+    Buffer -> Internal: pop_batch(pop_indices)
+    Internal -> Internal: clear storage and keys
+end
+
+Buffer -> Lock: acquire()
+Buffer -> Buffer: decrement _n_readers[indices]
+Buffer -> Buffer: update _is_being_read[indices]
+Buffer -> Buffer: update _is_idle[indices]
+Buffer -> Buffer: set _completed_rpc[indices, rpc_idx] = True
+Buffer -> Buffer: clean up popped indices
+Buffer -> Buffer: decrement _buf_size
+
+alt any indices become idle
+    Buffer -> Lock: notify(n_rpcs)
+end
+
+Buffer -> Lock: release()
+Buffer --> RPC: return (indices, SequenceSample)
+
+@enduml
\ No newline at end of file
diff --git a/docs/put_batch_sequence.puml b/docs/put_batch_sequence.puml
new file mode 100644
index 0000000000..17ab039e12
--- /dev/null
+++ b/docs/put_batch_sequence.puml
@@ -0,0 +1,34 @@
+@startuml PutBatchSequence
+
+participant Client
+participant "AsyncIOSequenceBuffer" as Buffer
+participant "_TensorDictSequenceBuffer" as Internal
+participant "asyncio.Condition" as Lock
+
+Client -> Buffer: put_batch(samples, birth_times)
+Buffer -> Lock: acquire()
+Buffer -> Buffer: _assert_valid_indicator()
+Buffer -> Buffer: find empty indices
+
+alt insufficient space
+    Buffer --> Client: raise BufferFull
+else sufficient space
+    Buffer -> Buffer: set _is_empty[indices] = False
+    Buffer -> Buffer: set _is_being_put[indices] = True
+    Buffer -> Lock: release()
+    
+    Buffer -> Internal: put_batch(indices, samples)
+    Internal -> Internal: create _ReplayEntry for each sample
+    
+    Buffer -> Lock: acquire()
+    Buffer -> Internal: _update_has_keys(indices)
+    Buffer -> Buffer: update _ready_for_rpcs
+    Buffer -> Buffer: set _is_being_put[indices] = False
+    Buffer -> Buffer: set _is_idle[indices] = True
+    Buffer -> Buffer: increment _buf_size
+    Buffer -> Lock: notify(n_rpcs)
+    Buffer -> Lock: release()
+    Buffer --> Client: return indices
+end
+
+@enduml
\ No newline at end of file