diff --git a/dev/core/src/com/google/gwt/dev/shell/remoteui/MessageTransport.java b/dev/core/src/com/google/gwt/dev/shell/remoteui/MessageTransport.java index 7987cfc..3daf864 100644 --- a/dev/core/src/com/google/gwt/dev/shell/remoteui/MessageTransport.java +++ b/dev/core/src/com/google/gwt/dev/shell/remoteui/MessageTransport.java @@ -22,28 +22,176 @@ import com.google.gwt.dev.shell.remoteui.RemoteMessageProto.Message.Response; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.Callable; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; /** * Responsible for exchanging requests and responses between services. */ public class MessageTransport { + class PendingRequest extends PendingSend { + private final ReentrantLock lock = new ReentrantLock(); + private final Condition availableResponseCondition = lock.newCondition(); + private Response responseMessage; + private Exception exception; + private final Message message; + + public PendingRequest(Message message) { + this.message = message; + } + + @Override + public void failed(Exception e) { + pendingRequestMap.remove(message.getRequest().getRequestId()); + + lock.lock(); + try { + exception = e; + availableResponseCondition.signal(); + } finally { + lock.unlock(); + } + } + + @Override + public void send(OutputStream outputStream) throws IOException { + Request request = message.getRequest(); + int requestId = request.getRequestId(); + message.writeDelimitedTo(outputStream); + pendingRequestMap.put(requestId, this); + } + + /** + * Sets the response that was received from the server, and signals the + * thread that is waiting on the response. + * + * @param responseMessage the server's response + * @throws InterruptedException + */ + public void setResponse(Message.Response responseMessage) + throws InterruptedException { + assert (responseMessage != null); + lock.lock(); + try { + if (this.responseMessage != null) { + throw new IllegalStateException("Response has already been set."); + } + this.responseMessage = responseMessage; + availableResponseCondition.signal(); + } finally { + lock.unlock(); + } + } + + /** + * Waits for a response to be returned for a given request. + * + * @return the response from the server + * @throws Exception if an exception occurred while processing the request + */ + public Response waitForResponse() throws Exception { + lock.lock(); + + try { + while (responseMessage == null && exception == null) { + availableResponseCondition.await(); + } + + if (exception != null) { + throw exception; + } + + return responseMessage; + } finally { + lock.unlock(); + } + } + } + + static class PendingRequestMap { + private final Lock mapLock = new ReentrantLock(); + private final Map requestIdToPendingServerRequest = new HashMap(); + private boolean noMoreAdds; + + public void blockAdds(Exception e) { + mapLock.lock(); + try { + noMoreAdds = true; + for (PendingRequest pendingRequest : requestIdToPendingServerRequest.values()) { + pendingRequest.failed(e); + } + } finally { + mapLock.unlock(); + } + } + + public PendingRequest remove(int requestId) { + mapLock.lock(); + try { + return requestIdToPendingServerRequest.remove(requestId); + } finally { + mapLock.unlock(); + } + } + + void put(int requestId, PendingRequest pendingServerRequest) { + mapLock.lock(); + try { + if (noMoreAdds) { + // TODO: Flag the exception + pendingServerRequest.failed(new IllegalStateException( + "InputStream is closed")); + } else { + requestIdToPendingServerRequest.put(requestId, pendingServerRequest); + } + } finally { + mapLock.unlock(); + } + } + } + + class PendingResponse extends PendingSend { + Message message; + + public PendingResponse(Message message) { + this.message = message; + } + + @Override + public void failed(Exception e) { + // Do nothing + } + + @Override + public void send(OutputStream outputStream) throws IOException { + message.writeDelimitedTo(outputStream); + } + } + + abstract class PendingSend { + public abstract void failed(Exception e); + + public abstract void send(OutputStream outputStream) throws IOException; + } private static final int DEFAULT_SERVICE_THREADS = 2; private final Thread messageProcessingThread; private final AtomicInteger nextMessageId = new AtomicInteger(); - private final ConcurrentHashMap requestIdToPendingServerRequest = new ConcurrentHashMap(); private final RequestProcessor requestProcessor; - private final LinkedBlockingQueue sendQueue = new LinkedBlockingQueue(); + private final LinkedBlockingQueue sendQueue = new LinkedBlockingQueue(); private final Thread sendThread; private final ExecutorService serverRequestExecutor; + private final PendingRequestMap pendingRequestMap = new PendingRequestMap(); /** * Create a new instance using the given streams and request processor. @@ -59,7 +207,6 @@ public class MessageTransport { this.requestProcessor = requestProcessor; serverRequestExecutor = Executors.newFixedThreadPool(DEFAULT_SERVICE_THREADS); - // Terminated when the inputStream is closed messageProcessingThread = new Thread(new Runnable() { public void run() { try { @@ -68,9 +215,9 @@ public class MessageTransport { processMessage(message); } } catch (IOException e) { - terminateDueToException(); + terminateDueToException(e); } catch (InterruptedException e) { - terminateDueToException(); + terminateDueToException(e); } } }); @@ -79,18 +226,21 @@ public class MessageTransport { // Terminated when the outputStream is closed sendThread = new Thread(new Runnable() { public void run() { - try { - while (true) { - Message msgToSend = sendQueue.take(); - msgToSend.writeDelimitedTo(outputStream); + while (true) { + try { + PendingSend pendingSend = sendQueue.take(); + try { + pendingSend.send(outputStream); + } catch (IOException e) { + pendingSend.failed(e); + } + } catch (InterruptedException e) { + break; } - } catch (InterruptedException e) { - terminateDueToException(); - } catch (IOException e) { - terminateDueToException(); } } }); + sendThread.setDaemon(true); sendThread.start(); } @@ -113,14 +263,11 @@ public class MessageTransport { messageBuilder.setMessageType(Message.MessageType.REQUEST); messageBuilder.setRequest(requestBuilder); - PendingServerRequest pendingServerRequest = new PendingServerRequest(); - requestIdToPendingServerRequest.put(requestId, pendingServerRequest); - Message message = messageBuilder.build(); + PendingRequest pendingRequest = new PendingRequest(message); + sendQueue.put(pendingRequest); - sendQueue.put(message); - - return pendingServerRequest.waitForResponse(); + return pendingRequest.waitForResponse(); } }); @@ -146,7 +293,9 @@ public class MessageTransport { responseBuilder.setRequestId(requestId); messageBuilder.setResponse(responseBuilder); - sendQueue.put(messageBuilder.build()); + PendingResponse pendingResponse = new PendingResponse( + messageBuilder.build()); + sendQueue.put(pendingResponse); } private void processMessage(final Message message) @@ -172,14 +321,15 @@ public class MessageTransport { } } - private void processServerResponse(int requestId, Response response) { - PendingServerRequest pendingServerRequest = requestIdToPendingServerRequest.remove(requestId); + private void processServerResponse(int requestId, Response response) + throws InterruptedException { + PendingRequest pendingServerRequest = pendingRequestMap.remove(requestId); if (pendingServerRequest != null) { pendingServerRequest.setResponse(response); } } - private void terminateDueToException() { - serverRequestExecutor.shutdownNow(); + private void terminateDueToException(Exception e) { + pendingRequestMap.blockAdds(e); } } diff --git a/dev/core/src/com/google/gwt/dev/shell/remoteui/PendingServerRequest.java b/dev/core/src/com/google/gwt/dev/shell/remoteui/PendingServerRequest.java deleted file mode 100644 index 0019b28..0000000 --- a/dev/core/src/com/google/gwt/dev/shell/remoteui/PendingServerRequest.java +++ /dev/null @@ -1,57 +0,0 @@ -package com.google.gwt.dev.shell.remoteui; - -import com.google.gwt.dev.shell.remoteui.RemoteMessageProto.Message; -import com.google.gwt.dev.shell.remoteui.RemoteMessageProto.Message.Response; - -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.ReentrantLock; - -/** - * Represents an outstanding server request. - */ -class PendingServerRequest { - private final ReentrantLock lock = new ReentrantLock(); - private final Condition availableResponseCondition = lock.newCondition(); - private Response responseMessage = null; - - /** - * Sets the response that was received from the server, and signals the thread - * that is waiting on the response. - * - * @param responseMessage the server's response - */ - public void setResponse(Message.Response responseMessage) { - assert (responseMessage != null); - lock.lock(); - try { - if (this.responseMessage != null) { - throw new IllegalStateException("Response has already been set."); - } - this.responseMessage = responseMessage; - availableResponseCondition.signal(); - } finally { - lock.unlock(); - } - } - - /** - * Waits for a response to be returned for a given request. - * - * @return the response from the server - * @throws InterruptedException - * - * @throws Exception if an exception occurred while processing the request - */ - public Response waitForResponse() throws InterruptedException { - lock.lock(); - - try { - while (responseMessage == null) { - availableResponseCondition.await(); - } - return responseMessage; - } finally { - lock.unlock(); - } - } -} \ No newline at end of file diff --git a/dev/core/test/com/google/gwt/dev/shell/remoteui/MessageTransportTest.java b/dev/core/test/com/google/gwt/dev/shell/remoteui/MessageTransportTest.java index faa5530..2c2840b 100644 --- a/dev/core/test/com/google/gwt/dev/shell/remoteui/MessageTransportTest.java +++ b/dev/core/test/com/google/gwt/dev/shell/remoteui/MessageTransportTest.java @@ -14,7 +14,6 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -176,16 +175,7 @@ public class MessageTransportTest extends TestCase { network.getServerSocket().getInputStream().close(); Future responseFuture = null; - try { - responseFuture = messageTransport.executeRequestAsync(request); - } catch (RejectedExecutionException re) { - // It is possible that the shutdown of the executor service - // occurred before the request could be submitted; in such a - // case, we're not able to submit the request, and this - // is the appropriate exception that we'd get. - return; - } - + responseFuture = messageTransport.executeRequestAsync(request); assertNotNull(responseFuture); try { @@ -195,7 +185,7 @@ public class MessageTransportTest extends TestCase { fail("Should not have timed out"); } catch (ExecutionException e) { // If the submission was not rejected, this is where we should hit - assertTrue(e.getCause() instanceof InterruptedException); + assertTrue(e.getCause() instanceof IllegalStateException); } catch (Exception e) { fail("Should not have thrown any other exception"); } @@ -343,20 +333,20 @@ public class MessageTransportTest extends TestCase { requestMessageBuilder.setServiceType(Message.Request.ServiceType.DEV_MODE); Message.Request request = requestMessageBuilder.build(); - Future responseFuture = messageTransport.executeRequestAsync(request); - assertNotNull(responseFuture); + // assertNotNull(response); // This will close the client's input stream network.getServerSocket().getOutputStream().close(); try { - responseFuture.get(2, TimeUnit.SECONDS); + Future response = messageTransport.executeRequestAsync(request); + response.get(10000, TimeUnit.SECONDS); fail("Should have thrown an exception"); } catch (TimeoutException te) { fail("Should not have timed out"); } catch (ExecutionException e) { // This is where we should hit - assertTrue(e.getCause() instanceof InterruptedException); + assertTrue(e.getCause() instanceof IllegalStateException); } catch (Exception e) { fail("Should not have thrown any other exception"); } diff --git a/eclipse/dev/.classpath b/eclipse/dev/.classpath index 9e5950e..5bf1714 100644 --- a/eclipse/dev/.classpath +++ b/eclipse/dev/.classpath @@ -6,7 +6,7 @@ - + @@ -42,6 +42,6 @@ - +