[SPARK-32010][PYTHON][CORE] Add InheritableThread for local properties and fixing a thread leak issue in pinned thread mode #28968

HyukjinKwon · 2020-07-01T15:09:47Z

What changes were proposed in this pull request?

This PR proposes:

To introduce InheritableThread class, that works identically with threading.Thread but it can inherit the inheritable attributes of a JVM thread such as InheritableThreadLocal.

This was a problem from the pinned thread mode, see also [SPARK-22340][PYTHON] Add a mode to pin Python thread into JVM's #24898. Now it works as below:
```
import pyspark

spark.sparkContext.setLocalProperty("a", "hi")
def print_prop():
    print(spark.sparkContext.getLocalProperty("a"))

pyspark.InheritableThread(target=print_prop).start()
```
```
hi
```

Also, it adds the resource leak fix into InheritableThread. Py4J leaks the thread and does not close the connection from Python to JVM. In InheritableThread, it manually closes the connections when PVM garbage collection happens. So, JVM threads finish safely. I manually verified by profiling but there's also another easy way to verify:

PYSPARK_PIN_THREAD=true ./bin/pyspark

>>> from threading import Thread
>>> Thread(target=lambda: spark.range(1000).collect()).start()
>>> Thread(target=lambda: spark.range(1000).collect()).start()
>>> Thread(target=lambda: spark.range(1000).collect()).start()
>>> spark._jvm._gateway_client.deque
deque([<py4j.clientserver.ClientServerConnection object at 0x119f7aba8>, <py4j.clientserver.ClientServerConnection object at 0x119fc9b70>, <py4j.clientserver.ClientServerConnection object at 0x119fc9e10>, <py4j.clientserver.ClientServerConnection object at 0x11a015358>, <py4j.clientserver.ClientServerConnection object at 0x119fc00f0>])
>>> Thread(target=lambda: spark.range(1000).collect()).start()
>>> spark._jvm._gateway_client.deque
deque([<py4j.clientserver.ClientServerConnection object at 0x119f7aba8>, <py4j.clientserver.ClientServerConnection object at 0x119fc9b70>, <py4j.clientserver.ClientServerConnection object at 0x119fc9e10>, <py4j.clientserver.ClientServerConnection object at 0x11a015358>, <py4j.clientserver.ClientServerConnection object at 0x119fc08d0>, <py4j.clientserver.ClientServerConnection object at 0x119fc00f0>])

This issue is fixed now.

Because now we have a fix for the issue here, it also proposes to deprecate collectWithJobGroup which was a temporary workaround added to avoid this leak issue.

Why are the changes needed?

To support pinned thread mode properly without a resource leak, and a proper inheritable local properties.

Does this PR introduce any user-facing change?

Yes, it adds an API InheritableThread class for pinned thread mode.

How was this patch tested?

Manually tested as described above, and unit test was added as well.

HyukjinKwon · 2020-07-01T15:10:49Z

cc @WeichenXu123 @mengxr, @dongjoon-hyun, @squito, @viirya

python/pyspark/util.py

dongjoon-hyun · 2020-07-01T15:30:42Z

Just a question: do we need [CORE] tag for this kind of Python-code-specific change? Although this touches PySpark RDD, we usually use [CORE] for core module literally.

HyukjinKwon · 2020-07-01T15:44:02Z

I think that's what people usually do. In particular, ML side often. I think it's better to classify it more explicitly.

dongjoon-hyun · 2020-07-01T22:57:10Z

Oh, got it. It seems that I misunderstand the standard. Thanks. :)

I think that's what people usually do. In particular, ML side often. I think it's better to classify it more explicitly.

dongjoon-hyun · 2020-07-01T22:57:44Z

Retest this please.

HyukjinKwon · 2020-07-02T01:50:10Z

@dongjoon-hyun no problem :-). I think this is more about preference things .. If it causes any problem or confusion, I will change my way.

HyukjinKwon · 2020-07-02T01:53:00Z

retest this please

SparkQA · 2020-07-02T02:37:56Z

Test build #124839 has finished for PR 28968 at commit f369666.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2020-07-09T08:50:11Z

gentle ping for a review :-).

…ssue in pinned thread mode

SparkQA · 2020-07-14T04:29:12Z

Test build #125802 has finished for PR 28968 at commit a78fd43.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class InheritableThread(threading.Thread):

HyukjinKwon · 2020-07-21T06:10:44Z

retest this please

SparkQA · 2020-07-21T06:40:20Z

Test build #126222 has finished for PR 28968 at commit a78fd43.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class InheritableThread(threading.Thread):

HyukjinKwon · 2020-07-21T07:07:19Z

retest this please

SparkQA · 2020-07-21T07:46:03Z

