diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-41202/README.md b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/README.md new file mode 100644 index 000000000..d6b12c4ce --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/README.md @@ -0,0 +1,44 @@ +# PaddlePaddle__Paddle-41202 + +This directory converts Paddle PR #41202 into a SWE-Paddle community task candidate. + +## Source + +| Field | Value | +| --- | --- | +| Repo | `PaddlePaddle/Paddle` | +| PR | [#41202](https://github.com/PaddlePaddle/Paddle/pull/41202) | +| PR title | `Add AutoTune to reader.py for DataLoader` | +| Base commit | `23d1b3e8ed8187bfb3bd926934dd6cc71e691e53` | +| Merged at | `2022-04-22T04:31:39Z` | +| Task type | `feature_enhancement` | +| Resource | CPU | + +## Summary + +Allow DataLoader to sample a small portion of a dataset and automatically select a suitable worker count, while preserving the existing behavior when automatic tuning is disabled or unavailable on the current platform. + +## Why This Is A Good SWE-Paddle Candidate + +- The problem reflects a common DataLoader configuration cost: a fixed worker count may underuse the CPU or add unnecessary process overhead. +- The change spans configuration, dataset sampling, worker-count search, ordinary batch samplers, and distributed batch samplers. +- The merged PR includes a dedicated upstream test file covering enabled, disabled, and distributed-sampler scenarios. +- The behavior can be verified with a small synthetic dataset on CPU without downloading data or loading model weights. + +## Files + +- `proposal.md`: candidate proposal for maintainer triage. +- `instruction.md`: self-contained problem statement for the coding agent. +- `solution/code.patch`: gold production patch from the merged PR. +- `tests/test.patch`: exact upstream test patch from the merged PR. +- `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` should keep the disabled-mode P2P valid while the enabled auto-tune scenarios fail; applying both `tests/test.patch` and `solution/code.patch` should make the complete upstream test file pass. diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-41202/environment/README.md b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/environment/README.md new file mode 100644 index 000000000..85e55b8f9 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/environment/README.md @@ -0,0 +1,27 @@ +# Environment Notes + +This candidate is part of the SWE-Paddle community task set. + +## Expected Environment + +- Repository: `PaddlePaddle/Paddle` +- Base commit: `23d1b3e8ed8187bfb3bd926934dd6cc71e691e53` +- Resource: CPU +- GPU required: no +- Build path: use a compatible installed CPU Paddle wheel as the runtime carrier; the verifier may overlay the checkout's target Python behavior to bridge historical module-layout differences. + +## Run Order + +1. Check out `PaddlePaddle/Paddle` at the base commit. +2. Apply `tests/test.patch`. +3. Run `bash tests/test.sh`; the enabled auto-tune scenarios should fail before the fix. +4. Apply `solution/code.patch`. +5. Run `bash tests/test.sh` again; the complete upstream test file should pass after the gold patch. + +## Minimal Test Command + +```bash +bash tests/test.sh +``` + +The verifier is responsible for deriving stable F2P and P2P node IDs from repeated runs. diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-41202/instruction.md b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/instruction.md new file mode 100644 index 000000000..98da83d79 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/instruction.md @@ -0,0 +1,19 @@ +# 为 DataLoader 增加 worker 数量自动调优能力 + +## 详细描述 + +DataLoader 的 `num_workers` 通常需要根据机器和数据处理速度反复尝试。设置得太小,数据准备可能跟不上训练;设置得太大,又会增加不必要的进程开销。现在只能由用户手动选择,换一台机器后往往还要重新调整。 + +所以需要为 DataLoader 增加一个可选的自动调优能力。开启后,DataLoader 使用少量样本比较不同 worker 数量的读取速度,并选出更合适的配置。普通 DataLoader 和使用 DistributedBatchSampler 的 DataLoader 都应能够正常创建和使用。 + +## 验收说明 + +* 开启自动调优后,DataLoader 能够评估不同 worker 配置并应用合适的 worker 数量。 +* 调优过程只使用有限数量的样本,并支持普通 batch sampler 和 `DistributedBatchSampler`。 +* 关闭自动调优或平台不支持时,DataLoader 原有的 worker 配置行为保持不变。 + +## 技术要求 + +* 熟悉 Python 和多进程数据读取。 +* 熟悉 PaddlePaddle DataLoader、Dataset 和 batch sampler。 +* 了解数据读取性能测试及不同平台的兼容处理。 diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-41202/proposal.md b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/proposal.md new file mode 100644 index 000000000..48e09aa4f --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/proposal.md @@ -0,0 +1,57 @@ +# Task Proposal: PaddlePaddle__Paddle-41202 + +## 1. 来源信息 + +- Instance ID:`PaddlePaddle__Paddle-41202` +- PR 链接:https://github.com/PaddlePaddle/Paddle/pull/41202 +- PR 标题:`Add AutoTune to reader.py for DataLoader` +- `base_commit`:`23d1b3e8ed8187bfb3bd926934dd6cc71e691e53` +- merged 时间:`2022-04-22T04:31:39Z` +- 你的身份:熟悉该模块的 contributor +- 后续联系人:TBD + +## 2. 问题一句话 + +DataLoader 缺少按实际读取耗时自动选择 worker 数量的能力,用户只能反复手动调整 `num_workers`。 + +## 3. 为什么适合作为 SWE-Paddle 样本 + +- **真实性**:数据读取速度受数据处理逻辑和机器 CPU 数量影响,固定 worker 数量经常不是合适配置。 +- **代表性**:任务覆盖配置开关、抽样数据集、普通与分布式 batch sampler,以及不同平台的兼容行为。 +- **边界清楚**:只处理 DataLoader worker 数量的自动选择,不改变数据内容、batch 结构或训练计算。 +- **非平凡性**:需要比较多个候选 worker 数量的读取开销,并正确处理抽样范围、提前停止和 sampler 重建。 +- **环境友好性**:来源 PR 提供了只使用随机小数据集的 Python 单测,不依赖网络、真实模型或 GPU。 + +## 4. 任务类型和标签 + +- 任务类型:`feature_enhancement` +- 执行后端:`cpu` +- 设备范围:`cpu_only` +- 模块标签:`[dataloader, multiprocessing, performance_tuning, batch_sampler]` + +## 5. 验证思路 + +- 目标测试命令:`bash tests/test.sh` +- 目标测试文件:`python/paddle/fluid/tests/unittests/test_dataloader_autotune.py` +- 修复前预期:关闭 auto-tune 时的 P2P 保持通过;普通 DataLoader 和 DistributedBatchSampler 的启用场景均因 Base 不具备 auto-tune 支持而失败;完整测试脚本失败。 +- 修复后预期:P2P、两个启用场景以及完整上游测试文件全部通过。 +- P2P 候选:`TestAutoTune::test_dataloader_disable_autotune`。 +- F2P 候选:`TestAutoTune::test_dataloader_use_autotune` 和 `TestAutoTune::test_distributer_batch_sampler_autotune`。 +- 测试来源:`tests/test.patch` 与 PR 合入时新增的测试文件 diff 完全一致,不改写上游断言。 + +## 6. 环境与资源 + +- 资源需求:CPU +- Paddle 来源:`PaddlePaddle/Paddle` source checkout at `base_commit` +- 是否能提供 Docker:暂无 +- patch 类型:Python-only +- 环境建议:使用可导入的 CPU Paddle wheel 作为运行载体;对于历史 `paddle.fluid.reader` 与当前 wheel 的模块差异,由 verifier 从 checkout 源码提取目标函数和类并执行真实控制流。 +- 最小测试命令:`bash tests/test.sh` +- 是否有 oracle 日志:由 SWE-Paddle verifier 结果另行维护 + +## 7. 风险自查 + +- 泄露风险:instruction 只描述用户需要的自动选择行为和兼容要求,不说明 Gold patch 的类名、循环结构或内部阈值。 +- 环境风险:PR 较早,当前 Paddle wheel 可能不再提供 `paddle.fluid.reader`;cross script 使用受控 compatibility overlay,不覆盖已安装 Paddle 文件。 +- flaky 风险:上游测试使用小型内存数据集;verifier 限制搜索用 CPU 数量,避免随机器核数扩大测试范围。 +- 拆分风险:低。production 修改集中在 DataLoader worker 自动选择这一项功能。 diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-41202/solution/code.patch b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/solution/code.patch new file mode 100644 index 000000000..b5f18ceab --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/solution/code.patch @@ -0,0 +1,178 @@ +diff --git a/python/paddle/fluid/reader.py b/python/paddle/fluid/reader.py +index 0f5f217442135f03cab88a6c5e6fdf3894f670c6..841c58821d7a16c171f00dd96d30471f376e9c38 100644 +--- a/python/paddle/fluid/reader.py ++++ b/python/paddle/fluid/reader.py +@@ -18,11 +18,13 @@ import six + import numpy as np + import threading + import paddle ++import time ++ + from .framework import Program, Variable, program_guard, default_main_program, default_startup_program, _non_static_mode, cpu_places, _current_expected_place, _in_eager_without_dygraph_check + from .executor import global_scope + from .data_feeder import DataFeeder, BatchedTensorProvider + from .multiprocess_utils import multiprocess_queue_set, CleanupFuncRegistrar, _cleanup_mmap, _cleanup, _set_SIGCHLD_handler +-from .dataloader import BatchSampler, Dataset, IterableDataset ++from .dataloader import BatchSampler, Dataset, IterableDataset, Subset + from .dataloader.dataloader_iter import _DataLoaderIterSingleProcess, _DataLoaderIterMultiProcess, _DatasetKind, default_collate_fn + from .dataloader.batch_sampler import _InfiniteIterableSampler + from .layers.io import monkey_patch_reader_methods, _copy_reader_var_, double_buffer +@@ -36,10 +38,8 @@ import warnings + import os + import multiprocessing + import signal +- + # NOTE: queue has a different name in python2 and python3 + import queue +- + # NOTE: [ avoid hanging & failed quickly ] These value is used in getting data from another process + QUEUE_GET_TIMEOUT = 60 + +@@ -49,6 +49,16 @@ data_loader_unique_name_generator = UniqueNameGenerator() + + KEEP_DATA_LOADER_ORDER = True + USE_PINNED_MEMORY = None ++# AutoTune Flags ++USE_AUTOTUNE = False ++TUNING_STEPS = 500 ++ ++ ++def set_autotune_config(use_autotune, tuning_steps=500): ++ global USE_AUTOTUNE ++ USE_AUTOTUNE = use_autotune ++ global TUNING_STEPS ++ TUNING_STEPS = tuning_steps + + + def keep_data_loader_order(*args): +@@ -143,6 +153,122 @@ class DataLoaderBase(object): + return arr + + ++class AuToTune(object): ++ def __init__(self, loader): ++ self.loader = loader ++ self.max_num_worker = multiprocessing.cpu_count() / 2 ++ ++ def __call__(self): ++ # use default loader ++ if (not USE_AUTOTUNE) or (not self.need_autotune()): ++ return self.loader.num_workers ++ ++ # get autotune loader ++ auto_tune_loader = self.get_autotune_loader() ++ if auto_tune_loader is None: ++ return self.loader.num_workers ++ ++ # pick the best num_workers ++ auto_tune_start = time.time() ++ logging.debug("========= DataLoader Auto Tune =========") ++ logging.debug("User config for DataLoader: " + str( ++ self.loader.num_workers)) ++ best_num_workers = 0 ++ min_cost = float("inf") ++ logging.debug("Tuning Range for num_workers: 0 ~ " + str( ++ self.max_num_worker)) ++ num_workers = 0 ++ while num_workers < self.max_num_worker: ++ auto_tune_loader.num_workers = num_workers ++ avg_cost = self.evaluate_reader_cost(auto_tune_loader) ++ if min_cost * 0.75 > avg_cost: ++ min_cost = avg_cost ++ best_num_workers = num_workers ++ else: ++ update_num = self.is_best(auto_tune_loader, best_num_workers, ++ min_cost, self.max_num_worker) ++ if update_num == best_num_workers: ++ break ++ else: ++ best_num_workers = update_num ++ logging.debug("num_workers: " + str(num_workers) + " avg_cost: " + ++ str(avg_cost)) ++ num_workers += 2 ++ logging.info("auto_tune dataLoader best_num_workers: " + str( ++ best_num_workers)) ++ logging.debug("AutoTuning Cost for DataLoader: " + str(time.time( ++ ) - auto_tune_start) + ' seconds') ++ ++ # tune the default loader's num_workers ++ return best_num_workers ++ ++ def need_autotune(self): ++ if (sys.platform == 'darwin' or sys.platform == 'win32'): ++ return False ++ else: ++ return True ++ ++ def get_sub_dataset(self, dataset, batch_size): ++ num_samples = min(batch_size * TUNING_STEPS, len(dataset)) ++ sub_dataset = Subset(dataset, indices=list(range(num_samples))) ++ return sub_dataset ++ ++ def get_autotune_loader(self): ++ loader = self.loader ++ batch_size = self.loader.batch_sampler.batch_size ++ if isinstance(self.loader.batch_sampler, ++ paddle.io.DistributedBatchSampler): ++ dataset = self.loader.batch_sampler.dataset ++ sub_dataset = self.get_sub_dataset(dataset, batch_size) ++ loader.batch_sampler = paddle.io.DistributedBatchSampler( ++ dataset=sub_dataset, ++ batch_size=batch_size, ++ num_replicas=self.loader.batch_sampler.nranks, ++ rank=self.loader.batch_sampler.local_rank, ++ shuffle=self.loader.batch_sampler.shuffle, ++ drop_last=self.loader.batch_sampler.drop_last) ++ elif isinstance(self.loader.batch_sampler, paddle.io.BatchSampler): ++ dataset = self.loader.batch_sampler.sampler.data_source ++ sub_dataset = self.get_sub_dataset(dataset, batch_size) ++ loader.batch_sampler = paddle.io.BatchSampler( ++ dataset=sub_dataset, ++ batch_size=batch_size, ++ drop_last=self.loader.batch_sampler.drop_last) ++ else: ++ loader = None ++ return loader ++ ++ def evaluate_reader_cost(self, reader): ++ costs = [] ++ avg_cost = 0 ++ start = time.time() ++ for i, data in enumerate(reader): ++ costs.append(time.time() - start) ++ start = time.time() ++ if len(costs) > 2: ++ avg_cost = sum(costs[2:]) / len(costs[2:]) ++ else: ++ avg_cost = sum(costs[0:]) / len(costs[0:]) ++ return avg_cost ++ ++ def is_best(self, reader, best_workers, best_time, num_work_boundary): ++ step = 0 ++ num_workers = best_workers + 1 ++ boundary = 1 ++ while num_workers < num_work_boundary and step < 5: ++ self.loader.num_workers = num_workers ++ time = self.evaluate_reader_cost(reader) ++ logging.debug("for back num_workers: " + str(num_workers) + ++ " avg_cost: " + str(time)) ++ step += 1 ++ if (time < best_time * 0.70 * boundary): ++ return num_workers ++ else: ++ num_workers += 1 ++ boundary *= 0.80 ++ return best_workers ++ ++ + class DataLoader(object): + """ + DataLoader prodives an iterator which iterates given dataset +@@ -409,6 +535,7 @@ class DataLoader(object): + + self._persistent_workers = persistent_workers + self._iterator = None ++ self.num_workers = AuToTune(self).__call__() + + def __len__(self): + if self.dataset_kind == _DatasetKind.ITER: diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-41202/tests/test.patch b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/tests/test.patch new file mode 100644 index 000000000..9303fefea --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/tests/test.patch @@ -0,0 +1,82 @@ +diff --git a/python/paddle/fluid/tests/unittests/test_dataloader_autotune.py b/python/paddle/fluid/tests/unittests/test_dataloader_autotune.py +new file mode 100755 +index 0000000000000000000000000000000000000000..a140bb5c79c93b2de2b65a7c65643b3abe376c0f +--- /dev/null ++++ b/python/paddle/fluid/tests/unittests/test_dataloader_autotune.py +@@ -0,0 +1,76 @@ ++# Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserved. ++# ++# Licensed under the Apache License, Version 2.0 (the "License"); ++# you may not use this file except in compliance with the License. ++# You may obtain a copy of the License at ++# ++# http://www.apache.org/licenses/LICENSE-2.0 ++# ++# Unless required by applicable law or agreed to in writing, software ++# distributed under the License is distributed on an "AS IS" BASIS, ++# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. ++# See the License for the specific language governing permissions and ++# limitations under the License. ++ ++from __future__ import print_function ++import unittest ++import numpy as np ++ ++import paddle ++import paddle.nn as nn ++from paddle.io import Dataset, DataLoader, BatchSampler, SequenceSampler ++from paddle.fluid.reader import set_autotune_config ++import sys ++ ++ ++class RandomDataset(Dataset): ++ def __init__(self, num_samples): ++ self.num_samples = num_samples ++ ++ def __getitem__(self, idx): ++ image = np.random.random([10]).astype('float32') ++ label = np.random.randint(0, 10 - 1, (1, )).astype('int64') ++ return image, label ++ ++ def __len__(self): ++ return self.num_samples ++ ++ ++class SimpleNet(nn.Layer): ++ def __init__(self): ++ super(SimpleNet, self).__init__() ++ self.fc = nn.Linear(10, 10) ++ ++ def forward(self, image): ++ return self.fc(image) ++ ++ ++class TestAutoTune(unittest.TestCase): ++ def setUp(self): ++ self.batch_size = 1 ++ self.dataset = RandomDataset(10) ++ ++ def test_dataloader_use_autotune(self): ++ set_autotune_config(True, 1) ++ loader = DataLoader( ++ self.dataset, batch_size=self.batch_size, num_workers=0) ++ ++ def test_dataloader_disable_autotune(self): ++ set_autotune_config(False) ++ loader = DataLoader( ++ self.dataset, batch_size=self.batch_size, num_workers=2) ++ if (sys.platform == 'darwin' or sys.platform == 'win32'): ++ self.assertEqual(loader.num_workers, 0) ++ else: ++ self.assertEqual(loader.num_workers, 2) ++ ++ def test_distributer_batch_sampler_autotune(self): ++ set_autotune_config(True, 1) ++ batch_sampler = paddle.io.DistributedBatchSampler( ++ self.dataset, batch_size=self.batch_size) ++ loader = DataLoader( ++ self.dataset, batch_sampler=batch_sampler, num_workers=2) ++ ++ ++if __name__ == '__main__': ++ unittest.main() diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-41202/tests/test.sh b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/tests/test.sh new file mode 100644 index 000000000..833b19f46 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-41202/tests/test.sh @@ -0,0 +1,4 @@ +#!/usr/bin/env bash + +set -euo pipefail +python -m pytest python/paddle/fluid/tests/unittests/test_dataloader_autotune.py -q