diff --git a/.gitignore b/.gitignore
index b1912e65ce..1305da50f8 100644
--- a/.gitignore
+++ b/.gitignore
@@ -178,4 +178,5 @@ outputs/
 tests/
 local/
 **/rollout_data/
-**/buffer_stats/
\ No newline at end of file
+**/buffer_stats/
+*.out
diff --git a/scripts/run-deepseek-r1-distill-qwen-1.5B.sh b/scripts/run-deepseek-r1-distill-qwen-1.5B.sh
new file mode 100644
index 0000000000..426424befe
--- /dev/null
+++ b/scripts/run-deepseek-r1-distill-qwen-1.5B.sh
@@ -0,0 +1,150 @@
+#!/bin/bash
+
+# for rerun the task
+pkill -9 sglang
+sleep 3
+ray stop --force
+pkill -9 ray
+pkill -9 python
+sleep 3
+pkill -9 ray
+pkill -9 python
+
+set -ex
+
+# will prevent ray from buffering stdout/stderr
+export PYTHONBUFFERED=16
+
+SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
+source "${SCRIPT_DIR}/models/qwen2.5-1.5B.sh"
+
+CKPT_ARGS=(
+   --hf-checkpoint /root/DeepSeek-R1-Distill-Qwen-1.5B
+   --ref-load /root/DeepSeek-R1-Distill-Qwen-1.5B_torch_dist
+   --load /root/DeepSeek-R1-Distill-Qwen-1.5B_slime/
+   --save /root/DeepSeek-R1-Distill-Qwen-1.5B_slime/
+   --save-interval 20
+)
+
+ROLLOUT_ARGS=(
+   --prompt-data /root/dapo-math-17k/dapo-math-17k.jsonl
+   --input-key prompt
+   --label-key label
+   --apply-chat-template
+   --rollout-shuffle
+   --rm-type deepscaler
+   --num-rollout 3000
+   --rollout-batch-size 128
+   --n-samples-per-prompt 8
+   --rollout-max-response-len 8192
+   --rollout-temperature 0.8
+
+   --global-batch-size 1024
+   --balance-data
+   --sampling-batch-size 128
+
+   # --partial-rollout
+   # --partial-rollout-min-response-length 20
+   # --partial-rollout-min-tokens 8
+   # --partial-rollout-mix-ratio 0.75 
+   # --over-sampling-filter-path slime.rollout.filter_hub.over_sampling_filters.sort_by_reward_std
+   # --over-sampling-filter-input-size 192
+   # --dynamic-sampling-filter-path slime.rollout.filter_hub.dynamic_sampling_filters.check_reward_nonzero_std
+)
+
+EVAL_ARGS=(
+   --eval-interval 20
+   --eval-prompt-data aime /root/aime-2024/aime-2024.jsonl
+   --n-samples-per-eval-prompt 16
+   --eval-max-response-len 16384
+   --eval-top-p 0.7
+)
+
+PERF_ARGS=(
+   --tensor-model-parallel-size 1
+   --sequence-parallel
+   --pipeline-model-parallel-size 1
+   --context-parallel-size 1
+   --expert-model-parallel-size 1
+   --expert-tensor-parallel-size 1
+
+   --recompute-granularity full
+   --recompute-method uniform
+   --recompute-num-layers 1
+
+   # --micro-batch-size 1
+   --use-dynamic-batch-size
+   --max-tokens-per-gpu 9216
+)
+
+GRPO_ARGS=(
+   --advantage-estimator grpo
+   --use-kl-loss
+   --kl-loss-coef 0.00
+   --kl-loss-type low_var_kl
+   --kl-coef 0.00
+   --entropy-coef 0.00
+   --eps-clip 0.2
+   --eps-clip-high 0.28
+)
+
+OPTIMIZER_ARGS=(
+   --optimizer adam
+   --lr 1e-6
+   --lr-decay-style constant
+   --weight-decay 0.1
+   --adam-beta1 0.9
+   --adam-beta2 0.98
+)
+
+WANDB_ARGS=(
+   # --use-wandb
+   # --wandb-project slime-dev
+   # --wandb-group qwen2.5-1.5B-test
+   # --wandb-key ${WANDB_KEY}
+)
+
+SGLANG_ARGS=(
+   --rollout-num-gpus-per-engine 1
+   --sglang-mem-fraction-static 0.7
+)
+
+MISC_ARGS=(
+   # default dropout in megatron is 0.1
+   --attention-dropout 0.0
+   --hidden-dropout 0.0
+   # should be good for model performance
+   --accumulate-allreduce-grads-in-fp32
+   --attention-softmax-in-fp32
+   # need to comment this when using model with MLA
+   --attention-backend flash
+)
+
+# launch the master node of ray in container
+export MASTER_ADDR=${MASTER_ADDR:-"127.0.0.1"}
+export MASTER_PORT=${MASTER_PORT:-"12345"}
+ray start --head --node-ip-address ${MASTER_ADDR} --num-gpus 8 --disable-usage-stats
+
+ray job submit --address="http://127.0.0.1:8265" \
+   --runtime-env-json='{
+     "env_vars": {
+        "PYTHONPATH": "/root/Megatron-LM/",
+        "CUDA_DEVICE_MAX_CONNECTIONS": "1",
+        "NCCL_CUMEM_ENABLE": "0"
+     }
+   }' \
+   -- python3 train.py \
+   --actor-num-nodes 1 \
+   --actor-num-gpus-per-node 8 \
+   --colocate \
+   ${MODEL_ARGS[@]} \
+   ${CKPT_ARGS[@]} \
+   ${ROLLOUT_ARGS[@]} \
+   ${OPTIMIZER_ARGS[@]} \
+   ${GRPO_ARGS[@]} \
+   ${DISTRIBUTED_ARGS[@]} \
+   ${WANDB_ARGS[@]} \
+   ${PERF_ARGS[@]} \
+   ${EVAL_ARGS[@]} \
+   ${SGLANG_ARGS[@]} \
+   ${MISC_ARGS[@]}
diff --git a/scripts/run-qwen3-4B.sh b/scripts/run-qwen3-4B.sh
index 185eedbf88..85a5c883ef 100644
--- a/scripts/run-qwen3-4B.sh
+++ b/scripts/run-qwen3-4B.sh
@@ -33,9 +33,7 @@ ROLLOUT_ARGS=(
    --label-key label
    --apply-chat-template
    --rollout-shuffle
-
    --rm-type deepscaler
-
    --num-rollout 3000
    --rollout-batch-size 32
    --n-samples-per-prompt 8
@@ -44,6 +42,15 @@ ROLLOUT_ARGS=(
 
    --global-batch-size 256
    --balance-data
+   --sampling-batch-size 32
+   
+   # --partial-rollout
+   # --partial-rollout-min-response-length 20
+   # --partial-rollout-min-tokens 8
+   # --partial-rollout-mix-ratio 0.75 
+   # --over-sampling-filter-path slime.rollout.filter_hub.over_sampling_filters.sort_by_reward_std
+   # --over-sampling-filter-input-size 48
+   # --dynamic-sampling-filter-path slime.rollout.filter_hub.dynamic_sampling_filters.check_reward_nonzero_std
 )
 
 EVAL_ARGS=(
diff --git a/slime/backends/megatron_utils/model.py b/slime/backends/megatron_utils/model.py
index b84147c861..c00215ae35 100644
--- a/slime/backends/megatron_utils/model.py
+++ b/slime/backends/megatron_utils/model.py
@@ -17,6 +17,7 @@
 from megatron.training.global_vars import get_args
 from megatron.training.training import get_model
 
+
 from .checkpoint import load_checkpoint, save_checkpoint
 from .data import get_batch, set_local_storage
 from .loss import get_log_probs_and_entropy, policy_loss_func
diff --git a/slime/ray/buffer.py b/slime/ray/buffer.py
index ffba7eeeb4..94d8a2afad 100644
--- a/slime/ray/buffer.py
+++ b/slime/ray/buffer.py
@@ -27,7 +27,7 @@ def convert_samples_to_train_data(samples: list[Sample]):
         "tokens": [sample.tokens for sample in samples],
         "response_lengths": [sample.response_length for sample in samples],
         "rewards": [sample.reward for sample in samples],
-        "truncated": [1 if sample.truncated else 0 for sample in samples],
+        "truncated": [1 if sample.status == Sample.Status.TRUNCATED else 0 for sample in samples],
     }
     if samples[0].loss_mask:
         train_data["loss_masks"] = []
@@ -49,7 +49,8 @@ def __init__(self, args):
         self.args = args
 
         self.buffer = []
-        self.buffer_filter = load_function(self.args.buffer_filter_path)
+        self.read_function = load_function(self.args.buffer_read_function_path)
+        self.write_function = load_function(self.args.buffer_write_function_path)
 
         self.train_data_pool = {}
         self.eval_data_pool = {}
@@ -123,13 +124,14 @@ def _init_wandb(self):
 
         wandb.init(**wandb_config, settings=wandb.Settings(mode="shared"))
 
-    async def get_samples(self, num_samples) -> list[Sample]:
+    async def get_samples(self, num_samples: int, rollout_info: dict[str, Any]) -> list[Sample]:
         """
         Return num_samples samples
         """
 
-        samples = await self._get_samples_from_buffer(num_samples)
+        samples = await self._get_samples_from_buffer(num_samples, rollout_info)
         num_samples -= len(samples)
+
         assert num_samples % self.args.n_samples_per_prompt == 0
         num_prompts = num_samples // self.args.n_samples_per_prompt
 
@@ -161,20 +163,27 @@ async def get_samples(self, num_samples) -> list[Sample]:
                 )
                 self.sample_index += 1
                 samples.append(sample)
