You are viewing a plain text version of this content. The canonical link for it is here.

Posted to reviews@livy.apache.org by bjoernlohrmann <gi...@git.apache.org> on 2018/11/13 22:21:27 UTC

[GitHub] incubator-livy pull request #128: Use setJobGroup/cancelJobGroup to cancel b...

GitHub user bjoernlohrmann opened a pull request:

    https://github.com/apache/incubator-livy/pull/128

    Use setJobGroup/cancelJobGroup to cancel bypass jobs

    ## What changes were proposed in this pull request?
    
    This PR fixes [LIVY-533](https://issues.apache.org/jira/browse/LIVY-533).
    
    The proposed fix invokes setJobGroup() in the worker thread before calling job code, and cancelJobGroup() when another thread tries to cancel job execution. synchronized blocks are employed to guard against race conditions.
    
    Also, this PR removes some unused code from the JobWrapper class.
    
    ## How was this patch tested?
    
    Manual as well as Livy test suite. No new tests were added.


You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/bjoernlohrmann/incubator-livy master

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/incubator-livy/pull/128.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 #128
    
----
commit 989225990ed139d2f5ed35e89e4dcbcead96b4dc
Author: Bjoern Lohrmann <53...@...>
Date:   2018-11-13T09:37:30Z

    Use setJobGroup/cancelJobGroup to cancel bypass jobs

----


---

[GitHub] incubator-livy pull request #128: [LIVY-533] Use setJobGroup/cancelJobGroup ...

Posted by asfgit <gi...@git.apache.org>.

Github user asfgit closed the pull request at:

    https://github.com/apache/incubator-livy/pull/128


---

[GitHub] incubator-livy pull request #128: [LIVY-533] Use setJobGroup/cancelJobGroup ...

Posted by vanzin <gi...@git.apache.org>.

Github user vanzin commented on a diff in the pull request:

    https://github.com/apache/incubator-livy/pull/128#discussion_r238385396
  
    --- Diff: rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java ---
    @@ -490,20 +487,72 @@ public void call(LivyClient client) throws Exception {
             String resultVal = (String) s.deserialize(ByteBuffer.wrap(status.result));
             assertEquals("hello", resultVal);
     
    -        // After the result is retrieved, the driver should stop tracking the job and release
    -        // resources associated with it.
    -        try {
    -          lclient.getBypassJobStatus(jobId).get(TIMEOUT, TimeUnit.SECONDS);
    -          fail("Should have failed to retrieve status of released job.");
    -        } catch (ExecutionException ee) {
    -          assertTrue(ee.getCause() instanceof RpcException);
    -          assertTrue(ee.getCause().getMessage().contains(
    -            "java.util.NoSuchElementException: " + jobId));
    +        assertJobIdUntracked(lclient, jobId);
    +      }
    +    });
    +  }
    +
    +  private void assertJobIdUntracked(RSCClient lclient, String jobId) throws InterruptedException,
    +          TimeoutException {
    --- End diff --
    
    nit: indented too far. I also slightly prefer to keep all the thrown exceptions in the same line.


---

[GitHub] incubator-livy pull request #128: [LIVY-533] Use setJobGroup/cancelJobGroup ...

Posted by vanzin <gi...@git.apache.org>.

Github user vanzin commented on a diff in the pull request:

    https://github.com/apache/incubator-livy/pull/128#discussion_r237956271
  
    --- Diff: rsc/src/main/java/org/apache/livy/rsc/driver/JobWrapper.java ---
    @@ -38,21 +34,31 @@
       public final String jobId;
     
       private final RSCDriver driver;
    +
       private final Job<T> job;
    -  private final AtomicInteger completed;
    +
    +  private boolean isCancelled = false;
     
       private Future<?> future;
     
       public JobWrapper(RSCDriver driver, String jobId, Job<T> job) {
         this.driver = driver;
         this.jobId = jobId;
         this.job = job;
    -    this.completed = new AtomicInteger();
       }
     
       @Override
       public Void call() throws Exception {
         try {
    +      // this is synchronized to avoid races with cancel()
    +      synchronized (this) {
    +        if (isCancelled) {
    +          throw new Exception("Job isCancelled");
    --- End diff --
    
    Throw `CancellationException` instead.


---

[GitHub] incubator-livy pull request #128: [LIVY-533] Use setJobGroup/cancelJobGroup ...

Posted by bjoernlohrmann <gi...@git.apache.org>.

Github user bjoernlohrmann commented on a diff in the pull request:

    https://github.com/apache/incubator-livy/pull/128#discussion_r238455398
  
    --- Diff: rsc/src/main/java/org/apache/livy/rsc/driver/BypassJobWrapper.java ---
    @@ -38,28 +38,50 @@ public BypassJobWrapper(RSCDriver driver, String jobId, Job<byte[]> serializedJo
     
       @Override
       public Void call() throws Exception {
    -    state = JobHandle.State.STARTED;
    +    // we ignore the return value here, because super.call() will also detect cancellation
    +    // and needs to perform some cleanup
    +    tryTransitionToState(JobHandle.State.STARTED);
    --- End diff --
    
    That is actually the standard case "job is cancelled while running" and should work as intended.


---

[GitHub] incubator-livy pull request #128: [LIVY-533] Use setJobGroup/cancelJobGroup ...

Posted by bjoernlohrmann <gi...@git.apache.org>.

Github user bjoernlohrmann commented on a diff in the pull request:

    https://github.com/apache/incubator-livy/pull/128#discussion_r237280060
  
    --- Diff: rsc/src/main/java/org/apache/livy/rsc/driver/JobWrapper.java ---
    @@ -39,20 +39,27 @@
     
       private final RSCDriver driver;
       private final Job<T> job;
    -  private final AtomicInteger completed;
     
    -  private Future<?> future;
    +  private volatile Future<?> future;
     
       public JobWrapper(RSCDriver driver, String jobId, Job<T> job) {
         this.driver = driver;
         this.jobId = jobId;
         this.job = job;
    -    this.completed = new AtomicInteger();
       }
     
       @Override
       public Void call() throws Exception {
         try {
    +      // this is synchronized to avoid races with cancel()
    --- End diff --
    
    `future` will only be null for synchronous bypass jobs, but those cannot be cancelled unless I am mistaken?
    
    For async bypass jobs, I think `future` should always be non-null during `call()/cancel()`, because it is assigned in `submit()` which also has the synchronized modifier and always runs before `call()` or `cancel()`.
    
    I can add a boolean cancelled flag if you prefer, but this feels like duplicating information that is already available in the future.


---

[GitHub] incubator-livy pull request #128: [LIVY-533] Use setJobGroup/cancelJobGroup ...

Posted by vanzin <gi...@git.apache.org>.

Github user vanzin commented on a diff in the pull request:

    https://github.com/apache/incubator-livy/pull/128#discussion_r238384975
  
    --- Diff: rsc/src/main/java/org/apache/livy/rsc/driver/BypassJobWrapper.java ---
    @@ -38,28 +38,50 @@ public BypassJobWrapper(RSCDriver driver, String jobId, Job<byte[]> serializedJo
     
       @Override
       public Void call() throws Exception {
    -    state = JobHandle.State.STARTED;
    +    // we ignore the return value here, because super.call() will also detect cancellation
    +    // and needs to perform some cleanup
    +    tryTransitionToState(JobHandle.State.STARTED);
    +
         return super.call();
       }
     
       @Override
       protected synchronized void finished(byte[] result, Throwable error) {
    -    if (error == null) {
    +    if (error == null && tryTransitionToState(JobHandle.State.SUCCEEDED)) {
           this.result = result;
    -      this.state = JobHandle.State.SUCCEEDED;
    -    } else {
    +    } else if (error != null && tryTransitionToState(JobHandle.State.FAILED)) {
           this.error = error;
    -      this.state = JobHandle.State.FAILED;
         }
       }
     
       @Override
    -  boolean cancel() {
    -    if (super.cancel()) {
    -      this.state = JobHandle.State.CANCELLED;
    -      return true;
    +  synchronized boolean cancel() {
    +    return tryTransitionToState(JobHandle.State.CANCELLED) && super.cancel();
    +  }
    +
    +  private synchronized boolean tryTransitionToState(JobHandle.State newState) {
    +    boolean success = false;
    +
    +    switch (this.state) {
    +      case QUEUED:
    +        if (newState == JobHandle.State.STARTED || newState == JobHandle.State.CANCELLED) {
    +          this.state = newState;
    +          success = true;
    +        }
    +        break;
    +      case STARTED:
    +        if (newState == JobHandle.State.CANCELLED
    +                || newState == JobHandle.State.SUCCEEDED
    --- End diff --
    
    nit: indented too far


---

[GitHub] incubator-livy pull request #128: [LIVY-533] Use setJobGroup/cancelJobGroup ...

Posted by bjoernlohrmann <gi...@git.apache.org>.

Github user bjoernlohrmann commented on a diff in the pull request:

    https://github.com/apache/incubator-livy/pull/128#discussion_r238458802
  
    --- Diff: rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java ---
    @@ -490,20 +487,72 @@ public void call(LivyClient client) throws Exception {
             String resultVal = (String) s.deserialize(ByteBuffer.wrap(status.result));
             assertEquals("hello", resultVal);
     
    -        // After the result is retrieved, the driver should stop tracking the job and release
    -        // resources associated with it.
    -        try {
    -          lclient.getBypassJobStatus(jobId).get(TIMEOUT, TimeUnit.SECONDS);
    -          fail("Should have failed to retrieve status of released job.");
    -        } catch (ExecutionException ee) {
    -          assertTrue(ee.getCause() instanceof RpcException);
    -          assertTrue(ee.getCause().getMessage().contains(
    -            "java.util.NoSuchElementException: " + jobId));
    +        assertJobIdUntracked(lclient, jobId);
    +      }
    +    });
    +  }
    +
    +  private void assertJobIdUntracked(RSCClient lclient, String jobId) throws InterruptedException,
    +          TimeoutException {
    +
    +    // After the result is retrieved, the driver should stop tracking the job and release
    +    // resources associated with it.
    +    try {
    +      lclient.getBypassJobStatus(jobId).get(TIMEOUT, TimeUnit.SECONDS);
    +      fail("Should have failed to retrieve status of released job.");
    +    } catch (ExecutionException ee) {
    +      assertTrue(ee.getCause() instanceof RpcException);
    +      assertTrue(ee.getCause().getMessage().contains(
    +              "java.util.NoSuchElementException: " + jobId));
    +    }
    +  }
    +
    +  @Test
    +  public void testBypassWithImmediateCancellation() throws Exception {
    +    runBypassCancellationTest(0, 1000);
    +  }
    +
    +  @Test
    +  public void testBypassWithCancellation() throws Exception {
    +    runBypassCancellationTest(200, 1000);
    +  }
    +
    +  public void runBypassCancellationTest(long waitBeforeCancel, long jobDuration) throws Exception {
    --- End diff --
    
    Yes, cancel() interrupts the sleeper job.
    
    I upped the job duration to one minute which I hope is enough. I am hesitant to make the job duration infinite. If cancellation fails due to some bug, the test will hang forever (or until some timeout kills the build) -- which is would be very annoying.


---

[GitHub] incubator-livy pull request #128: [LIVY-533] Use setJobGroup/cancelJobGroup ...

Posted by vanzin <gi...@git.apache.org>.

Github user vanzin commented on a diff in the pull request:

    https://github.com/apache/incubator-livy/pull/128#discussion_r238387799
  
    --- Diff: rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java ---
    @@ -490,20 +487,72 @@ public void call(LivyClient client) throws Exception {
             String resultVal = (String) s.deserialize(ByteBuffer.wrap(status.result));
             assertEquals("hello", resultVal);
     
    -        // After the result is retrieved, the driver should stop tracking the job and release
    -        // resources associated with it.
    -        try {
    -          lclient.getBypassJobStatus(jobId).get(TIMEOUT, TimeUnit.SECONDS);
    -          fail("Should have failed to retrieve status of released job.");
    -        } catch (ExecutionException ee) {
    -          assertTrue(ee.getCause() instanceof RpcException);
    -          assertTrue(ee.getCause().getMessage().contains(
    -            "java.util.NoSuchElementException: " + jobId));
    +        assertJobIdUntracked(lclient, jobId);
    +      }
    +    });
    +  }
    +
    +  private void assertJobIdUntracked(RSCClient lclient, String jobId) throws InterruptedException,
    +          TimeoutException {
    +
    +    // After the result is retrieved, the driver should stop tracking the job and release
    +    // resources associated with it.
    +    try {
    +      lclient.getBypassJobStatus(jobId).get(TIMEOUT, TimeUnit.SECONDS);
    +      fail("Should have failed to retrieve status of released job.");
    +    } catch (ExecutionException ee) {
    +      assertTrue(ee.getCause() instanceof RpcException);
    +      assertTrue(ee.getCause().getMessage().contains(
    +              "java.util.NoSuchElementException: " + jobId));
    +    }
    +  }
    +
    +  @Test
    +  public void testBypassWithImmediateCancellation() throws Exception {
    +    runBypassCancellationTest(0, 1000);
    +  }
    +
    +  @Test
    +  public void testBypassWithCancellation() throws Exception {
    +    runBypassCancellationTest(200, 1000);
    +  }
    +
    +  public void runBypassCancellationTest(long waitBeforeCancel, long jobDuration) throws Exception {
    --- End diff --
    
    I'm generally a bit wary of tests that rely on timing, especially when it's a little tight. Overloaded build machines may make this test flaky.
    
    Can the job duration be much larger than the wait, or even basically "infinite"? The cancel or the shutting down of the client should interrupt the sleep, right?


---