[
https://issues.apache.org/jira/browse/SPARK-57931?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
L. C. Hsieh updated SPARK-57931:
--------------------------------
Affects Version/s: 4.3.0
(was: 5.0.0)
> Restore worker channel blocking mode after pipelined Python UDF execution
> -------------------------------------------------------------------------
>
> Key: SPARK-57931
> URL: https://issues.apache.org/jira/browse/SPARK-57931
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.3.0
> Reporter: L. C. Hsieh
> Priority: Major
> Labels: pull-request-available
>
> SPARK-56642 added an opt-in pipelined Python UDF path. When enabled,
> createPipelinedDataIn() switches the shared worker's SocketChannel from
> non-blocking to blocking mode (channel.configureBlocking(true) +
> worker.refresh())
> so the writer thread and the task thread can do full-duplex blocking I/O.
> However, the channel mode is never restored. With worker reuse enabled
> (spark.python.worker.reuse=true, the default), the task completion listener
> only
> cancels the writer future; the worker is then released back into the idle pool
> (PythonWorkerFactory.releaseWorker -> idleWorkers.enqueue) with its channel
> still
> in blocking mode.
> When a subsequent task pulls that worker from the pool,
> PythonWorkerFactory.create()
> only calls worker.refresh(). Because refresh() decides whether to open a
> selector
> based on the current channel mode, a channel left in blocking mode results in
> selectorOpt = None / selectionKeyOpt = None. If that next task runs on the
> non-pipelined (single-threaded NIO selector) path, it dereferences
> worker.selector
> / worker.selectionKey and hits a NullPointerException.
> This is cross-task state pollution: a pooled worker borrowed once by the
> pipelined
> path corrupts the channel state for every task that later reuses it,
> regardless of
> whether that task uses pipelined mode. It also breaks setups that rely on
> pooled
> workers being in a known (non-blocking) state.
> Fix: capture the channel's original blocking mode before
> createPipelinedDataIn()
> changes it, and restore it in the task completion listener (after the writer
> has
> exited and only if the channel is still open), so a pooled worker is always
> returned in the exact mode it was borrowed in.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
