[SWE-Paddle] Add task PaddlePaddle__Paddle-65724

swe-paddle/tasks/PaddlePaddle__Paddle-65724/README.md

@@ -0,0 +1,53 @@
+# PaddlePaddle__Paddle-65724
+
+This directory converts Paddle PR #65724 into a SWE-Paddle community task candidate.
+
+## Source
+
+| Field | Value |
+| --- | --- |
+| Repo | `PaddlePaddle/Paddle` |
+| Issue | [48964](https://github.com/PaddlePaddle/Paddle/issues/48964) |
+| PR | [65724](https://github.com/PaddlePaddle/Paddle/pull/65724) |
+| PR title | `[Bugfix] fix dataloader when setting persistent_workers=True` |
+| Base commit | `9a4caad68bca019e85847eb99da57f060e01caa5` |
+| Gold commit | `b7c8439cd489e26c09c1db8b929285a96c64e3ed` |
+| Merged at | `2024-07-17` |
+| Task type | `bug_fix` |
+| Resource | CPU |
+
+## Summary
+
+修复 `DataLoader` 在 `persistent_workers=True` 且一个 epoch 被提前终止后，复用 workers 进入后续 epoch 时可能因 batch structure metadata 不一致而崩溃的问题。
+
+## Why This Is A Good SWE-Paddle Candidate
+
+- It is derived from a real merged BugFix PR linked to issue #48964.
+- The production change is limited to one Python file.
+- The failure concerns a practical benchmark workflow: breaking an epoch early while keeping workers persistent.
+- The verifier checks producer/consumer handoff and iterator reset behavior instead of matching source text.
+- The tests are deterministic and CPU-only; they do not require GPU training or long-running worker processes.
+- Existing FIFO structure restoration remains covered by P2P.
+
+## Files
+
+- `proposal.md`: candidate proposal for maintainer triage.
+- `instruction.md`: self-contained problem statement for the coding agent.
+- `solution/code.patch`: gold patch from the merged PR.
+- `tests/test.patch`: independent regression verifier.
+- `tests/test.sh`: minimal target test command.
+- `environment/README.md`: environment notes for reproduction.
+- `README.md`: task overview and verification entrypoint.
+
+## Verification
+
+```bash
+bash tests/test.sh
+```
+
+Expected behavior:
+
+- Applying `tests/test.patch` to `base_commit` keeps ordinary FIFO structure handoff green.
+- The producer/consumer handoff test and persistent-worker reset test fail on `base_commit`.
+- Applying both `tests/test.patch` and `solution/code.patch` makes all target tests pass.
+- The patched production file must match the Git blob from `gold_commit`.

swe-paddle/tasks/PaddlePaddle__Paddle-65724/environment/README.md

@@ -0,0 +1,40 @@
+# Environment Notes
+
+## Expected Environment
+
+- Repository: `PaddlePaddle/Paddle`
+- Base commit: `9a4caad68bca019e85847eb99da57f060e01caa5`
+- Gold commit: `b7c8439cd489e26c09c1db8b929285a96c64e3ed`
+- Resource: CPU
+- Python dependencies: PaddlePaddle, NumPy, pytest
+- Paddle rebuild required: no
+
+The verifier reads `python/paddle/io/dataloader/dataloader_iter.py` from the source checkout and loads the historical iterator classes through Python AST. Runtime dependencies come from the installed Paddle wheel.
+
+This design validates the exact Base and Gold Python implementation while avoiding a flaky end-to-end race that depends on OS process scheduling.
+
+## Run Order
+
+1. Check out the Base commit.
+2. Restore the exact Base blob for `python/paddle/io/dataloader/dataloader_iter.py`.
+3. Apply `tests/test.patch`.
+4. Run the FIFO P2P; it should pass.
+5. Run the thread handoff and reset-channel F2P tests; both should fail on Base.
+6. Apply `solution/code.patch`.
+7. Verify the target file blob matches the Gold commit.
+8. Run `bash tests/test.sh`; all verifier tests should pass.
+
+## Minimal Command
+
+```bash
+bash tests/test.sh
+```
+
+## Expected Matrix
+
+| State | FIFO P2P | Thread handoff F2P | Reset channel F2P | Full script |
+| --- | ---: | ---: | ---: | ---: |
+| Base + tests | PASS | FAIL | FAIL | FAIL |
+| Base + tests + solution | PASS | PASS | PASS | PASS |
+
+No GPU, model training, external service, or dataset download is required.

swe-paddle/tasks/PaddlePaddle__Paddle-65724/instruction.md

@@ -0,0 +1,29 @@
+# 修复 persistent workers 场景下提前结束 epoch 后的 DataLoader 崩溃
+
+## 详细描述
+
+当 `paddle.io.DataLoader` 同时启用多个 workers 和 `persistent_workers=True` 时，用户可能只消费一个 epoch 的前几个 batches，然后通过 `break` 提前结束该 epoch。随后复用同一个 DataLoader 进入下一个 epoch。
+
+在该工作流中，DataLoader 可能在恢复 batch 的 nested structure 时抛出 `IndexError: pop from empty list`，导致训练或 benchmark 在后续 epoch 中断。该问题通常与 background prefetch 同时发生，并可能在多次 epoch reuse 后出现。
+
+## 验收说明
+
+- 启用 `persistent_workers=True` 时，提前结束一个 epoch 后应能够继续复用同一个 DataLoader
+- 后续 epoch 返回的 nested batch structure 和数据内容应保持正确
+- 多次提前结束并重新开始 iteration 时，不应出现 `IndexError` 或其他 structure restoration 错误
+- `persistent_workers=False` 以及完整消费 epoch 的现有行为不得退化
+
+## 技术要求
+
+- 熟悉 Python threading 和 multiprocessing
+- 了解 Paddle DataLoader worker lifecycle、prefetch 和 iterator reset
+- 了解 producer/consumer synchronization 与 FIFO semantics
+- 了解 Paddle Python 单元测试开发流程
+
+## Acceptance Criteria
+
+- Reusing a persistent-worker DataLoader after an early epoch exit does not raise a structure-restoration error.
+- Batch structure metadata remains correctly paired with each produced batch.
+- Reset removes stale metadata without disconnecting active producers from consumers.
+- Existing non-persistent and fully consumed iteration behavior remains unchanged.
+- Do not satisfy the task by deleting tests, weakening assertions, or bypassing structure restoration.

swe-paddle/tasks/PaddlePaddle__Paddle-65724/proposal.md

@@ -0,0 +1,56 @@
+# Task Proposal: PaddlePaddle__Paddle-65724
+
+## 1. 来源信息
+
+- Instance ID：`PaddlePaddle__Paddle-65724`
+- Issue 链接：https://github.com/PaddlePaddle/Paddle/issues/48964
+- PR 链接：https://github.com/PaddlePaddle/Paddle/pull/65724
+- PR 标题：`[Bugfix] fix dataloader when setting persistent_workers=True`
+- `base_commit`：`9a4caad68bca019e85847eb99da57f060e01caa5`
+- merged 时间：`2024-07-17`
+- 你的身份：熟悉该模块的 contributor
+- 后续联系人：TBD
+
+## 2. 问题一句话
+
+修复 `DataLoader` 在启用 `persistent_workers=True`、提前结束 epoch 并复用 iterator 时，batch structure metadata 失配导致后续 iteration 崩溃的问题。
+
+## 3. 为什么适合作为 SWE-Paddle 样本
+
+- **真实性**：任务来自 issue #48964 和已合入的 Paddle BugFix PR，不是合成任务。
+- **代表性**：它覆盖 DataLoader lifecycle、background prefetch、persistent workers、producer/consumer synchronization 和 nested batch restoration。
+- **边界清楚**：production change 仅修改 `python/paddle/io/dataloader/dataloader_iter.py`。
+- **非平凡性**：来源 issue 的用户可观察现象是后续 epoch 在 `_restore_batch` 前出现 `IndexError: pop from empty list`。
+- **区分度信号**：verifier 将实际竞态约束拆解为可重复的 thread handoff 与 iterator reset invariants，避免依赖调度时序触发 flaky crash。
+
+## 4. 任务类型和标签
+
+- 任务类型：`bug_fix`
+- 执行后端：`cpu`
+- 设备范围：`cpu_only`
+- 模块标签：`[dataloader, persistent_workers, multiprocessing, threading, prefetch, batch_structure]`
+
+## 5. 验证思路
+
+- 目标测试命令：`bash tests/test.sh`
+- 目标测试文件：`test/legacy_test/test_dataloader_persistent_workers_structure.py`
+- 修复前预期：在 `base_commit` 上应用 `tests/test.patch` 后，ordinary FIFO structure handoff P2P 应 pass，background thread structure handoff 和 persistent-worker reset channel F2P 应 fail。
+- 修复后预期：继续应用 `solution/code.patch` 后，目标测试应 pass。
+- P2P 候选：ordinary FIFO structure handoff 用例可作为回归护栏。
+
+## 6. 环境与资源
+
+- 资源需求：CPU
+- Paddle 来源：`PaddlePaddle/Paddle` source checkout at `base_commit`
+- 是否能提供 Docker：暂无
+- patch 类型：Python-only production change
+- 环境建议：该样本涉及 sparse CPU kernel，不能只依赖已有 wheel；需要 source build 或等价的已编译环境
+- 最小测试命令：`bash tests/test.sh`
+- 是否有 oracle 日志：由 SWE-Paddle verifier 结果另行维护
+
+## 7. 风险自查
+
+- 泄露风险：`instruction.md` 描述 trigger condition、observable failure 和 expected behavior，不指出具体容器类型或修改行。
+- 环境风险：只加载目标 iterator classes，复用当前安装环境中的 Paddle dependencies。
+- flaky 风险：不依赖真实多进程调度时序；测试直接验证 producer/consumer 和 reset 的必要并发契约。
+- 拆分风险：该 PR 的目标集中在 persistent workers 场景下的 batch structure metadata handoff 与 iterator reset，适合作为一个样本。

swe-paddle/tasks/PaddlePaddle__Paddle-65724/solution/code.patch

@@ -0,0 +1,113 @@
+diff --git a/python/paddle/io/dataloader/dataloader_iter.py b/python/paddle/io/dataloader/dataloader_iter.py
+index 3ea19697d74b47ea269edbca4400ed81dbd889b7..94b4cd42c7f1c1b2fd99447c57953e7bd2e20e7d 100644
+--- a/python/paddle/io/dataloader/dataloader_iter.py
++++ b/python/paddle/io/dataloader/dataloader_iter.py
+@@ -26,6 +26,7 @@ import numpy as np
+ import paddle
+ from paddle import profiler
+ from paddle.base.framework import _current_expected_place, _set_expected_place
++from paddle.incubate import multiprocessing
+ from paddle.pir.core import datatype_to_vartype
+ from paddle.profiler.timer import benchmark
+ from paddle.profiler.utils import in_profiler_mode
+@@ -171,7 +172,7 @@ class _DataLoaderIterSingleProcess(_DataLoaderIterBase):
+         # only single process is used in single-process mode, we
+         # can record the data structure sequencely in a list without
+         # recording the send and recv index
+-        self._structure_infos = []
++        self._structure_infos = multiprocessing.Queue()
+
+         # NOTE: len(self._places) batch data compose as an output
+         # iteration, set blocking_queue can cache "self._prefetch_factor" iteration datas
+@@ -251,7 +252,7 @@ class _DataLoaderIterSingleProcess(_DataLoaderIterBase):
+
+             # flat batch and record structure infos
+             batch, structure = _flatten_batch(batch)
+-            self._structure_infos.append(structure)
++            self._structure_infos.put(structure)
+
+             if self._thread_done_event.is_set():
+                 break
+@@ -297,7 +298,8 @@ class _DataLoaderIterSingleProcess(_DataLoaderIterBase):
+                 data = core.eager.read_next_tensor_list(
+                     self._reader.read_next_list()[0]
+                 )
+-                data = _restore_batch(data, self._structure_infos.pop(0))
++                structure_info = self._structure_infos.get()
++                data = _restore_batch(data, structure_info)
+             else:
+                 # in static graph mode
+                 if self._return_list:
+@@ -305,7 +307,7 @@ class _DataLoaderIterSingleProcess(_DataLoaderIterBase):
+                     for i in range(len(data)):
+                         data[i] = data[i]._move_to_list()
+                     structs = [
+-                        self._structure_infos.pop(0)
++                        self._structure_infos.get()
+                         for _ in range(len(self._places))
+                     ]
+                     data = [_restore_batch(d, s) for d, s in zip(data, structs)]
+@@ -384,7 +386,7 @@ class _DataLoaderIterMultiProcess(_DataLoaderIterBase):
+         self._rcvd_idx = 0
+         self._batches_outstanding = 0
+         self._task_infos = {}
+-        self._structure_infos = []
++        self._structure_infos = multiprocessing.Queue()
+
+         # indices outstand as _outstanding_capacity at first, and
+         # blocking_queue capacity is also _outstanding_capacity.
+@@ -433,8 +435,6 @@ class _DataLoaderIterMultiProcess(_DataLoaderIterBase):
+         self._shutdown = False
+
+     def _init_workers(self):
+-        from paddle.incubate import multiprocessing
+-
+         # multiprocess worker and indice queue list initial as empty
+         self._workers = []
+         self._worker_status = []
+@@ -562,7 +562,8 @@ class _DataLoaderIterMultiProcess(_DataLoaderIterBase):
+         self._rcvd_idx = 0
+         self._batches_outstanding = 0
+         self._task_infos = {}
+-        self._structure_infos = []
++        while not self._structure_infos.empty():
++            self._structure_infos.get()
+
+         # set all worker status available
+         self._worker_status = [True] * self._num_workers
+@@ -690,7 +691,7 @@ class _DataLoaderIterMultiProcess(_DataLoaderIterBase):
+                 and len(self._task_infos[self._rcvd_idx]) == 3
+             ):
+                 info = self._task_infos.pop(self._rcvd_idx)
+-                self._structure_infos.append(info[2])
++                self._structure_infos.put(info[2])
+                 return info[1]
+
+             try:
+@@ -767,7 +768,7 @@ class _DataLoaderIterMultiProcess(_DataLoaderIterBase):
+                 if idx == self._rcvd_idx:
+                     if idx in self._task_infos:
+                         del self._task_infos[idx]
+-                    self._structure_infos.append(structure)
++                    self._structure_infos.put(structure)
+                     return batch
+                 else:
+                     self._task_infos[idx] += (batch, structure)
+@@ -838,14 +839,15 @@ class _DataLoaderIterMultiProcess(_DataLoaderIterBase):
+                 data = core.eager.read_next_tensor_list(
+                     self._reader.read_next_list()[0]
+                 )
+-                data = _restore_batch(data, self._structure_infos.pop(0))
++                structure_info = self._structure_infos.get()
++                data = _restore_batch(data, structure_info)
+             else:
+                 if self._return_list:
+                     data = self._reader.read_next_list()
+                     for i in range(len(data)):
+                         data[i] = data[i]._move_to_list()
+                     structs = [
+-                        self._structure_infos.pop(0)
++                        self._structure_infos.get()
+                         for _ in range(len(self._places))
+                     ]
+                     data = [_restore_batch(d, s) for d, s in zip(data, structs)]