-
-        assert len(samples) == num_samples
         return samples
 
-    async def _get_samples_from_buffer(self, num_samples) -> list[Sample]:
+    async def _get_samples_from_buffer(self, num_samples: int, rollout_info: dict[str, Any]) -> list[Sample]:
         if len(self.buffer) == 0 or num_samples == 0:
             return []
-        samples = self.buffer_filter(self.buffer, num_samples)
+
+        samples = self.read_function(self.args, self.buffer, num_samples, rollout_info)
         return samples
 
-    async def add_samples(self, samples: list[Sample]):
-        # TODO: we can save some partial rollout data here.
-        assert len(samples) % self.args.n_samples_per_prompt == 0
-        self.buffer.extend(samples)
+    async def add_samples(self, samples: list[Sample], rollout_info: dict[str, Any]):
+        """
+        Add a sample group to buffer.
+        """
+        if not samples:
+            return
+        assert (
+            len(samples) % self.args.n_samples_per_prompt == 0
+        ), f"Buffer add_samples got {len(samples)} samples, expected {self.args.n_samples_per_prompt}"
+
+        self.write_function(self.args, self.buffer, samples, rollout_info)
+        print(f"Buffer size after adding samples: {len(self.buffer)} (adding {len(samples)} samples)", flush=True)
 
     def generate(self, rollout_id, evaluation=False):
         if not evaluation and self.args.load_debug_rollout_data:
