[SPARK-21093][R] Terminate R's worker processes in the parent of R's daemon to prevent a leak

[HyukjinKwon]

What changes were proposed in this pull request?

mcfork in R looks opening a pipe ahead but the existing logic does not properly close it when it is executed hot. This leads to the failure of more forking due to the limit for number of files open.

This hot execution looks particularly for gapply/gapplyCollect. For unknown reason, this happens more easily in CentOS and could be reproduced in Mac too.

All the details are described in https://issues.apache.org/jira/browse/SPARK-21093

This PR proposes simply to terminate R's worker processes in the parent of R's daemon to prevent a leak.

How was this patch tested?

I ran the codes below on both CentOS and Mac with that configuration disabled/enabled.

df <- createDataFrame(list(list(1L, 1, "1", 0.1)), c("a", "b", "c", "d"))
collect(gapply(df, "a", function(key, x) { x }, schema(df)))
collect(gapply(df, "a", function(key, x) { x }, schema(df)))
...  # 30 times

Also, now it passes R tests on CentOS as below:

SparkSQL functions: Spark package found in SPARK_HOME: .../spark
..............................................................................................................................................................
..............................................................................................................................................................
..............................................................................................................................................................
..............................................................................................................................................................
..............................................................................................................................................................
....................................................................................................................................

[HyukjinKwon]

cc @felixcheung, @shivaram and @MLnick.

[felixcheung]

Hmm I'm not sure - I'm pretty sure the session / spark context is already initialized when this test is run and changing the setting here does it affect the existing daemon process already running?

[HyukjinKwon]

Yes, I believe you are correct and the daemon is already running but it avoids to use the problematic daemon -

spark/core/src/main/scala/org/apache/spark/api/r/RRunner.scala

Lines 363 to 392 in 478fbc8

	if (!Utils.isWindows && useDaemon) {
	synchronized {
	if (daemonChannel == null) {
	// we expect one connections
	val serverSocket = new ServerSocket(0, 1, InetAddress.getByName("localhost"))
	val daemonPort = serverSocket.getLocalPort
	errThread = createRProcess(daemonPort, "daemon.R")
	// the socket used to send out the input of task
	serverSocket.setSoTimeout(10000)
	val sock = serverSocket.accept()
	daemonChannel = new DataOutputStream(new BufferedOutputStream(sock.getOutputStream))
	serverSocket.close()
	}
	try {
	daemonChannel.writeInt(port)
	daemonChannel.flush()
	} catch {
	case e: IOException =>
	// daemon process died
	daemonChannel.close()
	daemonChannel = null
	errThread = null
	// fail the current task, retry by scheduler
	throw e
	}
	errThread
	}
	} else {
	createRProcess(port, "worker.R")
	}

up to my knowledge.

[SparkQA]

Test build #78147 has finished for PR 18320 at commit 505d75f.

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

[SparkQA]

Test build #78148 has finished for PR 18320 at commit 52c8abf.

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

[felixcheung]

thx - I think more importantly, does the issue manifest when someone manually call gapply in a similar way on RHEL/CentOS? We could workaround the test failure, but if user can use into this in normal use then we need to address this within gapply

[HyukjinKwon]

Yes, there is still the issue and this only fixes (avoid) the test failure. I believe running the codes should reproduce the issue for both Mac and CentOS.

What I don't get is, when the number of fork executions is not many, this is not reproduced sometimes (even increasing pipes were not observed sometimes). The number of the pipes decrease in a certain condition (it did not look related with time but some events).

The issue with gapply is exposed and found now as it invokes many forks via daemon.R but I guess this issue might still exist for all other APIs executing R native function with this daemon. I gave a shot to resolve the root cause within daemon.R with several tries but I could not make it.

Root cause is:

With a terminal executing watch -n 0.01 "lsof -c R | wc -l" (<- this actually should be -p PID but it is good enough to see the symptoms).

With another terminal:

for(i in 0:200) {
  p <- parallel:::mcfork()
  if (inherits(p, "masterProcess")) {
    tools::pskill(Sys.getpid(), tools::SIGUSR1)
    parallel:::mcexit(0L)
  }
}

The number of opened pipes just keep increasing. I double checked the processes and sockets are closed via netstats and ps.

We need to resolve this one.

[HyukjinKwon]

For normal usecases, I carefully suspect it might be fine because I executed 200 * ~10 tasks in a single machine quickly but I don't know if it happens frequently when it runs slowly in a cluster in a distributed manner.

