Don't crash the Supervisor when a worker exits during imap()

[LuisFigueroaG]

If a worker exits while an imap()/imap_unordered() still has tasks that no worker has picked up, _join_exited_workers() reads job._write_to/job._scheduled_for on the IMapIterator, which doesn't define them. The Supervisor thread crashes with AttributeError: 'IMapIterator' object has no attribute '_write_to', and PoolThread.run then calls os._exit(1), so the whole parent process goes down. One way to hit it: Pool(1), an apply_async whose task raises SystemExit, and an imap() queued behind it.

This adds both attributes to IMapIterator as None class attributes, as ApplyResult already has them. It only adds the missing defaults, so existing behaviour and the public API don't change.

Added a unit test that calls _join_exited_workers() directly on a pool without threads, so a failure shows up as a normal test failure instead of killing the test process.

[auvipy]

any old open issue related to this? @MehrazRumman can you try and review this?

[LuisFigueroaG]

I couldn't find an open issue for this exact crash. The closest is #141 (a map/imap with a hard timeout= makes the program exit with code 1), but that's a different failure. I ran the #141 script against current main, and it dies earlier, in the TimeoutHandler: 'IMapIterator' object has no attribute '_time_accepted'. It does the same with this branch, so this PR doesn't fix #141. That one needs time limits to work with IMapIterator, which I'd keep as a separate change.

This PR covers the case where a worker exits for any other reason (SystemExit in a task, OOM kill, maxtasksperchild) while an imap() still has tasks no worker has picked up.

[auvipy]

@GangEunzzang please have a look to this

[GangEunzzang]

Looks good to me. The else branch in _join_exited_workers() only runs for jobs that no exited worker has acked, and _write_to/_scheduled_for are only ever set by Celery's AsynPool on ApplyResult. So None on the iterator is exactly what a plain Pool already sees for ApplyResult.

Repro with a real Pool(1): apply_async() of a task that raises SystemExit, then imap(sq, range(6)) queued behind it.

	main (fe23965)	main + this PR
imap / imap_unordered, chunksize 1 and 2	Supervisor dies on _write_to, process exits 1	the replacement worker runs the queued tasks: all 6 results, then StopIteration
maxtasksperchild=1 + imap (control)	6 results	6 results

The new test fails on main and passes here. pool.py merges cleanly with #473.

Two things nearby that this PR doesn't cover (both also happen on main):

• pool.terminate_job(pid) while that worker runs an imap() task kills the process from the same function: AttributeError: 'IMapIterator' object has no attribute '_set_terminated'. That one is a method call, so a None attribute won't fix it.
• If the worker running an imap() task is SIGKILLed, ordered imap() blocks forever. mark_as_worker_lost() calls job._set(None, ...) and IMapIterator puts that in _unsorted[None], where it is never read. imap_unordered() yields the WorkerLostError, and map_async()/apply_async() raise it.

Up to you whether _set_terminated belongs here or in a follow-up. For #141: #473 skips IMapIterator in handle_timeouts(), and with it Pool(1, timeout=3) + imap() no longer crashes the TimeoutHandler.