Test build #126231 has finished for PR 28968 at commit a78fd43.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class InheritableThread(threading.Thread):

HyukjinKwon · 2020-07-25T05:14:39Z

retest this please

SparkQA · 2020-07-25T05:51:56Z

Test build #126528 has finished for PR 28968 at commit a78fd43.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class InheritableThread(threading.Thread):

HyukjinKwon · 2020-07-27T11:20:31Z

retest this please

SparkQA · 2020-07-27T11:48:44Z

Test build #126640 has finished for PR 28968 at commit a78fd43.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class InheritableThread(threading.Thread):

WeichenXu123 · 2020-07-28T09:27:36Z

sync with @HyukjinKwon offline, LGTM except one concern:
if we add this thread class instead of fixing it in py4j, can we enable pin thread mode by default in future release ?
if enable pin thread mode by default, many user existing code which use plain thread may meet memory leak issue.

But fixing it in py4j seems to be difficult, py4j do not know which thread is about to be GCed except thread notifying py4j initiatively

HyukjinKwon · 2020-07-29T06:34:31Z

Thanks @WeichenXu123. I will leave it open few more days before merging it.

HyukjinKwon · 2020-07-30T01:15:08Z

Merged to master.

viirya · 2020-08-29T04:14:12Z

docs/job-scheduling.md

-each thread with its own local properties. To work around this, you should manually copy and set the
-local properties from the parent thread to the child thread when you create another thread in PVM.
+to `true`. This pinned thread mode allows one PVM thread has one corresponding JVM thread. With this mode,
+`pyspark.InheritableThread` is recommanded to use together for a PVM thread to inherit the interitable attributes


typo: interitable -> inheritable

viirya · 2020-08-29T04:28:30Z

python/pyspark/util.py

+
+        if isinstance(sc._gateway, ClientServer):
+            # Here's when the pinned-thread mode (PYSPARK_PIN_THREAD) is on.
+            properties = sc._jsc.sc().getLocalProperties().clone()


Why we need to clone? Doesn't sc.localProperties get clone in childValue already?

Actually we're mimicking that behaviour here because the thread in JVM does not respect the inheritance here since the thread is always sepearately created via the JVM gateway whereas Scala Java side we can keep the inheritance by creating a thread within a thread.

viirya · 2020-08-29T04:43:05Z

I found I missed this and looked at now. LGTM. I'm just wondering we should use InheritableThread in the PR description to verify the fix?

>>> from threading import Thread
>>> Thread(target=lambda: spark.range(1000).collect()).start()
>>> Thread(target=lambda: spark.range(1000).collect()).start()
>>> Thread(target=lambda: spark.range(1000).collect()).start()
>>> spark._jvm._gateway_client.deque
deque([<py4j.clientserver.ClientServerConnection object at 0x119f7aba8>, <py4j.clientserver.ClientServerConnection object at 0x119fc9b70>, <py4j.clientserver.ClientServerConnection object at 0x119fc9e10>, <py4j.clientserver.ClientServerConnection object at 0x11a015358>, <py4j.clientserver.ClientServerConnection object at 0x119fc00f0>])
>>> Thread(target=lambda: spark.range(1000).collect()).start()
>>> spark._jvm._gateway_client.deque
deque([<py4j.clientserver.ClientServerConnection object at 0x119f7aba8>, <py4j.clientserver.ClientServerConnection object at 0x119fc9b70>, <py4j.clientserver.ClientServerConnection object at 0x119fc9e10>, <py4j.clientserver.ClientServerConnection object at 0x11a015358>, <py4j.clientserver.ClientServerConnection object at 0x119fc08d0>, <py4j.clientserver.ClientServerConnection object at 0x119fc00f0>])

HyukjinKwon · 2020-08-29T05:47:51Z

Oh yeah we should use InheritableThread instead of Thread to verify this PR.

HyukjinKwon · 2020-08-29T05:51:58Z

Thank you for taking a look @viirya.

probot-autolabeler bot added CORE PYTHON labels Jul 1, 2020

HyukjinKwon force-pushed the SPARK-32010 branch from 8d1bf49 to 6cbe399 Compare July 1, 2020 15:11

probot-autolabeler bot added the DOCS label Jul 1, 2020

HyukjinKwon changed the title ~~[SPARK-32010][PYTHON] Add InheritableThread for local properties and fixing a thread leak issue in pinned thread mode~~ [SPARK-32010][PYTHON][CORE] Add InheritableThread for local properties and fixing a thread leak issue in pinned thread mode Jul 1, 2020

dongjoon-hyun reviewed Jul 1, 2020

View reviewed changes

python/pyspark/util.py Outdated Show resolved Hide resolved

This comment has been minimized.

Sign in to view

Add InheritableThread for local properties and fixing a thread leak i…

a78fd43