At least, this was not reproduced sometimes when the number of fork executions is not many. Practically, it might be fine but need more investigation if this is important to prioritize this issue.

[felixcheung]

that's very interesting. that code has been around for 2 years - to be honest I'm not 100% sure about what it is doing. perhaps this is an existing leak with R when the socket is copied to a child fork.

perhaps this could also be fixed with a lower number of partitions?

again, I am actually not too concern about the test failure itself - this workaround seems fine, and we could even skip the test on RHEL. I am very concern with whether this is an ongoing memory leak that user can run into over time, and the fact that it fails here consistently in tests is only a factor on the number of operations performed so far and not in any way confined to gapply

[HyukjinKwon]

Yes, I guess it will pass if we reduce spark.sql.shuffle.partitions (< I didn't look carefully and test this either). Just to make sure (and to share what I investigated ...), from my code read,

With spark.sparkr.use.daemon enabled, for each task execution,

1. JVM ----start (if not started)--------> R daemon
2. JVM ----send port ------------------> R daemon ----fork with the port---> R worker

This looks being tested on OSs except for Windows.

With spark.sparkr.use.daemon disabled, for each task execution,

1. JVM ----forking processes with the port from Java (expensive)-----------> R worker

This looks being tested only on Windows.

This PR proposes to switch this one to latter case (which was the former before) by avoiding calling the (already running from other execution in the tests) R daemon.

I am fine with giving a shot with reducing the number of partitions if you are fond of it.

[HyukjinKwon]

I suspect this is an issue in R. I will raise this issue in R community soon and share it.

[HyukjinKwon]

@felixcheung, BTW, is it okay as a PR alone as is?

+and actually if this can't be fixed within daemon.R, I could avoid using this daemon in dapply/gapply always within execution path. However, I guess this brings performance regression.

[felixcheung]

Does it fail by running just gapply and nothing else?
From what you have found in your investigations and the code you pointed to, I suspect this isn't limited to gapply.

I think this PR so only works around the problem. I am concern that an user can also run into this issue.

An naive approach might be to change spark.sparkr.use.daemon inside gapply when it is called, but I suspect that only shifts the problem around, and it might fail then with other methods that shuffles or calls UDFs. If a long running demon process is the problem, either we find and fix the leak (close the pipe, socket etc) or we put a count on the number of execution and re-cycle the demon process periodically before this leak becomes fatal.

thought?

[HyukjinKwon]

I was thinking we should not work around for now if this can't be fixed within daemon.R and it is not practically problematic which we don't know yet.

Let try to deal withdaemon.R. Probably, the best way is to fix this problem together. I will be back after thinking more and taking a look.

[felixcheung]

also, perhaps this is a bug in RHEL/CentOS? it seems like mcfork only calls the OS fork function
I'll try CentOS 6.x

[HyukjinKwon]

This also happens in Mac to me. My assumption is, mcfork itself might be fine (opening a pipe between parent and child is okay) but when the child exits, it looks failed to return the resources properly in a certain condition (it was hot execution given my test).

[HyukjinKwon]

It looks we should exit children in the parent. Previously, the child looks sending the kill signal for itself. Killing those in the parent looks resolving this problem properly.

Before:

for(i in 0:200) {
  p <- parallel:::mcfork()
  if (inherits(p, "masterProcess")) {
    tools::pskill(Sys.getpid(), tools::SIGUSR1)
    parallel:::mcexit(0L)
  }
}

Roughly, after:

for(i in 0:200) {
  p <- parallel:::mcfork()
  if (inherits(p, "masterProcess")) {
    parallel:::mcexit(0L)
  }
  
  exitChildren <- parallel:::selectChildren()
  if (is.integer(exitChildren)) {
    lapply(exitChildren, function(x) { tools::pskill(x, tools::SIGUSR1) })
  }
}

[HyukjinKwon]

So, I intended to terminates the children for every second or before launching other workers.

[HyukjinKwon]

The logic here can be tested as below:

With a terminal (terminal A) running the command as below:

nc -l 7777

with another terminal (terminal B) running the codes below in R shell.

inputCon <- socketConnection(port = "7777", open = "rb", blocking = TRUE)  
print(socketSelect(list(inputCon), timeout = 1))

This prints FALSE as it does not have any data first.

In terminal A, if any character is typed first, this returns TRUE in terminal B with another execution.

[shivaram]

@HyukjinKwon Do we know what the old behavior was doing ? I'm thinking if we would spend more time in the loop below now and be less responsive to the inputCon. One thing -- Can we control the timeout based on whether there are any children running or not ?

[shivaram]

To make my point a bit more clear, if there are no workers running, we can block on this select and dont need the one second timeout in that case ?

[HyukjinKwon]

Yea, I got your point and was taking a look. I think probably feasible via parallel:::children. I will give a shot to deal with this here and be back.

[HyukjinKwon]

BTW, the old behaviour was simply to hang and wait there indefinitely.

https://stat.ethz.ch/R-manual/R-devel/library/base/html/socketSelect.html

numeric or NULL. Time in seconds to wait for a socket to become available; NULL means wait indefinitely.

[HyukjinKwon]

timeout for selectChildren looks 0 by default. So, this does not look hanging here.

[HyukjinKwon]

This can be tested as below:

vi tmp.R

copy and paste

p <- parallel:::mcfork()
if (inherits(p, "masterProcess")) {
  Sys.sleep(3L)
  parallel:::mcexit(0L)
}
print(parallel:::selectChildren())
Sys.sleep(4L)
print(parallel:::selectChildren())

and running Rscript tmp.R prints

[1] TRUE
[1] 2505

For return values, please refer https://stat.ethz.ch/R-manual/R-devel/library/parallel/html/children.html.

For testing resource leak, this can be tested as below:

for(i in 0:100000) {
  p <- parallel:::mcfork()
  if (inherits(p, "masterProcess")) {
    parallel:::mcexit(0L)
  }
  
  exitChildren <- parallel:::selectChildren()
  if (is.integer(exitChildren)) {
    lapply(exitChildren, function(x) { tools::pskill(x, tools::SIGUSR1) })
  }
}

The original code below:

for(i in 0:100000) {
  p <- parallel:::mcfork()
  if (inherits(p, "masterProcess")) {
    tools::pskill(Sys.getpid(), tools::SIGUSR1)
    parallel:::mcexit(0L)
  }
}

will probably complain in mcfork due to the lack of resources (e.g., ran out of file descriptors).

Actual number can be checked via watch -n 0.01 "lsof -c R | wc -l" (again, to check this more correctly, it should be -p PID instead of -c R after checking PID via ps -fe | grep /bin/exec/R) in another terminal.

[HyukjinKwon]

I tested this on Mac.I will double check this and if this works on CentOS on Monday. What do you think about the current approach @felixcheung?

[SparkQA]

Test build #78203 has finished for PR 18320 at commit 74deb23.

• This patch fails SparkR unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #78204 has finished for PR 18320 at commit 1658a44.

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

[SparkQA]

Test build #78208 has finished for PR 18320 at commit 99245fb.

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

[HyukjinKwon]

@felixcheung, I double checked that the current status solves the issue in both CentOS and Mac.

[HyukjinKwon]

mcexit alone does not terminate the child.

With terminal A:

R

With terminal B:

watch -n 0.01 "ps -fe | grep /bin/exec/R"

with terminal A:

for(i in 0:100) {
  p <- parallel:::mcfork()
  if (inherits(p, "masterProcess")) {
    parallel:::mcexit(0L)
  }
}

[HyukjinKwon]

BTW, up to my knowledge, this should work alone, at least in C (< I didn't test). Even if it was a bug in R and is fixed in the future, I guess the current logics would still work with the current status.

arbitrary <- 9999
tools::pskill(arbitrary, tools::SIGUSR1)

https://stat.ethz.ch/R-manual/R-devel/library/tools/html/pskill.html

it silently ignores invalid values of its arguments, including zero or negative pids.

[HyukjinKwon]

cc @shivaram too, could you take a look here too? I am very careful of this change and want to make sure.

[felixcheung]

very cool, thank you for spending time to investigate.
agreed since this is so core to everything we should be careful with this change.

[felixcheung]

why not do this under if (ready) { similar to before?

[HyukjinKwon]

Ah, yes. It looks confusing. Probably, let me add more comments around here. I intended to do ...

A: When data (port number via inputCon) is arrived in daemon.R:

Before this change, each child terminated itself.

• ready is TRUE and terminates children, worker.R, in the parent if they finished their works.
• Launch other worker.R children.

B: When data (port number via inputCon) is not arrived in daemon.R:

It hung in this case before this change at socketSelect(...) and I introduced timeout = 1 to execute this case -

https://stat.ethz.ch/R-manual/R-devel/library/base/html/socketSelect.html

numeric or NULL. Time in seconds to wait for a socket to become available; NULL means wait indefinitely.

• ready is FALSE and terminates children, worker.R, in the parent if they finished their works, every second.

If we do this within if (ready) {, the child processes will remain until A case happens again.

Forked children do some jobs and if the parent checks if any is finished right after forking by parallel:::selectChildren() will probably not return the finished children PIDs.

[HyukjinKwon]

The related test was performed here - #18320 (comment) and #18320 (comment)

[felixcheung]

ah got it.
could you add some comment in the code to document the behavior?

also I wonder if delay in connecting back to the JVM will now cause the code to abort prematurely?

[HyukjinKwon]

Definitely for comments.

Maybe I missed your point. Children will only return their PID on exit and parallel:::selectChildren() will only return children PIDs that finished they work and called parallel:::mcexit(0L) (related test was done #18320 (comment)) up to my knowledge. So, even if connecting to JVM is delayed in worker.R and RRunner.scala, it won't matter.

[HyukjinKwon]

@felixcheung, I tested with the change in worker.R below:

 port <- as.integer(Sys.getenv("SPARKR_WORKER_PORT"))
+Sys.sleep(5L)
 inputCon <- socketConnection(
     port = port, blocking = TRUE, open = "rb", timeout = connectionTimeout)
 outputCon <- socketConnection(

It looks fine. Does this deal with your concern?

[HyukjinKwon]

retest this please

[SparkQA]

Test build #78517 has finished for PR 18320 at commit 53e00d7.

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

[SparkQA]

Test build #78518 has finished for PR 18320 at commit 6db1f61.

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

[viirya]

One path described above is:

1. Every second if any socket connection is not available and if there are child workers running.

So after forking succeeded and we want to do check per second, but after one second and if no children sent data, we turn back immediately to wait indefinitely. Doesn't it mean we don't actually do the logic every second but just for the first second?

[HyukjinKwon]

It becomes NULL if there are no children. It returns TRUE if there are children here and the timeout is reached. So, we always turn back to wait indefinitely only if there are no children at all.

https://stat.ethz.ch/R-manual/R-devel/library/parallel/html/children.html

TRUE is the timeout was reached, FALSE if an error occurred (e.g., if the master process was interrupted) or an integer vector of process IDs with children that have data available, or NULL if there are no children.

Would this answer your question (or did I maybe misunderstand) ?

[viirya]

It returns NULL if there are no children or there are no children that have data available?

[viirya]

I have this question because you have a comment:

Only the process IDs of children sent data to the parent are returned below.

So I am not sure if there are no children with data sent to the parent, the returned value is NULL or other?

[HyukjinKwon]

I ran several tests for this #18320 (comment) and #18320 (comment)

vi tmp.R

copy and paste

p <- parallel:::mcfork()
if (inherits(p, "masterProcess")) {
  Sys.sleep(3L)
  parallel:::mcexit(0L)
}

print("TRUE - the timeout was reached")
print("parallel:::selectChildren(timeout = 0):")
print(parallel:::selectChildren(timeout = 0))

and ...

Rscript tmp.R

prints ...

[1] "TRUE - the timeout was reached"
[1] "parallel:::selectChildren(timeout = 0):"
[1] TRUE

If data is not available, this should wait as much as timeout up to my knowledge rather than returning NULL.

[viirya]

Oh, cool. LGTM.

[viirya]

nit: If it is NULL, there are no such children. -> If it is NULL, there are no any children.?

[HyukjinKwon]

Definitely.

[SparkQA]

Test build #78519 has finished for PR 18320 at commit 7d61153.

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

[viirya]

LGTM

[SparkQA]

Test build #78523 has finished for PR 18320 at commit c5fe347.

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

[felixcheung]

ok LGTM again.
@shivaram there're fair bit of changes since your last LG, do you want to take a look?

[HyukjinKwon]

Thanks for a close look. I am pretty confident of the current state now.

[shivaram]

LGTM. The update looks good. Thanks for the thorough testing. We could also post a note on the dev list about this change and especially ask people who use dapply or gapply or the old RDD based API to test this out.

[felixcheung]

merged to master, thanks!!

[HyukjinKwon]

I think I am used to the symtoms here as I have tested many cases here. Let me try to organise and send a post to dev mailing list.

[HyukjinKwon]

Hmmmm. I carefully suspect this one, in particular, the last non-comment change causes unknown test failure with -10 error code occasionally If nothing happened to Jenkins. I believe -10 error starts around this time.

[HyukjinKwon]

Meanwhile, let me try to find the root cause.

[HyukjinKwon]

Let me leave a link #18456 for the related discussion.