From 4afa2a96fc49d9331cd60ea8ef919169ce6f46ea Mon Sep 17 00:00:00 2001 From: kohsuke Date: Sun, 21 Oct 2007 23:39:03 +0000 Subject: [PATCH] when a thread that's waiting for a remote operation is interrupted, also interrupt the remote thread that's executing it, so that the actual operation gets cancelled. (#904) git-svn-id: https://hudson.dev.java.net/svn/hudson/trunk/hudson/main@5393 71c3de6d-444a-0410-be80-ed276b4c234a --- .../main/java/hudson/remoting/Channel.java | 14 ++++- .../main/java/hudson/remoting/Request.java | 56 ++++++++++++++++--- 2 files changed, 60 insertions(+), 10 deletions(-) diff --git a/remoting/src/main/java/hudson/remoting/Channel.java b/remoting/src/main/java/hudson/remoting/Channel.java index a26943b04a..4997e34246 100644 --- a/remoting/src/main/java/hudson/remoting/Channel.java +++ b/remoting/src/main/java/hudson/remoting/Channel.java @@ -10,11 +10,13 @@ import java.io.ObjectOutputStream; import java.io.OutputStream; import java.io.PrintWriter; import java.lang.reflect.Proxy; +import java.util.Collections; import java.util.Hashtable; import java.util.Map; import java.util.Vector; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.logging.Level; import java.util.logging.Logger; @@ -73,7 +75,7 @@ public class Channel implements VirtualChannel { private final ObjectInputStream ois; private final ObjectOutputStream oos; private final String name; - /*package*/ final Executor executor; + /*package*/ final ExecutorService executor; /** * If true, the incoming link is already shut down, @@ -88,6 +90,12 @@ public class Channel implements VirtualChannel { /*package*/ final Map> pendingCalls = new Hashtable>(); + /** + * Records the {@link Request}s being executed on this channel, sent by the remote peer. + */ + /*package*/ final Map> executingCalls = + Collections.synchronizedMap(new Hashtable>()); + /** * {@link ClassLoader}s that are proxies of the remote classloaders. */ @@ -104,7 +112,7 @@ public class Channel implements VirtualChannel { private final Vector listeners = new Vector(); private int gcCounter; - public Channel(String name, Executor exec, InputStream is, OutputStream os) throws IOException { + public Channel(String name, ExecutorService exec, InputStream is, OutputStream os) throws IOException { this(name,exec,is,os,null); } @@ -125,7 +133,7 @@ public class Channel implements VirtualChannel { * when the established communication channel might include some data that might * be useful for debugging/trouble-shooting. */ - public Channel(String name, Executor exec, InputStream is, OutputStream os, OutputStream header) throws IOException { + public Channel(String name, ExecutorService exec, InputStream is, OutputStream os, OutputStream header) throws IOException { this.name = name; this.executor = exec; diff --git a/remoting/src/main/java/hudson/remoting/Request.java b/remoting/src/main/java/hudson/remoting/Request.java index 4bdb7b891d..e2fbd8fb76 100644 --- a/remoting/src/main/java/hudson/remoting/Request.java +++ b/remoting/src/main/java/hudson/remoting/Request.java @@ -3,6 +3,7 @@ package hudson.remoting; import java.io.IOException; import java.io.Serializable; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.logging.Level; @@ -19,7 +20,6 @@ import java.util.logging.Logger; * @see Response */ abstract class Request extends Command { - /** * Executed on a remote system to perform the task. * @@ -43,6 +43,12 @@ abstract class Request extends C private volatile Response response; + /** + * While executing the call this is set to the handle of the execution. + */ + private volatile transient Future future; + + protected Request() { synchronized(Request.class) { id = nextId++; @@ -77,8 +83,14 @@ abstract class Request extends C } synchronized(this) { - while(response==null) - wait(); // wait until the response arrives + try { + while(response==null) + wait(); // wait until the response arrives + } catch (InterruptedException e) { + // if we are cancelled, abort the remote computation, too + channel.send(new Cancel(id)); + throw e; + } Object exc = response.exception; @@ -100,13 +112,13 @@ abstract class Request extends C * @throws IOException * If there's an error during the communication. */ - public final Future callAsync(Channel channel) throws IOException { + public final hudson.remoting.Future callAsync(final Channel channel) throws IOException { response=null; channel.pendingCalls.put(id,this); channel.send(this); - return new Future() { + return new hudson.remoting.Future() { /** * The task cannot be cancelled. */ @@ -124,8 +136,17 @@ abstract class Request extends C public RSP get() throws InterruptedException, ExecutionException { synchronized(Request.this) { - while(response==null) + try { + while(response==null) Request.this.wait(); // wait until the response arrives + } catch (InterruptedException e) { + try { + channel.send(new Cancel(id)); + } catch (IOException e1) { + // couldn't cancel. ignore. + } + throw e; + } if(response.exception!=null) throw new ExecutionException(response.exception); @@ -170,7 +191,8 @@ abstract class Request extends C * Schedules the execution of this request. */ protected final void execute(final Channel channel) { - channel.executor.execute(new Runnable() { + channel.executingCalls.put(id,this); + future = channel.executor.submit(new Runnable() { public void run() { try { RSP rsp; @@ -187,6 +209,8 @@ abstract class Request extends C // communication error. // this means the caller will block forever logger.log(Level.SEVERE, "Failed to send back a reply",e); + } finally { + channel.executingCalls.remove(id); } } }); @@ -214,4 +238,22 @@ abstract class Request extends C // throw new Error(e); // } //} + + /** + * Interrupts the execution of the remote computation. + */ + private static final class Cancel extends Command { + private final int id; + + Cancel(int id) { + this.id = id; + } + + protected void execute(Channel channel) { + Request r = channel.executingCalls.get(id); + if(r==null) return; // already completed + Future f = r.future; + if(f!=null) f.cancel(true); + } + } }