[maven-surefire] branch maven2surefire-jvm-communication updated (a69ad60 -> f92816d)

classic Classic list List threaded Threaded
4 messages Options
Reply | Threaded
Open this post in threaded view
|

[maven-surefire] branch maven2surefire-jvm-communication updated (a69ad60 -> f92816d)

Tibor Digana
This is an automated email from the ASF dual-hosted git repository.

tibordigana pushed a change to branch maven2surefire-jvm-communication
in repository https://gitbox.apache.org/repos/asf/maven-surefire.git.


    from a69ad60  resolved the previous problems
     new 33bbb50  improved DaemonThreadFactory
     new 795bda5  small two changes
     new f92816d  TCP server

The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.


Summary of changes:
 .../extensions/NetworkingProcessExecutor.java      | 180 ++++++++++++++++++---
 .../surefire/extensions/PipeProcessExecutor.java   |   2 +-
 .../surefire/extensions/SurefireForkChannel.java   |  33 +++-
 .../util/internal/DaemonThreadFactory.java         |  40 +----
 .../maven/surefire/extensions/CommandReader.java   |   4 +-
 5 files changed, 194 insertions(+), 65 deletions(-)

Reply | Threaded
Open this post in threaded view
|

[maven-surefire] 01/03: improved DaemonThreadFactory

Tibor Digana
This is an automated email from the ASF dual-hosted git repository.

tibordigana pushed a commit to branch maven2surefire-jvm-communication
in repository https://gitbox.apache.org/repos/asf/maven-surefire.git

commit 33bbb5014a04b562ac0eaa1ead0439a09137af3f
Author: tibordigana <[hidden email]>
AuthorDate: Fri Nov 8 23:50:16 2019 +0100

    improved DaemonThreadFactory
---
 .../util/internal/DaemonThreadFactory.java         | 40 ++++------------------
 1 file changed, 6 insertions(+), 34 deletions(-)

diff --git a/surefire-api/src/main/java/org/apache/maven/surefire/util/internal/DaemonThreadFactory.java b/surefire-api/src/main/java/org/apache/maven/surefire/util/internal/DaemonThreadFactory.java
index 3610a4b..06ddc53 100644
--- a/surefire-api/src/main/java/org/apache/maven/surefire/util/internal/DaemonThreadFactory.java
+++ b/surefire-api/src/main/java/org/apache/maven/surefire/util/internal/DaemonThreadFactory.java
@@ -19,8 +19,8 @@ package org.apache.maven.surefire.util.internal;
  * under the License.
  */
 
