8144906: Fix jshell's ToolBasicTest

Various fixes to fix the ToolBasicTest - line endings normalization, ordering for output from RemoteAgent, synchronization.

Reviewed-by: rfield

langtools/src/jdk.jshell/share/classes/jdk/internal/jshell/remote/RemoteAgent.java

@@ -28,6 +28,9 @@ import java.io.File;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.io.UnsupportedEncodingException;
 import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -35,7 +38,9 @@ import java.net.Socket;
 
 import java.util.ArrayList;
 import java.util.List;
+
 import static jdk.internal.jshell.remote.RemoteCodes.*;
+
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -59,7 +64,10 @@ class RemoteAgent {
     void commandLoop(Socket socket) throws IOException {
         // in before out -- so we don't hang the controlling process
         ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
-        ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
+        OutputStream socketOut = socket.getOutputStream();
+        System.setOut(new PrintStream(new MultiplexingOutputStream("out", socketOut), true));
+        System.setErr(new PrintStream(new MultiplexingOutputStream("err", socketOut), true));
+        ObjectOutputStream out = new ObjectOutputStream(new MultiplexingOutputStream("command", socketOut));
         while (true) {
             int cmd = in.readInt();
             switch (cmd) {
@@ -260,4 +268,64 @@ class RemoteAgent {
         }
         return sb.toString();
     }
+
+    private static final class MultiplexingOutputStream extends OutputStream {
+
+        private static final int PACKET_SIZE = 127;
+
+        private final byte[] name;
+        private final OutputStream delegate;
+
+        public MultiplexingOutputStream(String name, OutputStream delegate) {
+            try {
+                this.name = name.getBytes("UTF-8");
+                this.delegate = delegate;
+            } catch (UnsupportedEncodingException ex) {
+                throw new IllegalStateException(ex); //should not happen
+            }
+        }
+
+        @Override
+        public void write(int b) throws IOException {
+            synchronized (delegate) {
+                delegate.write(name.length); //assuming the len is small enough to fit into byte
+                delegate.write(name);
+                delegate.write(1);
+                delegate.write(b);
+                delegate.flush();
+            }
+        }
+
+        @Override
+        public void write(byte[] b, int off, int len) throws IOException {
+            synchronized (delegate) {
+                int i = 0;
+                while (len > 0) {
+                    int size = Math.min(PACKET_SIZE, len);
+
+                    delegate.write(name.length); //assuming the len is small enough to fit into byte
+                    delegate.write(name);
+                    delegate.write(size);
+                    delegate.write(b, off + i, size);
+                    i += size;
+                    len -= size;
+                }
+
+                delegate.flush();
+            }
+        }
+
+        @Override
+        public void flush() throws IOException {
+            super.flush();
+            delegate.flush();
+        }
+
+        @Override
+        public void close() throws IOException {
+            super.close();
+            delegate.close();
+        }
+
+    }
 }

langtools/src/jdk.jshell/share/classes/jdk/jshell/ExecutionControl.java

@@ -26,9 +26,12 @@
 package jdk.jshell;
 
 import static jdk.internal.jshell.remote.RemoteCodes.*;
+import java.io.DataInputStream;
+import java.io.InputStream;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.io.PrintStream;
 import java.net.ServerSocket;
 import java.net.Socket;
 import com.sun.jdi.*;
@@ -69,7 +72,9 @@ class ExecutionControl {
             socket = listener.accept();
             // out before in -- match remote creation so we don't hang
             out = new ObjectOutputStream(socket.getOutputStream());
-            in = new ObjectInputStream(socket.getInputStream());
+            PipeInputStream commandIn = new PipeInputStream();
+            new DemultiplexInput(socket.getInputStream(), commandIn, proc.out, proc.err).start();
+            in = new ObjectInputStream(commandIn);
         }
     }
 
@@ -117,11 +122,13 @@ class ExecutionControl {
                 String result = in.readUTF();
                 return result;
             }
-        } catch (EOFException ex) {
-            env.shutdown();
         } catch (IOException | ClassNotFoundException ex) {
+            if (!env.connection().isRunning()) {
+                env.shutdown();
+            } else {
                 proc.debug(DBG_GEN, "Exception on remote invoke: %s\n", ex);
                 return "Execution failure: " + ex.getMessage();
+            }
         } finally {
             synchronized (STOP_LOCK) {
                 userCodeRunning = false;
@@ -310,4 +317,112 @@ class ExecutionControl {
             }
         }
     }
