[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. Only map chunks pass the worker pid to on_soft_timeout/on_hard_timeout; plain apply_async() tasks still call them with just job, so existing overrides keep working. imap() results are skipped since they don't record accept times, and a worker exit while an imap() is running no longer crashes the Supervisor thread.

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.
• Later (macOS, Python 3.12, after merging main): pytest t/unit: 108 passed, 53 skipped. Added tests for a TimeoutHandler subclass whose callbacks take only job, the soft timeout of each map chunk, a finished imap() task on a timed-out chunk's worker, and a hard timeout while an imap() is running.

[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.

[auvipy]

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__.

@akashchamp was this addressed?

[auvipy]

@GangEunzzang @MehrazRumman can you check if this pr is fixing the same problem #481 ?

[GangEunzzang]

@auvipy #481 is a different fix, it only rejects chunksize < 1 in imap(). Merging both only conflicts on the 4.4.0 entry in CHANGES.txt.

@akashchamp the crash fix holds up on Linux and macOS, but I think a few things still need fixing.

The pid suggestion from the last review isn't in yet. A TimeoutHandler subclass with on_hard_timeout(self, job) now crashes the process on a plain apply_async() timeout, which works on main. I couldn't find a real subclass like that, but passing pid only for map chunks avoids it.

_runs_other_job can also be fooled by imap(). IMapIterator.worker_pids() keeps workers that already finished, so with a live imap() a timed-out chunk's worker gets spared and the chunk runs past its limit:

p = Pool(3, timeout=3)
it = p.imap(work, [0.1, 20], chunksize=1)  # work sleeps, then prints
time.sleep(0.5)
p.map_async(work, [12, 12, 12], chunksize=1).get()  # fails after ~4s, one 12s chunk still finishes

Without the imap() all three chunks are killed.

Once a map fails, its running chunks lose their limit too, not just the queued ones:

p = Pool(2, timeout=3)
r = p.map_async(work, [1.5, 12, 12], chunksize=1)
r.get()  # TimeLimitExceeded after 3-4s, but the third chunk still finishes at ~13.5s

Same thing when a soft limit fails the map: a chunk still cleaning up never hits its hard limit. apply_async() kills both. When a timeout fails the map main crashes, so it's not a regression, and I'd be fine with this as a follow-up as long as the issue says running chunks too. A quick prototype that keeps failed maps tracked fixed the running and queued cases, but it needs _ack and _join_exited_workers changes.

The per-chunk soft path also has no test. With the old dirty filter a chunk gets repeated soft signals, and with dirty keyed per job later chunks get no soft limit at all, and t/unit/test_pool.py passes both ways.

[akashchamp]

Not until now, sorry: as of 2037a38 only map chunks get pid (21bbc81), _runs_other_job only counts running imap() tasks (1b2d732), and a per-chunk soft-timeout test covers the dirty key (2037a38). Reproducing the imap() case also showed a Supervisor crash (AttributeError: _write_to), fixed in fccf636. Running and queued chunks of an already failed map still lose their limit; that goes in a follow-up issue.