[SPARK-9026] [SPARK-4514] Modifications to JobWaiter, FutureAction, and AsyncRDDActions to support non-blocking operation

[reggert]

These changes rework the implementations of SimpleFutureAction, ComplexFutureAction, JobWaiter, and AsyncRDDActions such that asynchronous callbacks on the generated Futures NEVER block waiting for a job to complete. A small amount of mutex synchronization is necessary to protect the internal fields that manage cancellation, but these locks are only held very briefly and in practice should almost never cause any blocking to occur. The existing blocking APIs of these classes are retained, but they simply delegate to the underlying non-blocking API and Await the results with indefinite timeouts.

Associated JIRA ticket: https://issues.apache.org/jira/browse/SPARK-9026
Also fixes: https://issues.apache.org/jira/browse/SPARK-4514

This pull request contains all my own original work, which I release to the Spark project under its open source license.

[SparkQA]

Test build #1952 has finished for PR 9264 at commit 1ad1abd.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[JoshRosen]

This seems like it overlaps with #7385, an old PR of mine which was put on hold when I became too busy with other tickets (and when the original motivation for it was worked around via another means). This would still be good to fix, though, and I'm happy to work with you to try to do this.

[markhamstra]

@JoshRosen Agreed that this is largely a duplicate. One of SPARK-9026 and SPARK-11296 should be marked as a dup and closed. And fwiw, at least the test failures on the jobFailed race that you were dealing with in that PR have since been dealt with in a separate PR, recently merged.

[reggert]

I agree that the JIRA issue is a duplicate. Odd that the old one didn't turn up for it when I searched for existing issues. I've linked the issues in JIRA.

[reggert]

It looks like our approaches are very similar, though my changes to JobWaiter and SimpleFutureAction are a bit smaller, and I also included fixes for ComplexFutureAction and AsyncRDDFunctions.takeAsync. I haven't been able to fully test mine yet, though, because doing a full build of Spark causes my junky old laptop to go into seizures.

[reggert]

I've added some unit tests that verify that my changes, do, indeed, stop callbacks from consuming threads before the jobs have completed.

[reggert]

What else needs to be done to get this PR merged?

[squito]

nit: I know you didn't introduce this, but as long as you're touching it -- the scala imports should go above the spark imports (still in their own group)

[squito]

Hi @reggert , I'm not the best person to review this, so I can't definitely say what needs to be done to get it merged. I pointed out a few minor style issues just to keep things moving along. You also need to resolve conflicts with master.

[squito]

Jenkins, ok to test

[SparkQA]

Test build #45395 has finished for PR 9264 at commit 93b3065.

• This patch fails Scala style tests.
• This patch does not merge cleanly.
• This patch adds no public classes.

[andrewor14]

why was this removed?

[reggert]

Git somehow managed to screw up the merge when I re-sync'd with the upstream master. I'll resurrect it.

[andrewor14]

@reggert I've closed your issue as a duplicate. Can you remove it from the title and use only SPARK-9026 to avoid potential confusion?

[reggert]

@squito I merged with master 2 days ago, and more conflicts were introduced on master less than 24 hours later. I'll merge one more time in order to fix the issue with the accidentally-deleted Utils class, but I won't do it again after that until it's the only thing left to do.

[SparkQA]

Test build #45491 has finished for PR 9264 at commit dcd2883.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[reggert]

Back up to date with master. All style issues should be resolved.

[reggert]

SparkQA seems to be happy now.
@JoshRosen What else needs to be done to move forward on this? The current implementation on master requires a dedicated thread pool for all callbacks on FutureActions to avoid depleting the global ExecutionContext (and this fact is entirely undocumented), so it would be good to have these changes in 1.6.

[reggert]

It appears that this PR may also fix SPARK-4514.

[reggert]

Correction: This PR only fixes SPARK-4514 as long as the ComplexFutureAction only spawns a single job. The job group information is lost in subsequent jobs.

[reggert]

I've implemented a two-line fix for SPARK-4514. Note that trying to fix it outside of this PR would have resulted in a merge conflict.

[SparkQA]

Test build #46498 has finished for PR 9264 at commit 5816489.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * trait JobSubmitter\n * class ComplexFutureAction[T](run : JobSubmitter => Future[T])\n

[SparkQA]

Test build #46499 has finished for PR 9264 at commit 8fe8000.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * trait JobSubmitter\n * class ComplexFutureAction[T](run : JobSubmitter => Future[T])\n

[JoshRosen]

@zsxwing, could you take a look at this patch? I think you'd be a good reviewer for this given your experience with similar patches in streaming.

[reggert]

Any feedback?

[reggert]

@zsxwing @JoshRosen I just want to make sure that you guys haven't forgotten about this. I haven't heard anything in a week and a half.

[reggert]

All I want for Christmas is a code review. :-)

[zsxwing]

Since the callback doesn't run, how to verify that we didn't break task deserialization?

[reggert]

Unless I'm mistaken, the Await.result call will throw an exception if the job fails.

[reggert]

To clarify, in the event that we've broken task deserialization, I would expect to see a NotSerializableException or similar error thrown from the Await.result call.

[zsxwing]

Just tested the logic locally. Task not serializable will be throw from mapPartitions if I added some non-serializable reference to the closure.

Since no place will call executionContextInvoked.failure, the only exception will be thrown from Await.result(executionContextInvoked.future, atMost = 15.seconds) is just TimeoutException.

[markhamstra]

How did you test it? My expectation is the same as Richard's, that Await.result will throw the underlying exception. For example:

scala> val fi: Future[Int] = Future { throw new java.lang.RuntimeException() }
fi: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@3f200884

scala> try { Await.result(fi, 3 seconds) } catch {
     | case _: java.util.concurrent.TimeoutException => println("Timeout")
     | case _: java.lang.RuntimeException => println("Runtime")
     | case _: Throwable => println("Huh?")
     | }
Runtime
res1: AnyVal = ()

[zsxwing]

The problem is here it uses executionContextInvoked.future to call Await.result. But the only place that completes the Promise is executionContextInvoked.success(()).

[markhamstra]

Ah, got it -- yes, that does appear to be swallowing any exception from mapPartitions. Good eyes!

[reggert]

If mapPartitions throws an exception, won't that cause the test to fail before we even get to Await.result?

[zsxwing]

If mapPartitions throws an exception, won't that cause the test to fail before we even get to Await.result?

I'm not against Await.result here. I just wanted to point out the comment is wrong.

[reggert]

Oh, is that all? :-)

Comment line removed.

[zsxwing]

@reggert sorry for the delay. LGTM except a nit for test. Thanks very much.

[SparkQA]

Test build #47629 has finished for PR 9264 at commit 539ac43.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * trait JobSubmitter\n * class ComplexFutureAction[T](run : JobSubmitter => Future[T])\n

[zsxwing]

LGTM

[andrewor14]

Thanks, merging into master.

[andrewor14]

By the way, @reggert how is this related to SPARK-4514?

[reggert]

@andrewor14 I included a couple lines to propagate the local properties (from the thread that created the ComplexFutureAction) to each of the spawned jobs (all but the first of which may be launched from other threads, losing the local properties unless setLocalProperties is called). AFAIK this resolves SPARK-4514.