[akashchamp]

Fixes #385 and #412.

With timeout or soft_timeout set on the pool, map()/starmap() crash the TimeoutHandler thread: MapResult keeps one accept time and worker pid per item, so the handler ends up doing list + int. The thread crash calls os._exit(1), so the whole process dies. imap() hits the same thing with an AttributeError because IMapIterator has no _time_accepted at all.

This times each running map() chunk on its own, the same way a single apply_async() task is timed: a soft timeout signals the worker running that chunk, and a hard timeout fails the map with TimeLimitExceeded and kills that worker. Finished chunks stop counting, so the timeout applies per chunk, not to the whole map. on_soft_timeout/on_hard_timeout take an optional pid and still work with just job, which is how Celery's asynpool calls them. imap() results are skipped since they don't record accept times.

Verification (Linux, Python 3.12):

• Before: Pool(2, timeout=2).map(time.sleep, [0.1, 6]) logged Thread 'TimeoutHandler' crashed: TypeError('can only concatenate list (not "int") to list') and the interpreter exited with status 1; imap died the same way with the AttributeError.
• After: timeout=2 raises TimeLimitExceeded(2) after about 4s, soft_timeout=2 raises SoftTimeLimitExceeded, and imap returns its results.
• Added a unit test for the handler with a MapResult (one finished chunk, one running) and an IMapIterator in the cache, plus an end-to-end map() hard-timeout test. pytest t/unit: 77 passed, 51 skipped. flake8 shows nothing new on the touched files; codespell is clean.

[auvipy]

this and some other changes seems to be changing the public API.... arent they? if so I think we have to update documentations and include this changes to a minor release like 4.4.0

[akashchamp]

