[
https://issues.apache.org/jira/browse/SPARK-57931?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
L. C. Hsieh resolved SPARK-57931.
---------------------------------
Fix Version/s: 4.3.0
Resolution: Fixed
Issue resolved by pull request 56995
[https://github.com/apache/spark/pull/56995]
> 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
> Assignee: L. C. Hsieh
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.3.0
>
>
> 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]
