[
https://issues.apache.org/jira/browse/SPARK-57931?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
L. C. Hsieh updated SPARK-57931:
--------------------------------
Description:
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. The channel is never restored, so with worker reuse enabled
(spark.python.worker.reuse=true, the default) the worker is returned to the
idle pool with its channel still in blocking mode.
PythonWorker.refresh() only opens a selector when the channel is non-blocking.
A pooled worker left in blocking mode therefore comes back with a null selector
/ selectionKey, and selector-path (non-pipelined) code that dereferences
worker.selector / worker.selectionKey would hit a NullPointerException.
In current OSS this is not an end-to-end failure: the worker-factory cache key
(PythonWorkersKey) includes the worker envVars, and the pipelined path adds
SPARK_PIPELINED_UDF=1 to envVars before requesting a worker. Pipelined and
non-pipelined tasks therefore draw from separate idle pools -- a worker left in
blocking mode only returns to the pipelined pool, whose next borrower is again
a pipelined task that re-sets the channel to blocking and does not use the
selector. So OSS masks the broken invariant via pool isolation.
That masking is fragile: it relies on the two pools staying disjoint via
envVars and does not fix the underlying invariant that a pooled daemon worker
is non-blocking. Any worker-management layer that pools or reuses workers
across that boundary will hand a blocking-mode worker to selector-path code and
hit the NPE.
Fix: normalize a reused daemon worker's channel back to non-blocking in
PythonWorkerFactory.create() (the single pool exit point), so a pooled worker
is always handed out in the same non-blocking mode as a fresh one. This is done
in create() rather than the pipelined path's task-completion listener because
the worker is released back to the pool when the reader reaches END_OF_STREAM,
which runs before the completion listener.
was:
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.
> 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. The channel is never restored, so with worker reuse enabled
> (spark.python.worker.reuse=true, the default) the worker is returned to the
> idle pool with its channel still in blocking mode.
> PythonWorker.refresh() only opens a selector when the channel is
> non-blocking. A pooled worker left in blocking mode therefore comes back with
> a null selector / selectionKey, and selector-path (non-pipelined) code that
> dereferences worker.selector / worker.selectionKey would hit a
> NullPointerException.
> In current OSS this is not an end-to-end failure: the worker-factory cache
> key (PythonWorkersKey) includes the worker envVars, and the pipelined path
> adds SPARK_PIPELINED_UDF=1 to envVars before requesting a worker. Pipelined
> and non-pipelined tasks therefore draw from separate idle pools -- a worker
> left in blocking mode only returns to the pipelined pool, whose next borrower
> is again a pipelined task that re-sets the channel to blocking and does not
> use the selector. So OSS masks the broken invariant via pool isolation.
> That masking is fragile: it relies on the two pools staying disjoint via
> envVars and does not fix the underlying invariant that a pooled daemon worker
> is non-blocking. Any worker-management layer that pools or reuses workers
> across that boundary will hand a blocking-mode worker to selector-path code
> and hit the NPE.
> Fix: normalize a reused daemon worker's channel back to non-blocking in
> PythonWorkerFactory.create() (the single pool exit point), so a pooled worker
> is always handed out in the same non-blocking mode as a fresh one. This is
> done in create() rather than the pipelined path's task-completion listener
> because the worker is released back to the pool when the reader reaches
> END_OF_STREAM, which runs before the completion listener.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
