You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "rangadi (via GitHub)" <gi...@apache.org> on 2023/09/21 04:06:55 UTC
[GitHub] [spark] rangadi opened a new pull request, #43023: [SPARK-45245] PythonWorkerFactory: Timeout if worker does not connect back.
rangadi opened a new pull request, #43023:
URL: https://github.com/apache/spark/pull/43023
### What changes were proposed in this pull request?
`createSimpleWorker()` method in `PythonWorkerFactory` waits forever if the worker fails to connect back to the server.
This is because it calls accept() without a timeout. If the worker does not connect back, accept() waits forever. There is supposed to be 10 seconds timeout, but it was not implemented correctly.
This PR adds a 10 second timeout.
### Why are the changes needed?
Otherwise create method could be stuck forever.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
- Unit test
- Manual
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: ChatGPT 4.0
Asked ChatGPT to generate sample code to do non-blocking accept() on a socket channel in Java.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on a diff in pull request #43023: [SPARK-45245] PythonWorkerFactory: Timeout if worker does not connect back.
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #43023:
URL: https://github.com/apache/spark/pull/43023#discussion_r1333512872
##########
core/src/test/scala/org/apache/spark/api/python/PythonWorkerFactorySuite.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import java.net.SocketTimeoutException
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+import org.scalatest.matchers.must.Matchers
+
+import org.apache.spark.SharedSparkContext
+import org.apache.spark.SparkException
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.util.ThreadUtils
+
+// Tests for PythonWorkerFactory.
+class PythonWorkerFactorySuite extends SparkFunSuite with Matchers with SharedSparkContext {
+
+ test("createSimpleWorker() fails with a timeout error if worker does not connect back") {
+ // It verifies that server side times out in accept(), if the worker does not connect back.
+ // E.g. the worker might fail at the beginning before it tries to connect back.
+
+ val workerFactory = new PythonWorkerFactory(
+ "python3", "pyspark.testing.non_existing_worker_module", Map.empty
+ )
+
+ // Create the worker in a separate thread so that if there is a bug where it does not
+ // return (accept() used to be blocking), the test doesn't hang for a long time.
+ val createFuture = Future {
+ val ex = intercept[SparkException] {
+ workerFactory.createSimpleWorker(blockingMode = true)
Review Comment:
May I know if this blockingMode=true would affect `serverSocketChannel.configureBlocking(false)` above?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-45245][PYTHON][CONNECT] PythonWorkerFactory: Timeout if worker does not connect back. [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #43023: [SPARK-45245][PYTHON][CONNECT] PythonWorkerFactory: Timeout if worker does not connect back.
URL: https://github.com/apache/spark/pull/43023
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] rangadi commented on a diff in pull request #43023: [SPARK-45245] PythonWorkerFactory: Timeout if worker does not connect back.
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on code in PR #43023:
URL: https://github.com/apache/spark/pull/43023#discussion_r1332415585
##########
core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala:
##########
@@ -184,10 +185,20 @@ private[spark] class PythonWorkerFactory(
redirectStreamsToStderr(workerProcess.getInputStream, workerProcess.getErrorStream)
// Wait for it to connect to our socket, and validate the auth secret.
- serverSocketChannel.socket().setSoTimeout(10000)
Review Comment:
There was supposed to be 10 second timeout. But this call does not seem to affect `serverSocketChannel.accept()`.
This set might only take effect if we did `serverSocketChannel.socket().accept()`, but hat returns a socket, not a channel.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-45245][PYTHON][CONNECT] PythonWorkerFactory: Timeout if worker does not connect back. [spark]
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on PR #43023:
URL: https://github.com/apache/spark/pull/43023#issuecomment-1786222604
> I think we should switch this to use the daemonized worker instead of simple workers soon.
I see. I have been looking for reasons doing so. This seems to be one of them.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-45245][PYTHON][CONNECT] PythonWorkerFactory: Timeout if worker does not connect back. [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #43023:
URL: https://github.com/apache/spark/pull/43023#discussion_r1519396300
##########
core/src/test/scala/org/apache/spark/api/python/PythonWorkerFactorySuite.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import java.net.SocketTimeoutException
+
+// scalastyle:off executioncontextglobal
+import scala.concurrent.ExecutionContext.Implicits.global
+// scalastyle:on executioncontextglobal
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+import org.scalatest.matchers.must.Matchers
+
+import org.apache.spark.SharedSparkContext
+import org.apache.spark.SparkException
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.util.ThreadUtils
+
+// Tests for PythonWorkerFactory.
+class PythonWorkerFactorySuite extends SparkFunSuite with Matchers with SharedSparkContext {
Review Comment:
Thanks. Made a followup PR: https://github.com/apache/spark/pull/45459
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WweiL commented on pull request #43023: [SPARK-45245] PythonWorkerFactory: Timeout if worker does not connect back.
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on PR #43023:
URL: https://github.com/apache/spark/pull/43023#issuecomment-1730174950
Seems the tests are failing because of format, maybe try
```
./build/mvn -Pscala-2.12 scalafmt:format -Dscalafmt.skip=false -Dscalafmt.validateOnly=false -Dscalafmt.changedOnly=false -pl connector/connect/common -pl connector/connect/server -pl connector/connect/client/jvm
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] rangadi commented on a diff in pull request #43023: [SPARK-45245] PythonWorkerFactory: Timeout if worker does not connect back.
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on code in PR #43023:
URL: https://github.com/apache/spark/pull/43023#discussion_r1333519063
##########
core/src/test/scala/org/apache/spark/api/python/PythonWorkerFactorySuite.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import java.net.SocketTimeoutException
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+import org.scalatest.matchers.must.Matchers
+
+import org.apache.spark.SharedSparkContext
+import org.apache.spark.SparkException
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.util.ThreadUtils
+
+// Tests for PythonWorkerFactory.
+class PythonWorkerFactorySuite extends SparkFunSuite with Matchers with SharedSparkContext {
+
+ test("createSimpleWorker() fails with a timeout error if worker does not connect back") {
+ // It verifies that server side times out in accept(), if the worker does not connect back.
+ // E.g. the worker might fail at the beginning before it tries to connect back.
+
+ val workerFactory = new PythonWorkerFactory(
+ "python3", "pyspark.testing.non_existing_worker_module", Map.empty
+ )
+
+ // Create the worker in a separate thread so that if there is a bug where it does not
+ // return (accept() used to be blocking), the test doesn't hang for a long time.
+ val createFuture = Future {
+ val ex = intercept[SparkException] {
+ workerFactory.createSimpleWorker(blockingMode = true)
Review Comment:
It doesn't. It is for client socket. Shall I remove it?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] rangadi commented on pull request #43023: [SPARK-45245] PythonWorkerFactory: Timeout if worker does not connect back.
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on PR #43023:
URL: https://github.com/apache/spark/pull/43023#issuecomment-1728734215
cc: @HyukjinKwon, @WweiL
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-45245][PYTHON][CONNECT] PythonWorkerFactory: Timeout if worker does not connect back. [spark]
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on PR #43023:
URL: https://github.com/apache/spark/pull/43023#issuecomment-1785280116
@HyukjinKwon the test failures seem unrelated. I tried multiple times. different types of tests are failing. Do you think we can merge this?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-45245][PYTHON][CONNECT] PythonWorkerFactory: Timeout if worker does not connect back. [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #43023:
URL: https://github.com/apache/spark/pull/43023#issuecomment-1786219936
Yeah I am merging it to master but I think we should switch this to use the daemonized worker instead of simple workers soon.
Merged to master.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-45245][PYTHON][CONNECT] PythonWorkerFactory: Timeout if worker does not connect back. [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43023:
URL: https://github.com/apache/spark/pull/43023#discussion_r1519318552
##########
core/src/test/scala/org/apache/spark/api/python/PythonWorkerFactorySuite.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import java.net.SocketTimeoutException
+
+// scalastyle:off executioncontextglobal
+import scala.concurrent.ExecutionContext.Implicits.global
+// scalastyle:on executioncontextglobal
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+import org.scalatest.matchers.must.Matchers
+
+import org.apache.spark.SharedSparkContext
+import org.apache.spark.SparkException
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.util.ThreadUtils
+
+// Tests for PythonWorkerFactory.
+class PythonWorkerFactorySuite extends SparkFunSuite with Matchers with SharedSparkContext {
Review Comment:
Does this suite have to extend `Matchers`? @HyukjinKwon can you double check?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org