+import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Creates new daemon Thread.
@@ -28,29 +28,16 @@ import java.util.concurrent.atomic.AtomicInteger;
 public final class DaemonThreadFactory
     implements ThreadFactory
 {
-    private static final AtomicInteger POOL_NUMBER = new AtomicInteger( 1 );
-
-    private final AtomicInteger threadNumber = new AtomicInteger( 1 );
-
-    private final ThreadGroup group;
-
-    private final String namePrefix;
+    private static final ThreadFactory DEFAULT_THREAD_FACTORY = Executors.defaultThreadFactory();
 
     private DaemonThreadFactory()
     {
-        SecurityManager s = System.getSecurityManager();
-        group = s != null ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
-        namePrefix = "pool-" + POOL_NUMBER.getAndIncrement() + "-thread-";
     }
 
     @Override
     public Thread newThread( Runnable r )
     {
-        Thread t = new Thread( group, r, namePrefix + threadNumber.getAndIncrement() );
-        if ( t.getPriority() != Thread.NORM_PRIORITY )
-        {
-            t.setPriority( Thread.NORM_PRIORITY );
-        }
+        Thread t = DEFAULT_THREAD_FACTORY.newThread( r );
         t.setDaemon( true );
         return t;
     }
@@ -71,34 +58,19 @@ public final class DaemonThreadFactory
 
     public static Thread newDaemonThread( Runnable r )
     {
-        SecurityManager s = System.getSecurityManager();
-        ThreadGroup group = s == null ? Thread.currentThread().getThreadGroup() : s.getThreadGroup();
-        Thread t = new Thread( group, r );
-        if ( t.getPriority() != Thread.NORM_PRIORITY )
-        {
-            t.setPriority( Thread.NORM_PRIORITY );
-        }
-        t.setDaemon( true );
-        return t;
+        return new DaemonThreadFactory().newThread( r );
     }
 
     public static Thread newDaemonThread( Runnable r, String name )
     {
-        SecurityManager s = System.getSecurityManager();
-        ThreadGroup group = s == null ? Thread.currentThread().getThreadGroup() : s.getThreadGroup();
-        Thread t = new Thread( group, r, name );
-        if ( t.getPriority() != Thread.NORM_PRIORITY )
-        {
-            t.setPriority( Thread.NORM_PRIORITY );
-        }
-        t.setDaemon( true );
+        Thread t = new DaemonThreadFactory().newThread( r );
+        t.setName( name );
         return t;
     }
 
     private static class NamedThreadFactory
         implements ThreadFactory
     {
-
         private final String name;
 
         private NamedThreadFactory( String name )

Reply | Threaded
Open this post in threaded view
|

[maven-surefire] 02/03: small two changes

Tibor Digana
In reply to this post by Tibor Digana
This is an automated email from the ASF dual-hosted git repository.

tibordigana pushed a commit to branch maven2surefire-jvm-communication
in repository https://gitbox.apache.org/repos/asf/maven-surefire.git

commit 795bda5ce32bcc15b0eb93a5ca734f1f4964517b
Author: tibordigana <[hidden email]>
AuthorDate: Sat Nov 9 03:17:58 2019 +0100

    small two changes
---
 .../apache/maven/plugin/surefire/extensions/PipeProcessExecutor.java  | 2 +-
 .../main/java/org/apache/maven/surefire/extensions/CommandReader.java | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/PipeProcessExecutor.java b/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/PipeProcessExecutor.java
index 562a350..1ef9494 100644
--- a/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/PipeProcessExecutor.java
+++ b/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/PipeProcessExecutor.java
@@ -126,7 +126,7 @@ final class PipeProcessExecutor
             }
 
             @SuppressWarnings( "checkstyle:magicnumber" )
-            int b =  currentBuffer[currentPos++] & 0xff;
+            int b = currentBuffer[currentPos++] & 0xff;
             if ( currentPos == currentBuffer.length )
             {
                 currentBuffer = null;
diff --git a/surefire-extensions-api/src/main/java/org/apache/maven/surefire/extensions/CommandReader.java b/surefire-extensions-api/src/main/java/org/apache/maven/surefire/extensions/CommandReader.java
index eb15d8e..8dd9d96 100644
--- a/surefire-extensions-api/src/main/java/org/apache/maven/surefire/extensions/CommandReader.java
+++ b/surefire-extensions-api/src/main/java/org/apache/maven/surefire/extensions/CommandReader.java
@@ -33,9 +33,9 @@ public interface CommandReader
 {
 
     /**
-     * Waits for the next command and reads complete stream of encoded command.
+     * Waits for the next command and returns it.
      *
-     * @return encoded command, or null if closed
+     * @return the command, or null if closed
      */
     Command readNextCommand() throws IOException;
     void close();

Reply | Threaded
Open this post in threaded view
|

[maven-surefire] 03/03: TCP server

Tibor Digana
In reply to this post by Tibor Digana
This is an automated email from the ASF dual-hosted git repository.

tibordigana pushed a commit to branch maven2surefire-jvm-communication
in repository https://gitbox.apache.org/repos/asf/maven-surefire.git

commit f92816de454102dc79cc81eb0ae1b69571c99f49
Author: tibordigana <[hidden email]>
AuthorDate: Sat Nov 9 03:18:18 2019 +0100

    TCP server
---
 .../extensions/NetworkingProcessExecutor.java      | 180 ++++++++++++++++++---
 .../surefire/extensions/SurefireForkChannel.java   |  33 +++-
 2 files changed, 185 insertions(+), 28 deletions(-)

diff --git a/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/NetworkingProcessExecutor.java b/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/NetworkingProcessExecutor.java
index d640e02..c5da33b 100644
--- a/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/NetworkingProcessExecutor.java
+++ b/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/NetworkingProcessExecutor.java
@@ -21,6 +21,8 @@ package org.apache.maven.plugin.surefire.extensions;
 
 import org.apache.maven.shared.utils.cli.CommandLineUtils;
 import org.apache.maven.shared.utils.cli.Commandline;
+import org.apache.maven.surefire.booter.Command;
+import org.apache.maven.surefire.booter.MasterProcessCommand;
 import org.apache.maven.surefire.extensions.CommandReader;
 import org.apache.maven.surefire.extensions.EventHandler;
 import org.apache.maven.surefire.extensions.ExecutableCommandline;
@@ -28,9 +30,19 @@ import org.apache.maven.surefire.extensions.StdErrStreamLine;
 import org.apache.maven.surefire.extensions.StdOutStreamLine;
 
 import javax.annotation.Nonnull;
-import java.net.ServerSocket;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousServerSocketChannel;
+import java.nio.channels.AsynchronousSocketChannel;
+import java.nio.channels.CompletionHandler;
+import java.util.Scanner;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 
+import static java.nio.ByteBuffer.wrap;
 import static java.nio.charset.StandardCharsets.US_ASCII;
 
 /**
@@ -39,42 +51,168 @@ import static java.nio.charset.StandardCharsets.US_ASCII;
  */
 final class NetworkingProcessExecutor implements ExecutableCommandline
 {
-    private final ServerSocket ss;
+    private final AsynchronousServerSocketChannel server;
+    private final ExecutorService executorService;
 
-    NetworkingProcessExecutor( ServerSocket ss )
+    NetworkingProcessExecutor( AsynchronousServerSocketChannel server, ExecutorService executorService )
     {
-        this.ss = ss;
+        this.server = server;
+        this.executorService = executorService;
     }
 
     @Nonnull
     @Override
     public <T> Callable<Integer> executeCommandLineAsCallable( @Nonnull T cli,
-                                                               @Nonnull CommandReader commands,
-                                                               @Nonnull EventHandler events,
+                                                               @Nonnull final CommandReader commands,
+                                                               @Nonnull final EventHandler events,
                                                                StdOutStreamLine stdOut,
                                                                StdErrStreamLine stdErr,
                                                                @Nonnull Runnable runAfterProcessTermination )
             throws Exception
     {
-        /*
-        Call in Threads:
-
-        Socket s = ss.accept();
-
-        for ( Scanner scanner = new Scanner( s.getInputStream(), "ASCII" ); scanner.hasNextLine(); )
+        server.accept( null, new CompletionHandler<AsynchronousSocketChannel, Object>()
         {
-            events.handleEvent( scanner.nextLine() );
-        }
+            @Override
+            public void completed( final AsynchronousSocketChannel client, Object attachment )
+            {
+                executorService.submit( new Runnable()
+                {
+                    @Override
+                    public void run()
+                    {
+                        InputStream is = toInputStream( client );
+                        try
+                        {
+                            for ( Scanner scanner = new Scanner( is, "ASCII" ); scanner.hasNextLine(); )
+                            {
+                                if ( scanner.ioException() != null )
+                                {
+                                    break;
+                                }
+                                events.handleEvent( scanner.nextLine() );
+                            }
+                        }
+                        catch ( IllegalStateException e )
+                        {
+                            // scanner and InputStream is closed
+                            try
+                            {
+                                client.close();
+                            }
+                            catch ( IOException ex )
+                            {
+                                // couldn't close the client channel
+                            }
+                        }
+                    }
+                } );
 
-        Command cmd = commands.readNextCommand();
-        if ( cmd != null )
-        {
-            MasterProcessCommand cmdType = cmd.getCommandType();
-            s.getOutputStream()
-                    .write( cmdType.hasDataType() ? cmdType.encode( cmd.getData() ) : cmdType.encode() );
-        }*/
+                executorService.submit( new Runnable()
+                {
+                    @Override
+                    public void run()
+                    {
+                        try
+                        {
+                            for ( Command cmd; ( cmd = commands.readNextCommand() ) != null;  )
+                            {
+                                MasterProcessCommand cmdType = cmd.getCommandType();
+                                byte[] b = cmdType.hasDataType() ? cmdType.encode( cmd.getData() ) : cmdType.encode();
+                                ByteBuffer bb = wrap( b );
+                                int writtenBytesTotal = 0;
+                                do
+                                {
+                                    Future<Integer> writtenBytes = client.write( bb );
+                                    int writtenCount = writtenBytes.get();
+                                    writtenBytesTotal += writtenCount;
+                                }
+                                while ( writtenBytesTotal < bb.limit() );
+                            }
+                        }
+                        catch ( Exception e )
+                        {
+                            // finished stream or error
+                            try
+                            {
+                                client.close();
+                            }
+                            catch ( IOException ex )
+                            {
+                                // couldn't close the client channel
+                            }
+                        }
+                    }
+                } );
+            }
+
+            @Override
+            public void failed( Throwable exc, Object attachment )
+            {
+                // write to dump file
+                // close the server
+            }
+        } );
 
         return CommandLineUtils.executeCommandLineAsCallable( (Commandline) cli, null,
                 new StdOutAdapter( stdOut ), new StdErrAdapter( stdErr ), 0, runAfterProcessTermination, US_ASCII );
     }
+
+    private static InputStream toInputStream( final AsynchronousSocketChannel client )
+    {
+        return new InputStream()
+        {
+            private final ByteBuffer bb = ByteBuffer.allocate( 64 * 1024 );
+            private int read;
+            private boolean closed;
+
+            @Override
+            public int read() throws IOException
+            {
+                if ( closed )
+                {
+                    return -1;
+                }
+
+                try
+                {
+                    if ( read == 0 )
+                    {
+                        bb.clear();
+                        read = client.read( bb ).get();
+                        if ( read == 0 )
+                        {
+                            return -1;
+                        }
+                    }
+                    read--;
+                    return bb.get();
+                }
+                catch ( InterruptedException e )
+                {
+                    closed = true;
+                    return -1;
+                }
+                catch ( ExecutionException e )
+                {
+                    closed = true;
+                    Throwable cause = e.getCause();
+                    if ( cause instanceof IOException )
+                    {
+                        throw (IOException) cause;
+                    }
+                    else
+                    {
+                        return -1;
+                    }
+                }
+            }
+
+            @Override
+            public void close() throws IOException
+            {
+                closed = true;
+                super.close();
+            }
+        };
+    }
 }