diff --git a/slime/rollout/buffer_hub/fifo.py b/slime/rollout/buffer_hub/fifo.py
new file mode 100644
index 0000000000..866907e236
--- /dev/null
+++ b/slime/rollout/buffer_hub/fifo.py
@@ -0,0 +1,23 @@
+from typing import Any
+
+from slime.utils.types import Sample
+
+
+# Simple operations
+def push_end(args, buffer: list[list[Sample]], samples: list[Sample], rollout_info: dict[str, Any]):
+    """
+    Simply append the samples to the end of the buffer.
+    """
+    buffer.append(samples)
+
+
+def pop_first(args, buffer: list[list[Sample]], num_samples: int, rollout_info: dict[str, Any]):
+    """
+    Try to pop the first `num_samples` from the buffer.
+    """
+    num_to_pop = min(len(buffer), num_samples // args.n_samples_per_prompt)
+    return_samples = []
+    for i in range(num_to_pop):
+        return_samples.extend(buffer[i])
+    del buffer[:num_to_pop]
+    return return_samples
diff --git a/slime/rollout/buffer_hub/partial_fifo.py b/slime/rollout/buffer_hub/partial_fifo.py
new file mode 100644
index 0000000000..62a605a421
--- /dev/null
+++ b/slime/rollout/buffer_hub/partial_fifo.py
@@ -0,0 +1,97 @@
+from typing import Any
+
+from slime.utils.types import Sample
+
+
+# partial filters
+def clear_sample_response(sample: Sample):
+    """
+    Clear the response, start_rollout_id, and status of a sample.
+    """
+    sample.response = None
+    sample.metadata["start_rollout_id"] = None
+    sample.status = Sample.Status.PENDING
+
+
+def valid_partial_sample(args, sample: Sample):
+    """
+    Check if a sample is a valid partial sample.
+    """
+    if sample.response and len(sample.response.strip()) < args.partial_rollout_min_response_length:
+        return False
+    if sample.response_length and sample.response_length < args.partial_rollout_min_tokens:
+        return False
+    return True
+
+
+def filter_partial_samples(args, samples: list[Sample], rollout_info: dict[str, Any]):
+    """
+    Update the partial samples.
+    If partial samples are too short, we clear the response and status.
+    """
+    for sample in samples:
+        if sample.status != Sample.Status.PENDING and not valid_partial_sample(args, sample):
+            clear_sample_response(sample)
+    # TODO(jiajun): Staleness may be handled here using rollout_info.
+
+
+def partial_push_end(args, buffer: list[list[Sample]], samples: list[Sample], rollout_info: dict[str, Any]):
+    """
+    Push the samples to the end of the buffer.
+    """
+    filter_partial_samples(args, samples, rollout_info)
+    for sample in samples:
+        # reset partial sample's index without start_rollout_id
+        if sample.status != Sample.Status.PENDING and sample.metadata.get("start_rollout_id", None) == None:
+            sample.metadata["start_rollout_id"] = rollout_info["rollout_id"]
+    buffer.append(samples)
+
+
+def partial_pop_first(args, buffer: list[list[Sample]], num_samples: int, rollout_info: dict[str, Any]):
+    """
+    Filter for partial rollout.
+    This function pops the front `num_samples` from the buffer.
+    Partial samples are prioritized.
+    """
+    for samples in buffer:
+        filter_partial_samples(args, samples, rollout_info)
+
+    # Group samples by n_samples_per_prompt
+    partial_groups_idx = []
+    new_groups_idx = []
+
+    for i in range(0, len(buffer)):
+        samples = buffer[i]
+        # Check if all samples in the group are PENDING
+        all_pending = all(sample.status == Sample.Status.PENDING for sample in samples)
+        if all_pending:
+            new_groups_idx.append(i)
+        else:
+            partial_groups_idx.append(i)
+
+    # Calculate how many partial groups we can take
+    num_groups = num_samples // args.n_samples_per_prompt
+    num_partial_groups = min(int(num_groups * args.partial_rollout_mix_ratio), len(partial_groups_idx))
+    selected_partial_groups_idx = partial_groups_idx[:num_partial_groups]
+
+    # Select new groups to fill the remaining quota
+    num_new_groups = min(num_groups - num_partial_groups, len(new_groups_idx))
+    selected_new_groups_idx = new_groups_idx[:num_new_groups]
+
+    # Collect all selected samples
+    selected_samples = []
+    selected_idx = [False] * len(buffer)
+    new_buffer = []
+    for i in selected_partial_groups_idx:
+        selected_samples.extend(buffer[i])
+        selected_idx[i] = True
+    for i in selected_new_groups_idx:
+        selected_samples.extend(buffer[i])
+        selected_idx[i] = True
+    for i in range(len(buffer)):
+        if not selected_idx[i]:
+            new_buffer.append(buffer[i])
+
+    buffer = new_buffer
+
+    return selected_samples
diff --git a/slime/rollout/filter_hub/buffer_filters.py b/slime/rollout/filter_hub/buffer_filters.py
deleted file mode 100644
index ef5f92548f..0000000000
--- a/slime/rollout/filter_hub/buffer_filters.py
+++ /dev/null
@@ -1,10 +0,0 @@
-def pop_first(buffer, num_samples):
-    samples = []
-    for _ in range(num_samples):
-        if buffer:
-            samples.append(buffer.pop(0))
-    return samples
-
-
-def get_newest_samples(buffer, num_samples):
-    return buffer[-num_samples:]
diff --git a/slime/rollout/filter_hub/over_sampling_filters.py b/slime/rollout/filter_hub/over_sampling_filters.py
index f00dfe40b1..2417287149 100644
--- a/slime/rollout/filter_hub/over_sampling_filters.py
+++ b/slime/rollout/filter_hub/over_sampling_filters.py
@@ -1,18 +1,19 @@
 import torch
-from slime.utils.types import Samples
+from slime.utils.types import Sample
 
 
 __all__ = ["sort_by_reward_std"]
 
 
-def sort_by_reward_std(args, samples: list[Samples], **kwargs):
+def sort_by_reward_std(args, samples: list[Sample], **kwargs):
     args.n_samples_per_prompt
     samples_with_std = []
     for i in range(0, len(samples), args.n_samples_per_prompt):
         batch = samples[i : i + args.n_samples_per_prompt]
-        rewards = [item[3] for item in batch]
+        rewards = [item.reward for item in batch]
         std = torch.tensor(rewards, dtype=torch.float).std()
-        for j in range(args.n_samples_per_prompt):
-            samples_with_std.append(batch[i + j], torch.tensor(rewards, std))
+        for sample in batch:
+            samples_with_std.append((sample, std))
+    # python sort is stable, so the order of samples with the same std is preserved
     samples_with_std.sort(key=lambda x: x[1], reverse=True)
     return [item[0] for item in samples_with_std]
diff --git a/slime/rollout/sglang_example.py b/slime/rollout/sglang_example.py
index 4f69d150ea..92f01a4ffc 100644
--- a/slime/rollout/sglang_example.py
+++ b/slime/rollout/sglang_example.py
@@ -1,6 +1,10 @@
 import asyncio
 import copy
-from dataclasses import dataclass
+import json
+from collections import defaultdict
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Counter, DefaultDict
 
 from tqdm import tqdm
 from transformers import AutoTokenizer
@@ -18,8 +22,21 @@
 
 @dataclass
 class GenerateState:
-    remaining_batch_size: int = 0
-    pendings: set = None
+    remaining_batch_size: int = 0  # the number of samples that are done/running/pending
+    pendings: set = None  # the set of pending tasks
+    num_stats: DefaultDict[str, Counter] = field(default_factory=lambda: defaultdict(Counter))
+
+    def bump(self, group: str, value: str, n: int):
+        self.num_stats[group][value] += n
+
+    def stats_to_string(self) -> str:
+        flat = {f"{g}_{k}_samples": v for g, ctr in self.num_stats.items() for k, v in ctr.items()}
+        return json.dumps(flat, ensure_ascii=False)
+
+
+def update_num_stats(state: GenerateState, group: str, value: Any, n: int = 1):
+    vstr = value.name.lower() if isinstance(value, Enum) else str(value)
+    state.bump(group, vstr, n)
 
 
 TOKENIZER = None
@@ -37,33 +54,69 @@ async def generate(args, sample: Sample, sampling_params) -> Sample:
         )
 
     url = f"http://{args.sglang_router_ip}:{args.sglang_router_port}/generate"