…ssue in pinned thread mode

HyukjinKwon force-pushed the SPARK-32010 branch from f369666 to a78fd43 Compare July 14, 2020 03:58

HyukjinKwon closed this Jul 27, 2020

HyukjinKwon deleted the SPARK-32010 branch July 27, 2020 07:43

HyukjinKwon restored the SPARK-32010 branch July 27, 2020 07:46

HyukjinKwon reopened this Jul 27, 2020

HyukjinKwon closed this in 89d9b7c Jul 30, 2020

viirya reviewed Aug 29, 2020

View reviewed changes

zero323 mentioned this pull request Aug 31, 2020

[SPARK-32010] Thread leaks in pinned thread mode zero323/pyspark-stubs#504

Closed

HyukjinKwon deleted the SPARK-32010 branch December 7, 2020 02:06

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-32010][PYTHON][CORE] Add InheritableThread for local properties and fixing a thread leak issue in pinned thread mode #28968

[SPARK-32010][PYTHON][CORE] Add InheritableThread for local properties and fixing a thread leak issue in pinned thread mode #28968

HyukjinKwon commented Jul 1, 2020 •

edited

Loading

HyukjinKwon commented Jul 1, 2020 •

edited

Loading

dongjoon-hyun commented Jul 1, 2020 •

edited

Loading

This comment has been minimized.

HyukjinKwon commented Jul 1, 2020

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

dongjoon-hyun commented Jul 1, 2020

dongjoon-hyun commented Jul 1, 2020

This comment has been minimized.

HyukjinKwon commented Jul 2, 2020

HyukjinKwon commented Jul 2, 2020

SparkQA commented Jul 2, 2020

HyukjinKwon commented Jul 9, 2020

SparkQA commented Jul 14, 2020

HyukjinKwon commented Jul 21, 2020

SparkQA commented Jul 21, 2020

HyukjinKwon commented Jul 21, 2020

SparkQA commented Jul 21, 2020

HyukjinKwon commented Jul 25, 2020

SparkQA commented Jul 25, 2020

HyukjinKwon commented Jul 27, 2020

SparkQA commented Jul 27, 2020

WeichenXu123 commented Jul 28, 2020

HyukjinKwon commented Jul 29, 2020

HyukjinKwon commented Jul 30, 2020

viirya Aug 29, 2020

viirya Aug 29, 2020 •

edited

Loading

HyukjinKwon Aug 29, 2020

viirya commented Aug 29, 2020

HyukjinKwon commented Aug 29, 2020

HyukjinKwon commented Aug 29, 2020 •

edited

Loading

[SPARK-32010][PYTHON][CORE] Add InheritableThread for local properties and fixing a thread leak issue in pinned thread mode #28968

[SPARK-32010][PYTHON][CORE] Add InheritableThread for local properties and fixing a thread leak issue in pinned thread mode #28968

Conversation

HyukjinKwon commented Jul 1, 2020 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

HyukjinKwon commented Jul 1, 2020 • edited Loading

dongjoon-hyun commented Jul 1, 2020 • edited Loading

This comment has been minimized.

HyukjinKwon commented Jul 1, 2020

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

dongjoon-hyun commented Jul 1, 2020

dongjoon-hyun commented Jul 1, 2020

This comment has been minimized.

HyukjinKwon commented Jul 2, 2020

HyukjinKwon commented Jul 2, 2020

SparkQA commented Jul 2, 2020

HyukjinKwon commented Jul 9, 2020

SparkQA commented Jul 14, 2020

HyukjinKwon commented Jul 21, 2020

SparkQA commented Jul 21, 2020

HyukjinKwon commented Jul 21, 2020

SparkQA commented Jul 21, 2020

HyukjinKwon commented Jul 25, 2020

SparkQA commented Jul 25, 2020

HyukjinKwon commented Jul 27, 2020

SparkQA commented Jul 27, 2020

WeichenXu123 commented Jul 28, 2020

HyukjinKwon commented Jul 29, 2020

HyukjinKwon commented Jul 30, 2020

viirya Aug 29, 2020

Choose a reason for hiding this comment

viirya Aug 29, 2020 • edited Loading

Choose a reason for hiding this comment

HyukjinKwon Aug 29, 2020

Choose a reason for hiding this comment

viirya commented Aug 29, 2020

HyukjinKwon commented Aug 29, 2020

HyukjinKwon commented Aug 29, 2020 • edited Loading

HyukjinKwon commented Jul 1, 2020 •

edited

Loading

HyukjinKwon commented Jul 1, 2020 •

edited

Loading

dongjoon-hyun commented Jul 1, 2020 •

edited

Loading

viirya Aug 29, 2020 •

edited

Loading

HyukjinKwon commented Aug 29, 2020 •

edited

Loading