The only signature changes are an optional pid=None on TimeoutHandler.on_soft_timeout/on_hard_timeout and optional soft_timeout/timeout keywords on MapResult.__init__. Existing callers (including Celery's asynpool, which calls them with just job) behave exactly as before, so I don't think anything breaks, and the rest (_running_chunks, _chunks_done) is private. Happy to add a changelog entry, and whether it goes out as 4.4.0 or a patch release is your call.

[auvipy]

please fix the merge conflicts

[akashchamp]

The merge conflicts are resolved; the branch now contains current main. I also added a 4.4.0 (unreleased) entry to CHANGES.txt describing the optional pid argument on TimeoutHandler.on_soft_timeout/on_hard_timeout and the optional soft_timeout/timeout keywords on MapResult, pushed in 19178f7. Whether it ships as 4.4.0 or a patch release is your call.

[auvipy]

@GangEunzzang can you review this?

[GangEunzzang]

The crash fix works. On main, map()/starmap()/imap() on a pool with timeout or soft_timeout kill the process (TypeError/AttributeError in the TimeoutHandler). With this branch the same calls raise TimeLimitExceeded/SoftTimeLimitExceeded or return, and a randomized run of 144 maps and starmaps with timeout=1 gave the expected result every time (main crashed on every seed). Celery's prefork path is unchanged: its concurrency/worker unit tests and a real prefork worker with soft and hard limits behave the same as on main.

Two things I ran into:

1. Only the first chunk that hits the hard limit gets killed. That fails the map and removes it from the cache, so the other chunks of the same map, running or still queued, are no longer timed and run to completion:

with Pool(2, timeout=1) as p:
    r = p.map_async(time.sleep, [25] * 4, chunksize=1)
    r.get()                                  # TimeLimitExceeded after ~1s
    p.apply_async(time.sleep, (0.1,)).get()  # returns after ~26s

The same four sleeps sent with apply_async() are all killed, and the 0.1s task returns after ~3.5s.

2. MapResult.worker_pids() still returns the workers of finished chunks, so killing a worker fails any other map that ran a chunk on it earlier, with WorkerLostError about 11s later:

with Pool(2, timeout=20) as p:
    b = p.map_async(time.sleep, [0.1, 14], chunksize=1)
    time.sleep(0.5)
    p.apply_async(time.sleep, (30,), timeout=1)  # lands on the worker that ran b's first chunk
    b.get()                                      # WorkerLostError

Not new (main does the same with maxtasksperchild=1), but timeouts now make it reachable. Returning only running chunks fixes both cases, and the unit tests give the same results:

def worker_pids(self):
    return [pid for _, _, pid in self._running_chunks() if pid]

On the API question: the new arguments are optional and Celery calls both methods with just job, but the handler now always passes pid, so a subclass that overrides on_soft_timeout(self, job) or on_hard_timeout(self, job) would get a TypeError. I couldn't find one on GitHub outside type stubs. The conflict with main is only in t/unit/test_pool.py, where both sides add tests in the same place.

[auvipy]

I tried to fix the merge conflicts, lets see if that works in the CI

[auvipy]

@akashchamp can sou please check the last review?

[auvipy]

@LuisFigueroaG can you please cross check this one?

[akashchamp]

Thanks for the careful review. Pushed 5da5112: worker_pids() now returns only the workers of running chunks, which fixes your second case, and the other chunks that have also timed out are now killed after the first one fails the map. One gap remains: chunks still queued when the map fails are not timed once a worker picks them up, so your first example still delays the following task. That needs the failed map to stay tracked by the TimeoutHandler, so I left it. On the subclass point: the handler now always passes pid, so an override taking only job would fail, and that is worth a line in the release notes.

[MehrazRumman]

Checked at 5da5112 on macOS (Python 3.14, spawn). pytest t/unit: 89 passed, 53 skipped. CI is green.

I ran the same scripts on main and on this branch:

Scenario	main	this PR
Pool(2, timeout=2).map(time.sleep, [0.1, 6])	TimeoutHandler crashes with TypeError, process exits 1	TimeLimitExceeded after ~4s
same with soft_timeout=2	crashes, exit 1	SoftTimeLimitExceeded after ~4s
Pool(2, timeout=1).map(time.sleep, [0.4] * 8, chunksize=1)	crashes, exit 1	returns in 2s, so the limit applies per chunk
@GangEunzzang's second case (stale pid → WorkerLostError)	crashes, exit 1	returns [None, None]
@GangEunzzang's first case (queued chunks after the map fails)	crashes, exit 1	the next task waits ~8.8s

The crash fix, the per-chunk timing and the worker_pids() change all work. Reading the chunk indexing: _ack() fills every item of a chunk, so reading _time_accepted[start] / _worker_pid[start] in _running_chunks() is right.

Keeping the old call signature for apply_async() tasks

On @auvipy's API question: the handler now always calls on_soft_timeout(job, pid) / on_hard_timeout(job, pid), so a TimeoutHandler subclass that overrides them with (self, job) breaks for plain apply_async() tasks too, even though those worked before. Passing pid only for map chunks avoids that. A map() with a timeout crashed before anyway, so only the new code path needs the new argument:

                else:
                    tasks = [(i, job._time_accepted, None)]
                for key, ack_time, pid in tasks:
                    extra = () if pid is None else (pid,)
                    if _timed_out(ack_time, hard_timeout):
                        on_hard_timeout(job, *extra)
                    elif key not in dirty and _timed_out(ack_time,
                                                         soft_timeout):
                        on_soft_timeout(job, *extra)
                        dirty.add(key)

With that, the only change to the public API is the optional soft_timeout/timeout keywords on MapResult.__init__.

Smaller points

• Please add a CHANGES.txt entry. Besides the fix, it should say that the pool timeout/soft_timeout now apply to each map()/starmap() chunk, and that imap()/imap_unordered() results are still not timed. Today they don't crash, but they silently ignore the pool timeouts.
• The queued-chunk gap is fine to leave out of this PR, but it would be good to open an issue for it so it isn't lost: once a map fails, its chunks that are still queued run without a time limit.
• The description says flake8 shows nothing new, but t/unit/test_pool.py gets three new warnings: W293 on the whitespace-only line after test_map_hard_timeout (line 151), and two E501 lines (80 and 85 characters).

[GangEunzzang]

@akashchamp The worker_pids() change fixes the second case, but killing the other timed-out chunks can now kill an unrelated task. The pid is read before the first worker is killed, so if the second chunk finishes in the meantime, its worker has already picked up the next queued task and that task fails with WorkerLostError. I think the kill should check that the worker is still running that chunk.

[akashchamp]

Thanks, you were right. Pushed 9a36281: before killing the worker of another timed-out chunk, the handler now checks that no other job in the cache is running on it, and a regression test covers the case where that worker has already moved on to another task. The check can only see what the result handler has already processed, so it narrows the window rather than closing it completely.