+
+    assert (
+        sample.status == Sample.Status.PENDING or sample.status == Sample.Status.ABORTED
+    ), f"Sample status is {sample.status}"
+    is_partial_sample = sample.status == Sample.Status.ABORTED
+    # Handle partial rollout samples: continue generation from existing response
+    if is_partial_sample:
+        # Continue generation from partial response
+        input_text = sample.prompt + sample.response
+    else:
+        # Regular generation from prompt
+        input_text = sample.prompt
+
     payload = {
-        "text": sample.prompt,
+        "text": input_text,
         "sampling_params": sampling_params,
     }
-    while True:
+
+    max_retries = 60
+    retry_count = 0
+    while retry_count < max_retries:
         try:
             async with SEMAPHORE:
                 output = await post(url, payload, use_http2=args.use_http2)
         except Exception as e:
-            print(f"Error: {e}, retrying...")
+            retry_count += 1
+            print(f"Error: {e}, retrying... (attempt {retry_count}/{max_retries})")
+            if retry_count >= max_retries:
+                print(f"Max retries ({max_retries}) reached, failing...")
+                raise e
             await asyncio.sleep(1)
             continue
         break
 
     prompt_tokens_ids = TOKENIZER(sample.prompt, add_special_tokens=False)["input_ids"]
-    response_token_ids = TOKENIZER(output["text"], add_special_tokens=False)["input_ids"]
 
+    if is_partial_sample:
+        # For partial rollout: combine existing response with new generation
+        sample.response = sample.response + output["text"]
+    else:
+        # Regular generation
+        sample.response = output["text"]
+
+    response_token_ids = TOKENIZER(sample.response, add_special_tokens=False)["input_ids"]
     sample.tokens = prompt_tokens_ids + response_token_ids
     sample.response_length = len(response_token_ids)
