[copybara-service]

Prevent ProcessPrefetchDatasetIterator exit hang after set_state() is called.

Why:
When ProcessPrefetchDatasetIterator.set_state() is called (e.g. during checkpoint restoration), self._set_state_queue.put() starts the multiprocessing.Queue background feeder thread, which registers Queue._finalize_close at exitpriority=10. Because ProcessPrefetchDatasetIterator previously registered _close_if_alive via multiprocessing.util.Finalize at exitpriority=1, Python interpreter shutdown ran Queue._finalize_close before _close_if_alive.

When _close_if_alive subsequently called _stop_prefetch() -> _clear_set_state_queue(), q.get_nowait() raised OSError: handle is closed on the already-closed queue. Because _clear_queue_and_maybe_unlink_shm only caught queue.Empty and _stop_prefetch() did not guard process termination with try...finally, the OSError aborted _stop_prefetch() before reaching self._prefetch_process.join(timeout=_PROCESS_KILL_TIMEOUT_S) and self._prefetch_process.kill(). Python's multiprocessing.util._exit_function() then fell back to sending SIGTERM and calling p.join() with no timeout on active_children(), causing the parent process to hang indefinitely at exit if the worker process did not exit on SIGTERM.

What:

1. Increase the exitpriority of _close_if_alive in ProcessPrefetchDatasetIterator.__init__() from 1 to 20 so it runs before multiprocessing.Queue's exitpriority=10 finalizer closes _set_state_queue and _buffer.
2. Catch (queue.Empty, ValueError, OSError, EOFError) around q.get_nowait() in _clear_queue_and_maybe_unlink_shm() so already-closed or broken queues do not raise during shutdown cleanup.
3. Wrap _unregister_fn(), _prefetch_should_stop.set(), and queue draining in _stop_prefetch() in a try...finally block so self._prefetch_process.join(timeout=_PROCESS_KILL_TIMEOUT_S) and self._prefetch_process.kill() are guaranteed to run even if cleanup raises an exception, and call self._prefetch_process.join(timeout=_PROCESS_KILL_TIMEOUT_S) after .kill() to reap the child process and remove it from multiprocessing.active_children().
4. Add test_worker_process_killed_during_queue_put_after_set_state in process_prefetch_test.py to cover interpreter shutdown after set_state() when the worker is blocked and ignores SIGTERM.