You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by davies <gi...@git.apache.org> on 2014/10/17 23:07:40 UTC
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
GitHub user davies opened a pull request:
https://github.com/apache/spark/pull/2838
[SPARK-3993] [PySpark] fix bug while reuse worker after take()
After take(), maybe there are some garbage left in the socket, then next task assigned to this worker will hang because of corrupted data.
We should make sure the socket is clean before reuse it, write END_OF_STREAM at the end, and check it after read out all result from python.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/davies/spark fix_reuse
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/2838.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2838
----
commit 660875b5cafb86e231d9d0b3a3f44ba9a13790a3
Author: Davies Liu <da...@gmail.com>
Date: 2014-10-17T20:59:40Z
fix bug while reuse worker after take()
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/2838#discussion_r19056695
--- Diff: python/pyspark/worker.py ---
@@ -57,7 +57,7 @@ def main(infile, outfile):
boot_time = time.time()
split_index = read_int(infile)
if split_index == -1: # for unit tests
- return
--- End diff --
Why this change? Was this just a debugging aid that allowed us to trigger failures but which was inadvertently broken by the worker re-use patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59594214
[QA tests have started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/21871/consoleFull) for PR 2838 at commit [`660875b`](https://github.com/apache/spark/commit/660875b5cafb86e231d9d0b3a3f44ba9a13790a3).
* This patch merges cleanly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59578891
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21865/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59595763
[QA tests have finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/21871/consoleFull) for PR 2838 at commit [`660875b`](https://github.com/apache/spark/commit/660875b5cafb86e231d9d0b3a3f44ba9a13790a3).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59635035
take() is not the only one which will introduce problems, user could call mapPartitions(), and read parts of the items in the infile.
Not only re-use the worker, we also want to re-use the socket.
try to call next() on infile maybe better than current approach. We still need a special code to tell JVM that the socket can be re-used or not, and any special code could be conflicted by random data. Such as the serialized data is broken, we will read the special code from it. In order to reduce the change of conflict, we could use VERY long special code, such as 64 bits or 128 bits or even more.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/2838#discussion_r19057007
--- Diff: python/pyspark/worker.py ---
@@ -57,7 +57,7 @@ def main(infile, outfile):
boot_time = time.time()
split_index = read_int(infile)
if split_index == -1: # for unit tests
- return
--- End diff --
The main() is called in a loop after re-use patch, so 'return' will not let the worker exit, this break the assumption that hold before the patch.
A better approach may be remove this special cases, and rewrite the test cases without any special treatments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59632902
The `take()`-handling code is a little tricky and has been the source of multiple bugs in the past, so it may be worth considering several different approaches for fixing this; I would hate for this problem to re-occur again.
For those who are unfamiliar with this code / issue, here a quick summary of this issue:
- When I call `take()` in PySpark, this translates down to a `mapPartitions()` call where the map function returns the `take()` results before consuming all of the partition's iterator.
- As a result, Python workers can complete their tasks before their input writers are closed.
- Previously, this could lead to various IOExceptions because the Java thread that writes the Python worker's input would still be running when the Python worker process shut down on task completion.
- Now that we re-use workers, the unread data written by the Java data writer thread will be left in the socket. This causes problems for subsequent tasks run on that worker, since the worker attempts to interpret whatever random data was left in the socket as Python code, configuration settings, etc.
- This bug, like the old IOException issue, only surfaces in cases where we have a large partition and a small `take()` size. If the partition is small or the `take()` is big, this stdin writer thread's output may be completely read by the worker. Basically, there's a race condition here which is data-size / take-size dependent.
This PR proposes to fix this by having the worker check the input stream to determine whether there was unread data left in the stream. If so, the worker exits and is not re-used.
I have two questions:
- Is there an approach that allows the worker to be (safely) re-used?
- Is there a less-brittle way to check whether the worker has consumed all of its input? I think we might be able to do this by calling `next()` on the input stream after we have run the user-provided code and checking for the "end-of-stream" code there, or checking for it inside of the serializers themselves. As I explained in [my comment in the diff](https://github.com/apache/spark/pull/2838#discussion_r19056687), I think that this may be less brittle in the face of future changes to our serialization code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59596462
[QA tests have started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/21872/consoleFull) for PR 2838 at commit [`8872914`](https://github.com/apache/spark/commit/88729145f6504397cc51a8851ea99e92f7a8938e).
* This patch merges cleanly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59596407
@aarondav Yes, before reuse workers, every python task will fork a new python worker.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59597959
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21872/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/2838#discussion_r19304373
--- Diff: python/pyspark/worker.py ---
@@ -131,6 +130,14 @@ def process():
for (aid, accum) in _accumulatorRegistry.items():
pickleSer._write_with_length((aid, accum._value), outfile)
+ # check end of stream
+ if read_int(infile) == SpecialLengths.END_OF_STREAM:
--- End diff --
Even without this assumption, There is only 1/2^32 chance to misunderstanding the next 4 bytes. I think it's still acceptable in practice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59595770
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21871/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-60328732
Since this approach seems to be technically correct given our current code, and since it fixes an urgent blocker, I'm going to merge this. We can revisit other solutions later, such as out-of-band control or checksumming of our streams.
If the assumptions in this patch are violated, the worst-case outcomes aren't any worse than what happens now:
- If the Python worker has shut down but the JVM doesn't realize this, then subsequent tasks scheduled for that worker will crash and Java will detect the shutdown at that point.
- If the Python worker does not realize that extra data was left in the stream, then it will read this extra data and probably crash while deserializing random data as a Python UDF.
So, in either case we crash the worker, causing a fresh one to be launched. This shouldn't happen, since this patch seems okay, but it's good to know that the likely worst-case outcome is a few extra crash failures and not data corruption.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59594023
Jenkins, retest this please.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/2838#discussion_r19056687
--- Diff: python/pyspark/worker.py ---
@@ -131,6 +130,14 @@ def process():
for (aid, accum) in _accumulatorRegistry.items():
pickleSer._write_with_length((aid, accum._value), outfile)
+ # check end of stream
+ if read_int(infile) == SpecialLengths.END_OF_STREAM:
--- End diff --
This seems to assume that `infile` will point to the beginning of a (length, [optional data]) pair rather than at some random data in the middle of a data block. I believe that this is actually correct given our current code, but I'm worried that the reason _why_ it's correct is subtle / brittle.
Our current serializers seem to maintain this invariant that once control flow has exited the serializer then `infile` will point to a valid length. For example, if I call `deserializer.load_stream(infile)` then call `next()` on this iterator, then once my `next()` call returns `infile` will point to a valid length. This invariant holds for our batched pickle serializer because we first deserialize an entire block (e.g. read a whole (length, data) pair) then yield its items one-by-one. I think it also holds for all of the other serializers, but it could potentially be broken by a streaming deserializer which doesn't need to read the entire data block in order to yield the first item.
Basically, my concern is that there's a lot of implicit / subtle assumptions here that we should document if we're going to take this approach.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59597958
[QA tests have finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/21872/consoleFull) for PR 2838 at commit [`8872914`](https://github.com/apache/spark/commit/88729145f6504397cc51a8851ea99e92f7a8938e).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/2838
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-3993] [PySpark] fix bug while reuse wor...
Posted by aarondav <gi...@git.apache.org>.
Github user aarondav commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59594220
Just for my understanding, is this solution that take() will cause workers to die rather than be reused with bad data in the socket?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org