صف محدود در ThreadPoolExecutor؛ مدیریت کارهای پایتون

نام فایل: bounded_workers.py؛ این فایل را کامل و بدون حذف سطرها ذخیره کنید.

from collections import deque
from concurrent.futures import ThreadPoolExecutor
from itertools import islice


def bounded_map(function, values, *, workers=4, pending=8):
    """Yield ordered results while retaining at most pending submitted jobs."""
    if workers < 1 or pending < 1:
        raise ValueError('workers and pending must be positive')
    source = iter(values)
    with ThreadPoolExecutor(max_workers=workers) as pool:
        jobs = deque(pool.submit(function, value)
                     for value in islice(source, pending))
        try:
            while jobs:
                yield jobs.popleft().result()
                try:
                    value = next(source)
                except StopIteration:
                    continue
                jobs.append(pool.submit(function, value))
        finally:
            for job in jobs:
                job.cancel()

این فایل همهٔ وابستگی‌های خود را وارد می‌کند و bounded_map را صادر می‌کند. سقف pending هنگام ارسال رعایت می‌شود؛ پایان executor به معنای توقف اجباری نخ فعال نیست.

نام فایل: python314_map.py؛ این فایل را کامل و بدون حذف سطرها ذخیره کنید.

import inspect
from concurrent.futures import ThreadPoolExecutor


def main():
    if 'buffersize' not in inspect.signature(ThreadPoolExecutor.map).parameters:
        print('SKIP: this example requires Python 3.14 or newer')
        return
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(lambda n: n * n, range(100), buffersize=8))
    assert results == [n * n for n in range(100)]
    print(sum(results))  # 328350


if __name__ == '__main__':
    main()

با python python314_map.py اجرا کنید. نتیجهٔ نسخهٔ پشتیبانی‌شده 328350 است؛ در نسخهٔ قدیمی‌تر به‌روشنی SKIP چاپ می‌شود.

نام فایل: test_bounded_workers.py؛ این فایل را کامل و بدون حذف سطرها ذخیره کنید.

import unittest
from bounded_workers import bounded_map


class BoundedMapTests(unittest.TestCase):
    def test_order_and_bounded_input_consumption(self):
        consumed = []
        def source():
            for value in range(20):
                consumed.append(value)
                yield value
        stream = bounded_map(lambda n: n * n, source(), workers=2, pending=4)
        self.assertEqual(next(stream), 0)
        self.assertEqual(len(consumed), 4)
        self.assertEqual(next(stream), 1)
        self.assertEqual(len(consumed), 5)
        self.assertEqual(list(stream), [n * n for n in range(2, 20)])

    def test_worker_error_is_not_silently_dropped(self):
        def fail(value):
            if value == 2:
                raise RuntimeError('invalid item')
            return value
        with self.assertRaisesRegex(RuntimeError, 'invalid item'):
            list(bounded_map(fail, range(10), pending=3))

    def test_invalid_capacity(self):
        for options in ({'workers': 0}, {'pending': 0}):
            with self.assertRaises(ValueError):
                list(bounded_map(str, [1], **options))


if __name__ == '__main__':
    unittest.main()

با python -m unittest test_bounded_workers.py هر سه آزمون را اجرا کنید. تعداد ورودی‌های خوانده‌شده نیز سنجیده می‌شود تا محدودبودن صف فقط یک ادعا نباشد.

نام فایل: run_workers.py؛ این فایل را کامل و بدون حذف سطرها ذخیره کنید.

import argparse
import json
from bounded_workers import bounded_map


parser = argparse.ArgumentParser()
parser.add_argument('--count', type=int, default=100)
parser.add_argument('--workers', type=int, default=4)
parser.add_argument('--pending', type=int, default=8)
args = parser.parse_args()
if args.count < 0:
    parser.error('--count must not be negative')
values = list(bounded_map(lambda n: n * n, range(args.count),
                          workers=args.workers, pending=args.pending))
print(json.dumps({'count': len(values), 'sum': sum(values)}))

اجرای python run_workers.py --workers 2 --pending 4 باید count برابر 100 و sum برابر 328350 بدهد. ظرفیت صفر و تعداد منفی پذیرفته نمی‌شود.