-    sample.truncated = output["meta_info"]["finish_reason"]["type"] == "length"
-    sample.response = output["text"]
-    sample.aborted = output["meta_info"]["finish_reason"]["type"] == "abort"
+    match output["meta_info"]["finish_reason"]["type"]:
+        case "length":
+            sample.status = Sample.Status.TRUNCATED
+        case "abort":
+            sample.status = Sample.Status.ABORTED
+        case "stop":
+            sample.status = Sample.Status.COMPLETED
 
     return sample
 
 
 async def generate_and_rm(args, sample: Sample, sampling_params: dict, evaluation=False) -> Sample:
+    # For samples with existing response, check if they're complete
+    if sample.status == Sample.Status.COMPLETED or sample.status == Sample.Status.TRUNCATED:
+        assert sample.response is not None and sample.reward is not None
+        return sample
+
     # generate
     if args.custom_generate_function_path is not None:
         custom_generate_func = load_function(args.custom_generate_function_path)
@@ -71,7 +124,7 @@ async def generate_and_rm(args, sample: Sample, sampling_params: dict, evaluatio
     else:
         sample = await generate(args, sample, sampling_params)
 
-    if sample.aborted:
+    if sample.status == Sample.Status.ABORTED:
         return sample
 
     # for the rm that need the whole group, we will not do the rm here
@@ -88,7 +141,7 @@ async def generate_and_rm(args, sample: Sample, sampling_params: dict, evaluatio
     return sample
 
 
-async def generate_rollout_async(args, rollout_id, data_buffer) -> list[Sample]:
+async def generate_rollout_async(args, rollout_id: int, data_buffer) -> list[Sample]:
     """An example to implement the generate_rollout function for an rule based rm rollout generation.
 
     Args:
@@ -97,7 +150,7 @@ async def generate_rollout_async(args, rollout_id, data_buffer) -> list[Sample]:
         data_buffer: the data buffer to store the generated samples
 
     Returns:
-        list[Sample]: a list of samples generated by the rollout
+        list[Sample]: a list of samples generated by the rollout, the length of the list is exactly the same as the `rollout_batch_size`
     """
     assert args.rollout_global_dataset
 
@@ -113,21 +166,36 @@ async def generate_rollout_async(args, rollout_id, data_buffer) -> list[Sample]:
         spaces_between_special_tokens=False,
     )
 
-    # if over_sampling is set, the sampling batch size can be larger than
-    # the required rollout batch size
-    sampling_batch_size = (
-        args.over_sampling_batch_size if args.over_sampling_batch_size is not None else args.rollout_batch_size
-    )
-    # get data from the global_dataset
-    samples = await data_buffer.get_samples(sampling_batch_size * args.n_samples_per_prompt)
+    # sampling_batch_size refers to the number of samples to get at a time.
+    # if the number of valid samples obtained is insufficient to support rollout, start the next round of sampling.
+    # redundant samples: aborted samples and excess completed and truncated samples.
+    # for non-partial rollout with dynamic filter, on-policy is required and all redundant samples are dropped, so the sampling_batch_size should not be too large.
+    # for partial rollout, redundant samples are stored in the buffer and will be used in the next round of sampling.
+
+    if args.sampling_batch_size is not None:
+        sampling_batch_size = args.sampling_batch_size
+    else:
+        sampling_batch_size = args.rollout_batch_size
 
     state = GenerateState(
         remaining_batch_size=0,
         pendings=set(),
     )
 
-    def submit_generate_tasks(samples):
+    dynamic_filter = None
+    if args.dynamic_sampling_filter_path is not None:
+        # dynamic filter is used to filter out samples that is not suitable for training
+        dynamic_filter = load_function(args.dynamic_sampling_filter_path)
+    over_sampling_filter = None
+    if args.over_sampling_filter_path is not None:
+        assert args.over_sampling_filter_input_size is not None
+        # over sampling filter ensures over_sampling_filter_input_size samples are rollout. And pick rollout_batch_size samples from them.
+        over_sampling_filter = load_function(args.over_sampling_filter_path)
+
+    def submit_generate_tasks(samples: list[Sample]):
         for sample in samples:
+            if sample.status == Sample.Status.PENDING:
+                assert len(sample.metadata) == 0, f"Sample {sample} has metadata {sample.metadata}"
             state.pendings.add(
                 asyncio.create_task(
                     generate_and_rm(
@@ -140,28 +208,48 @@ def submit_generate_tasks(samples):
             )
         state.remaining_batch_size += len(samples) // args.n_samples_per_prompt
 
-    # submit the generation requests.
-    submit_generate_tasks(samples)
-
-    do_dynamic_sampling = args.over_sampling_batch_size and args.dynamic_sampling_filter_path is not None
-    # load multiple time, so the filter should have no side effect, which should be rational?
-    if do_dynamic_sampling:
-        assert args.dynamic_sampling_filter_path is not None
-        dynamic_sampling_filter = load_function(args.dynamic_sampling_filter_path)
-    elif args.over_sampling_batch_size is not None:
-        assert args.over_sampling_filter_path is not None
-        over_sampling_filter = load_function(args.over_sampling_filter_path)
-
     data_group = {}
     data = []
-    do_print = True
 
-    # when doing dynamic sampling, we will use the first rollout_batch_size samples.
+    async def abort_rollout():
+        print(f"DEBUG: Sending abort. Current data: {len(data)}, pending tasks: {len(state.pendings)}", flush=True)
+        try:
+            response = await get(
+                f"http://{args.sglang_router_ip}:{args.sglang_router_port}/list_workers", use_http2=args.use_http2
+            )
+            print(f"DEBUG: List workers: {response}", flush=True)
+        except Exception as e:
+            print(f"Error: {e}, Failed to get list_workers", flush=True)
+            return []
+
+        for url in response["urls"]:
+            # abort all the requests
+            # NOTE: Using empty string as rid to abort ALL requests by startswith() match
+            print(f"Abort request for {url}", flush=True)
+            await post(f"{url}/abort_request", {"rid": ""}, use_http2=False)
+
+    have_aborted = False
+
+    # target_data_size is the total number of valid samples to get
+    # if over_sampling_filter_input_size is set, we will use it as the target data size, otherwise, we will use the rollout_batch_size
     target_data_size = (
-        args.rollout_batch_size if do_dynamic_sampling else sampling_batch_size
+        args.over_sampling_filter_input_size if over_sampling_filter is not None else args.rollout_batch_size
     ) * args.n_samples_per_prompt
+    # rollout_info is used for sending info to the buffer
+    rollout_info = {
+        "rollout_id": rollout_id,
+    }
+
     pbar = tqdm(total=target_data_size, desc="Rollout generation")
     while len(data) < target_data_size:
+        while state.remaining_batch_size < target_data_size // args.n_samples_per_prompt:
+            # get samples from the buffer and submit the generation requests.
+            samples = await data_buffer.get_samples(sampling_batch_size * args.n_samples_per_prompt, rollout_info)
+            submit_generate_tasks(samples)
+            for sample in samples:
+                update_num_stats(state, "input", sample.status)
+
+        # wait for the generation to finish
         done, state.pendings = await asyncio.wait(state.pendings, return_when=asyncio.FIRST_COMPLETED)
         # Always finish all done tasks. This will make the code of partial rollout cleaner.
         # The assumption here is that group_rm is not too slow.
@@ -174,9 +262,10 @@ def submit_generate_tasks(samples):
                 data_group[group_index] = []
             data_group[group_index].append(sample)
 
-            if do_print:
-                print([sample.prompt + sample.response], flush=True)
-                do_print = False
+            assert (
+                sample.status == Sample.Status.COMPLETED or sample.status == Sample.Status.TRUNCATED
+            ), f"Sample {sample.index} has status {sample.status}, but should be completed or truncated. Rollout {rollout_id}, Sample prompt: {sample.prompt}, Sample response: {sample.response}"
+            update_num_stats(state, "output", sample.status)
 
             if not len(data_group[group_index]) == args.n_samples_per_prompt:
                 # wait for the data_group for this prompt finishing
@@ -192,67 +281,74 @@ def submit_generate_tasks(samples):
                     else:
                         sample.reward = rewards[i]
 
-            if do_dynamic_sampling:
-                # the group is ready
-                if dynamic_sampling_filter(args, data_group[group_index]):
-                    # When having enough samples, don't add to data.
-                    if len(data) == target_data_size:
-                        continue
-                    data.extend(data_group[group_index])
-                    del data_group[group_index]
-                    pbar.update(args.n_samples_per_prompt)
-                else:
-                    # Delete the invalid samples, don't use them in partial rollout.
-                    del data_group[group_index]
-                    state.remaining_batch_size -= 1
-                    if state.remaining_batch_size < args.rollout_batch_size:
-                        print(
-                            f"Remaining batch size not enough, add {sampling_batch_size} prompts, "
-                            f"sample response: {[sample.prompt + sample.response]}"
-                        )
-                        new_samples = await data_buffer.get_samples(sampling_batch_size * args.n_samples_per_prompt)
-                        submit_generate_tasks(new_samples)
-            else:
-                # if not dynamic sampling, we will just add the samples to the data
+            if dynamic_filter is not None and not dynamic_filter(args, data_group[group_index]):
+                # Delete the invalid samples, don't use them in partial rollout.
+                del data_group[group_index]
+                state.remaining_batch_size -= 1
+                update_num_stats(state, "excluded", "dynamic_sampling_filter", args.n_samples_per_prompt)
+                continue
+
+            # add the samples to the data
+            if len(data) < target_data_size:
                 data.extend(data_group[group_index])
-                pbar.update(args.n_samples_per_prompt)
                 del data_group[group_index]
+                pbar.update(args.n_samples_per_prompt)
+
+            # When having enough samples, try abort the rollout and continue
+            if len(data) >= target_data_size and not have_aborted:
+                await abort_rollout()
+                have_aborted = True
+
     pbar.close()
 
-    print(f"Got {len(data)} samples, sample response: {[sample.prompt + sample.response]}")
+    print(f"[DEBUG] Rollout {rollout_id}: Got {len(data)} samples", flush=True)
 
-    if do_dynamic_sampling:
-        response = await get(
-            f"http://{args.sglang_router_ip}:{args.sglang_router_port}/list_workers", use_http2=args.use_http2
-        )
-        for url in response["urls"]:
-            # abort all the requests
-            print(f"Abort request for {url}")
-            await post(f"{url}/abort_request", {"rid": ""}, use_http2=False)
+    # there are still some unfinished requests, abort them
+    if state.pendings:
+        if not have_aborted:
+            await abort_rollout()
+            have_aborted = True
 
+    if args.partial_rollout:
+        # put unfinished samples to data group
         while state.pendings:
             done, state.pendings = await asyncio.wait(state.pendings, return_when=asyncio.FIRST_COMPLETED)
             for task in done:
                 sample = task.result()
+                update_num_stats(state, "output", sample.status)
+
                 group_index = sample.index // args.n_samples_per_prompt
                 if group_index not in data_group:
                     data_group[group_index] = []
                 data_group[group_index].append(sample)
 
-    for group_index, samples in data_group.items():
-        assert (
-            len(samples) == args.n_samples_per_prompt
-        ), f"Got {len(samples)} samples, expected {args.n_samples_per_prompt}"
-        if args.partial_rollout:
-            data_buffer.add_samples(samples)
-
-    assert len(data) == target_data_size, f"Got {len(data)} samples, expected {target_data_size}"
-
-    if not do_dynamic_sampling and args.over_sampling_batch_size is not None:
+        # try cache unfinished samples and excess valid samples back to buffer
+        for group_index, samples in data_group.items():
+            assert (
+                len(samples) == args.n_samples_per_prompt
+            ), f"Got {len(samples)} samples, expected {args.n_samples_per_prompt}"
+            cached_samples = []
+            for sample in samples:
+                update_num_stats(state, "cached", sample.status)
+                cached_samples.append(sample)
+
+            if len(cached_samples) > 0:
+                # Add cached samples back to buffer for next iteration
+                await data_buffer.add_samples(cached_samples, rollout_info)
+                print(f"[DEBUG] Rollout {rollout_id}: Cached {len(cached_samples)} samples back to buffer", flush=True)
+
+    if over_sampling_filter is not None:
+        update_num_stats(
+            state, "excluded", "over_sampling_filter", len(data) - args.rollout_batch_size * args.n_samples_per_prompt
+        )
         data = over_sampling_filter(args, data)[: args.rollout_batch_size * args.n_samples_per_prompt]
     else:
         data.sort(key=lambda sample: sample.index)
 
+    print(f"[DEBUG] Rollout {rollout_id}: {state.stats_to_string()}", flush=True)
+    assert (
+        len(data) == args.rollout_batch_size * args.n_samples_per_prompt
+    ), f"Got {len(data)} samples, expected {args.rollout_batch_size * args.n_samples_per_prompt}"
     return data
 
 
@@ -344,7 +440,7 @@ async def eval_rollout_single_dataset(args, rollout_id, name, path):
     return {
         name: {
             "rewards": [sample.reward for sample in data],
-            "truncated": [sample.truncated for sample in data],
+            "truncated": [sample.status == Sample.Status.TRUNCATED for sample in data],
         }
     }
 
diff --git a/slime/utils/arguments.py b/slime/utils/arguments.py
index 5b1052e300..6318fb27c5 100644
--- a/slime/utils/arguments.py
+++ b/slime/utils/arguments.py
@@ -194,22 +194,28 @@ def add_rollout_arguments(parser):
                 ),
             )
 
-            # over sampling
+            # sampling
             parser.add_argument(
-                "--over-sampling-batch-size",
+                "--sampling-batch-size",
                 type=int,
                 default=None,
                 help=(
-                    "The batch size for over sampling. "
-                    "There are 2 cases for over sampling: "
-                    "1. If `over_sampling_batch_size` is set, and `dynamic_sampling_filter_path` is set, "
-                    "we will do dynamic sampling as in DAPO, in which, we will first sample `over_sampling_batch_size` of prompts, "
-                    "use the function in `dynamic_sampling_filter_path` to check if the responses of the prompt is valid, "
-                    "e.g. not all correct or all wrong. And if there are not enough remaining prompts, "
-                    "we will take another `over_sampling_batch_size` of prompts. When there are enough valid prompts, we will abort the ongoing sampling."
-                    "2. If `over_sampling_batch_size` is set, and `over_sampling_filter_path` is set, "
-                    "the first `rollout_batch_size` of `over_sampling_batch_size` will be selected as the result of the prompt. "
-                    "The `over_sampling_filter_path` should be able to sort prompts by its responses and rewards."
+                    "This defines the granularity of the sampling batch in the rollout function. "
+                    "When the number of available samples falls below the target, a sampling "
+                    "operation of size sampling_batch_size will be triggered."
+                    "Regardless of whether partial rollout is used or filters are applied, "
+                    "the sampling granularity is always determined by this value. "
+                    "If this value is None, rollout_batch_size will be used as the default sampling_batch_size."
+                ),
+            )
+            parser.add_argument(
+                "--over-sampling-filter-input-size",
+                type=int,
+                default=None,
+                help=(
+                    "This is the input size for the over sampling filter."
+                    "This value will replace the rollout_batch_size as target batch size "
+                    "(number of complete, valid samples to be generated) when the over sampling filter is applied."
                 ),
             )
             parser.add_argument(
@@ -217,26 +223,24 @@ def add_rollout_arguments(parser):
                 type=str,
                 default=None,
                 help=(
-                    "Path to the over-sampling filter function. "
-                    "It should be able to sort prompts by its responses and rewards"
-                    "When --over-sampling-filter-path is set, the first `rollout_batch_size` of "
-                    "`over_sampling_batch_size` will be selected as the result of the prompt. "
-                    "You could use `slime.rollout.filter_hub.oversampling_sampling_filters.sort_by_reward_std` as an example."
+                    "This parameter is used with the over_sampling_filter_input_size. "
+                    "The over sampling filter is applied only after enough data has been generated."
+                    "You could use `slime.rollout.filter_hub.over_sampling_filters.sort_by_reward_std` as an example."
                 ),
             )
-            # dynamic sampling
             parser.add_argument(
                 "--dynamic-sampling-filter-path",
                 type=str,
                 default=None,
                 help=(
-                    "Path to the dynamic sampling filter function. "
-                    "It should be able to judge whether the result of a prompt should be selected or not. "
-                    "When --dynamic-sampling-filter-path is set, the first `rollout_batch_size` that satisfy the filter "
-                    "will be selected as the result of the prompt and --over-sampling-filter-path will be ignored. "
+                    "This is the filter function for dynamic sampling. "
+                    "It should be able to judge whether the result of a prompt should be selected or not."
+                    "We will do dynamic filter for sampling as in DAPO. e.g. not all correct or all wrong samples."
                     "You could use `slime.rollout.filter_hub.dynamic_sampling_filters.check_reward_nonzero_std` as an example."
                 ),
             )
+
+            # partial rollout
             parser.add_argument(
                 "--partial-rollout",
                 action="store_true",
@@ -247,6 +251,32 @@ def add_rollout_arguments(parser):
                     "This is useful for long responses."
                 ),
             )