+
+    private final class DemultiplexInput extends Thread {
+
+        private final DataInputStream delegate;
+        private final PipeInputStream command;
+        private final PrintStream out;
+        private final PrintStream err;
+
+        public DemultiplexInput(InputStream input,
+                                         PipeInputStream command,
+                                         PrintStream out,
+                                         PrintStream err) {
+            super("output reader");
+            this.delegate = new DataInputStream(input);
+            this.command = command;
+            this.out = out;
+            this.err = err;
+        }
+
+        public void run() {
+            try {
+                while (true) {
+                    int nameLen = delegate.read();
+                    if (nameLen == (-1))
+                        break;
+                    byte[] name = new byte[nameLen];
+                    DemultiplexInput.this.delegate.readFully(name);
+                    int dataLen = delegate.read();
+                    byte[] data = new byte[dataLen];
+                    DemultiplexInput.this.delegate.readFully(data);
+                    switch (new String(name, "UTF-8")) {
+                        case "err":
+                            err.write(data);
+                            break;
+                        case "out":
+                            out.write(data);
+                            break;
+                        case "command":
+                            for (byte b : data) {
+                                command.write(Byte.toUnsignedInt(b));
+                            }
+                            break;
+                    }
+                }
+            } catch (IOException ex) {
+                proc.debug(ex, "Failed reading output");
+            } finally {
+                command.close();
+            }
+        }
+
+    }
+
+    public static final class PipeInputStream extends InputStream {
+        public static final int INITIAL_SIZE = 128;
+
+        private int[] buffer = new int[INITIAL_SIZE];
+        private int start;
+        private int end;
+        private boolean closed;
+
+        @Override
+        public synchronized int read() {
+            while (start == end) {
+                if (closed) {
+                    return -1;
+                }
+                try {
+                    wait();
+                } catch (InterruptedException ex) {
+                    //ignore
+                }
+            }
+            try {
+                return buffer[start];
+            } finally {
+                start = (start + 1) % buffer.length;
+            }
+        }
+
+        public synchronized void write(int b) {
+            if (closed)
+                throw new IllegalStateException("Already closed.");
+            int newEnd = (end + 1) % buffer.length;
+            if (newEnd == start) {
+                //overflow:
+                int[] newBuffer = new int[buffer.length * 2];
+                int rightPart = (end > start ? end : buffer.length) - start;
+                int leftPart = end > start ? 0 : start - 1;
+                System.arraycopy(buffer, start, newBuffer, 0, rightPart);
+                System.arraycopy(buffer, 0, newBuffer, rightPart, leftPart);
+                buffer = newBuffer;
+                start = 0;
+                end = rightPart + leftPart;
+                newEnd = end + 1;
+            }
+            buffer[end] = b;
+            end = newEnd;
+            notifyAll();
+        }
+
+        @Override
+        public synchronized void close() {
+            closed = true;
+            notifyAll();
+        }
+
+    }
 }

langtools/src/jdk.jshell/share/classes/jdk/jshell/JDIConnection.java

@@ -133,7 +133,7 @@ class JDIConnection {
         return vm;
     }
 