diff --git a/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/SurefireForkChannel.java b/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/SurefireForkChannel.java
index da35ac4..fe5e2a2 100644
--- a/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/SurefireForkChannel.java
+++ b/maven-surefire-common/src/main/java/org/apache/maven/plugin/surefire/extensions/SurefireForkChannel.java
@@ -24,36 +24,55 @@ import org.apache.maven.surefire.extensions.ForkChannel;
 
 import javax.annotation.Nonnull;
 import java.io.IOException;
-import java.net.ServerSocket;
+import java.net.InetSocketAddress;
+import java.nio.channels.AsynchronousServerSocketChannel;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static java.nio.channels.AsynchronousChannelGroup.withThreadPool;
+import static java.nio.channels.AsynchronousServerSocketChannel.open;
+import static org.apache.maven.surefire.util.internal.DaemonThreadFactory.newDaemonThreadFactory;
 
 /**
  *
  */
 final class SurefireForkChannel implements ForkChannel
 {
-    private final ServerSocket ss;
+    private final ExecutorService executorService;
+    private final AsynchronousServerSocketChannel server;
+    private final int serverPort;
 
     SurefireForkChannel() throws IOException
     {
-        ss = new ServerSocket( 0 );
+        executorService = Executors.newCachedThreadPool( newDaemonThreadFactory() );
+        server = open( withThreadPool( executorService ) )
+                .bind( new InetSocketAddress( 0 ) );
+        serverPort = ( (InetSocketAddress) server.getLocalAddress() ).getPort();
     }
 
     @Override
     public String getForkNodeConnectionString()
     {
-        return "tcp://127.0.0.1:" + ss.getLocalPort();
+        return "tcp://127.0.0.1:" + serverPort;
     }
 
     @Nonnull
     @Override
-    public ExecutableCommandline createExecutableCommandline() throws IOException
+    public ExecutableCommandline createExecutableCommandline()
     {
-        return new NetworkingProcessExecutor( ss );
+        return new NetworkingProcessExecutor( server, executorService );
     }
 
     @Override
     public void close() throws IOException
     {
-        ss.close();
+        try
+        {
+            server.close();
+        }
+        finally
+        {
+            executorService.shutdownNow();
+        }
     }
 }