jonathansudhakar1 commented on issue #25374:
URL: https://github.com/apache/beam/issues/25374#issuecomment-6710411217
@claudevdm, I'd like to fix this, building on your analysis above. Since it
touches `transforms/core.py`, I wanted to check the approach with you before
sending a PR.
**It still reproduces on master** (431b7c7ef0a, Python 3.13, DirectRunner):
```python
main | beam.FlatMap(lambda x, side: [x + sum(side)],
beam.pvalue.AsIter(side_pcoll)) \
.with_exception_handling(use_subprocess=True)
```
Every element goes to the dead-letter output with `AttributeError:
'_thread._local' object has no attribute 'side_input_cache_tokens'`.
- **Subprocess mode:** the pool's queue feeder thread pickles the
`_StateBackedIterable`, and its `__reduce__` reads state through a handler
whose context is thread-local.
- **Timeout mode:** the same pipeline with `timeout=30` and no subprocess
fails the same way, because `_TimeoutDoFn` iterates the side input on a
`ThreadPoolExecutor` thread.
- **Unaffected:** `AsList` and `AsDict` work in both modes, since they're
already materialized.
**Proposed fix.** I tried parts 1 and 2 locally, and with them the repro
passes in both modes:
1. `_SubprocessDoFn.process` pickles `(args, kwargs)` itself, in the
dispatching thread, and passes the bytes to `_remote_process`, which unpickles
them. The pickling is the same as today (a lazy side input still becomes a list
through `__reduce__`); it just happens in the thread that has the state context.
2. `_TimeoutDoFn.process` turns any `_StateBackedIterable` argument into a
list in the dispatching thread before submitting to its pool. That gives up
lazy iteration in timeout mode, which fails outright today anyway.
- The alternative is to carry the state handler's thread-local context
into the worker thread, the way #32571 carried the state sampler. But `core.py`
has no handle on the state handler, so that looks more invasive.
3. Tests: an `AsIter` side-input case in `DeadLettersTest` in
`ptransform_test.py`, which already runs with `use_subprocess` False and True,
plus a timeout-only case.
**Out of scope:** `AsMultiMap`. With `use_subprocess=True` it still raises
`TypeError` from `MultiMap.__reduce__` (the existing TODO), and with `timeout`
its per-key lookups would still run in the worker thread.
**Question:** does this split work for you? In particular, is it OK to
materialize `AsIter` side inputs in the timeout path (part 2)? Or would you
rather this fix only cover the subprocess path, or propagate the state context
instead?
.take-issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
