Fix ThreadPrefetch close deadlock when producer blocks in parent

[MohammadSadeghSalehi]

Fixes #1372.

Problem

ThreadPrefetchDatasetIterator.close() sets a stop event, joins the producer, and only
then closes the parent. The producer checks that stop event only at the top of its loop,
so when it is blocked inside parent.__next__() the event has no effect and the join
waits forever. The parent that could unblock it is not closed until after the join has
returned.

This is deterministic, 10 of 10 runs, for both single-level ThreadPrefetch and the
nested ThreadPrefetch -> map -> ThreadPrefetch shape that device_put produces. It is
not a nesting-only interaction.

Change

Cancellation is split from waiting.

• request_stop() marks the iterator closed, signals the local producer, wakes buffer
  waiters, and propagates the same non-blocking request down the parent chain. It never
  joins.
• close() calls request_stop(), then parent.close(), then joins. The existing
  sys.is_finalizing() guard on the join is preserved.
• __del__ calls request_stop() only, so a finalizer never blocks on a join.

Behaviour preserved

next() after close() still raises ValueError at every prefetch buffer size,
including 0, where queue.Queue(maxsize=0) is unbounded and a stop sentinel must not be
read as end of stream. There is a parameterized test for this over sizes 0, 1 and 5.

Performance

The healthy path is still a blocking get. The short timeout poll applies only after a
stop has been requested. Median throughput measured against the base commit on this
machine is within roughly 1 to 2 percent on both a fast producer and a slow producer,
which is inside run to run variance here.

Tests

test_close_does_not_hang_when_producer_blocked_in_parent, parameterized over
single-level and nested prefetch. It runs in a subprocess with a hard join timeout, so a
deadlock fails the suite rather than hanging it, and it asserts child exit code, wall
time, and the absence of live grain-thread-prefetch threads. It fails on the base
commit on behaviour and passes here.

Question on API placement

Should request_stop() live on the base DatasetIterator, with a default that
propagates to parents, rather than staying local to ThreadPrefetch? Every iterator that
owns a worker has the same problem, and a base method would make the ownership rule
explicit: request_stop is non-blocking cancellation, close is the blocking path that
joins. I kept it local here to keep the change small, and I am happy to move it if you
prefer the general shape.

Not included

Parent StopIteration is a BaseException, so the producer's except Exception never
puts end of stream on the buffer and a consumer blocked in get is not woken by it. That
is a separate defect, it is not required for the tests here, and I have left it out of
this change rather than bundling it.

Relationship to #1196

This is separate. The reporter's device_put script in #1196 does not hang at HEAD on
macOS arm64 on CPython 3.12, 3.13 or 3.14, and the sys.is_finalizing() join guard from
bc17ead appears to cover that shutdown path. Details are in a comment on that issue.

---

📚 Documentation preview 📚: https://google-grain--1373.org.readthedocs.build/

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[MohammadSadeghSalehi]

@googlebot I signed it!