+            parser.add_argument(
+                "--partial-rollout-min-response-length",
+                type=int,
+                default=10,
+                help=(
+                    "Minimum response length (in characters) for a sample to be considered for partial rollout recycling. "
+                    "Shorter responses will be discarded."
+                ),
+            )
+            parser.add_argument(
+                "--partial-rollout-min-tokens",
+                type=int,
+                default=5,
+                help=(
+                    "Minimum number of tokens in the response for a sample to be considered for partial rollout recycling."
+                ),
+            )
+            parser.add_argument(
+                "--partial-rollout-mix-ratio",
+                type=float,
+                default=0.3,
+                help=(
+                    "Maximum ratio of partial rollout samples to use in each batch. "
+                    "Value between 0.0 and 1.0. Default 0.3 means up to 30% partial samples."
+                ),
+            )
 
             parser.add_argument(
                 "--custom-generate-function-path",
@@ -259,16 +289,25 @@ def add_rollout_arguments(parser):
             )
 
             parser.add_argument(
-                "--buffer-filter-path",
+                "--buffer-read-function-path",
                 type=str,
-                default="slime.rollout.filter_hub.buffer_filters.pop_first",
+                default="slime.rollout.buffer_hub.fifo.pop_first",
                 help=(
                     "Path to the buffer filter function. "
                     "It should be able to select the samples in the buffer. "
                     "The function should take a list of samples and return a list of samples."
                 ),
             )