-    boolean setConnectorArg(String name, String value) {
+    synchronized boolean setConnectorArg(String name, String value) {
         /*
          * Too late if the connection already made
          */
@@ -165,7 +165,7 @@ class JDIConnection {
         }
     }
 
-    boolean isOpen() {
+    synchronized boolean isOpen() {
         return (vm != null);
     }
 
@@ -173,13 +173,17 @@ class JDIConnection {
         return (connector instanceof LaunchingConnector);
     }
 
-    public void disposeVM() {
+    synchronized boolean isRunning() {
+        return process != null && process.isAlive();
+    }
+
+    public synchronized void disposeVM() {
         try {
             if (vm != null) {
                 vm.dispose(); // This could NPE, so it is caught below
                 vm = null;
             }
-        } catch (VMDisconnectedException | NullPointerException ex) {
+        } catch (VMDisconnectedException ex) {
             // Ignore if already closed
         } finally {
             if (process != null) {

langtools/test/jdk/jshell/ReplToolTesting.java

@@ -152,13 +152,13 @@ public class ReplToolTesting {
     }
 
     public String getCommandOutput() {
-        String s = cmdout.toString();
+        String s = normalizeLineEndings(cmdout.toString());
         cmdout.reset();
         return s;
     }
 
     public String getCommandErrorOutput() {
-        String s = cmderr.toString();
+        String s = normalizeLineEndings(cmderr.toString());
         cmderr.reset();
         return s;
     }
@@ -168,13 +168,13 @@ public class ReplToolTesting {
     }
 
     public String getUserOutput() {
-        String s = userout.toString();
+        String s = normalizeLineEndings(userout.toString());
         userout.reset();
         return s;
     }
 
     public String getUserErrorOutput() {
-        String s = usererr.toString();
+        String s = normalizeLineEndings(usererr.toString());
         usererr.reset();
         return s;
     }
@@ -461,6 +461,10 @@ public class ReplToolTesting {
         }
     }
 
+    private String normalizeLineEndings(String text) {
+        return text.replace(System.getProperty("line.separator"), "\n");
+    }
+
     public static abstract class MemberInfo {
         public final String source;
         public final String type;

langtools/test/jdk/jshell/ToolBasicTest.java

@@ -23,14 +23,16 @@
 
 /*
  * @test
- * @bug 8143037 8142447 8144095 8140265
+ * @bug 8143037 8142447 8144095 8140265 8144906
+ * @requires os.family != "solaris"
  * @summary Tests for Basic tests for REPL tool
  * @library /tools/lib
  * @ignore 8139873
  * @build KullaTesting TestingInputStream ToolBox Compiler
- * @run testng ToolBasicTest
+ * @run testng/timeout=600 ToolBasicTest
  */
 
+import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
@@ -460,8 +462,7 @@ public class ToolBasicTest extends ReplToolTesting {
         Path unknown = compiler.getPath("UNKNOWN.jar");
         test(true, new String[]{unknown.toString()},
                 "|  File '" + unknown
-                + "' is not found: " + unknown
+                + "' is not found: " + unresolvableMessage(unknown) + "\n");
-                + " (No such file or directory)\n");
     }
 
     public void testReset() {
@@ -514,8 +515,7 @@ public class ToolBasicTest extends ReplToolTesting {
             test(
                     (a) -> assertCommand(a, s + " " + unknown,
                             "|  File '" + unknown
-                                    + "' is not found: " + unknown
+                                    + "' is not found: " + unresolvableMessage(unknown) + "\n")
-                                    + " (No such file or directory)\n")
             );
         }
     }
@@ -874,6 +874,15 @@ public class ToolBasicTest extends ReplToolTesting {
         );
     }
 
+    private String unresolvableMessage(Path p) {
+        try {
+            new FileInputStream(p.toFile());
+            throw new AssertionError("Expected exception did not occur.");
+        } catch (IOException ex) {
+            return ex.getMessage();
+        }
+    }
+
     public void testCommandPrefix() {
         test(a -> assertCommandCheckOutput(a, "/s",
                       assertStartsWith("|  Command: /s is ambiguous: /seteditor, /save, /setstart")),