-
+            parser.add_argument(
+                "--buffer-write-function-path",
+                type=str,
+                default="slime.rollout.buffer_hub.fifo.push_end",
+                help=(
+                    "Path to the buffer write function. "
+                    "It should be able to write the samples to the buffer. "
+                    "The function should take a list of samples and write them to the buffer."
+                ),
+            )
             # update weight
             parser.add_argument(
                 "--update-weight-buffer-size",
@@ -810,14 +849,6 @@ def parse_args(add_custom_arguments=None):
     if args.eps_clip_high is None:
         args.eps_clip_high = args.eps_clip
 
-    if args.over_sampling_batch_size is not None:
-        assert (
-            args.over_sampling_batch_size >= args.rollout_batch_size
-        ), "over_sampling_batch_size must be greater than rollout_batch_size"
-        assert (
-            args.dynamic_sampling_filter_path is not None or args.over_sampling_filter_path is not None
-        ), "over_sampling_batch_size must be used with dynamic_sampling_filter_path or over_sampling_filter_path"
-
     if args.eval_reward_key is None:
         args.eval_reward_key = args.reward_key
 
@@ -870,6 +901,10 @@ def parse_args(add_custom_arguments=None):
     if args.vocab_size and not args.padded_vocab_size:
         args.padded_vocab_size = _vocab_size_with_padding(args.vocab_size, args)
 
+    if args.partial_rollout and args.buffer_read_function_path is None:
+        args.buffer_read_function_path = "slime.rollout.buffer_hub.partial_fifo.partial_pop_first"
+        args.buffer_write_function_path = "slime.rollout.buffer_hub.partial_fifo.partial_push_end"
+
     # placeholders
     args.seq_length = 4096
     args.max_position_embeddings = args.seq_length
diff --git a/slime/utils/data.py b/slime/utils/data.py
index 875e530963..736e043adb 100644
--- a/slime/utils/data.py
+++ b/slime/utils/data.py
@@ -41,7 +41,7 @@ def __init__(
                     Sample(
                         prompt=prompt,
                         label=data[label_key] if label_key is not None else None,
-                        metadata=data.get(metadata_key, None),
+                        metadata=data.get(metadata_key) or {},
                     )
                 )
 
diff --git a/slime/utils/types.py b/slime/utils/types.py
index 54da03be6c..0a8d1bdd9c 100644
--- a/slime/utils/types.py
+++ b/slime/utils/types.py
@@ -1,5 +1,7 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from enum import Enum
 from typing import Optional
+
 import torch
 
 
@@ -13,12 +15,18 @@ class Sample:
     response: Optional[str] = None
     tokens: Optional[list[int]] = None
     response_length: Optional[int] = None
-    truncated: Optional[bool] = None
     reward: Optional[float] = None
     loss_mask: Optional[list[int]] = None
-    metadata: Optional[dict] = None
+    metadata: dict = field(default_factory=dict)
     version: int = 0
-    aborted: bool = False
+
+    class Status(Enum):
+        PENDING = "pending"
+        COMPLETED = "completed"
+        TRUNCATED = "truncated"
+        ABORTED = "aborted"
+
+    status: Status = Status.PENDING
 
 
 @dataclass