Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[MSHARED-1072] fix blocking in StreamFeeder #113

Merged
merged 3 commits into from
Apr 21, 2023
Merged

[MSHARED-1072] fix blocking in StreamFeeder #113

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
756cd5b
[MSHARED-1072] fix blocking in StreamFeeder
slawekjaranowski Jul 27, 2022
587e480
Merge remote-tracking branch 'origin/master' into MSHARED-1072
slawekjaranowski Apr 20, 2023
a0bde68
Interrupt Feeder thread not current thread
slawekjaranowski Apr 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions src/main/java/org/apache/maven/shared/utils/cli/CommandLineTimeOutException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,16 +20,15 @@
*/

/**
* Report a timeout for executing process.
*
* @author Olivier Lamy
*
*
*/
public class CommandLineTimeOutException
extends CommandLineException
{

/**
*
*/
private static final long serialVersionUID = 7322428741683224481L;

/**
Expand All @@ -41,4 +40,11 @@ public CommandLineTimeOutException( String message, Throwable cause )
super( message, cause );
}

/**
* @param message The message of the exception.
*/
public CommandLineTimeOutException( String message )
slawekjaranowski marked this conversation as resolved.
Show resolved Hide resolved
{
super( message );
}
}
87 changes: 11 additions & 76 deletions src/main/java/org/apache/maven/shared/utils/cli/CommandLineUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import java.util.Map;
import java.util.Properties;
import java.util.StringTokenizer;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
Expand All @@ -49,7 +50,7 @@ public static class StringStreamConsumer
implements StreamConsumer
{

private final StringBuffer string = new StringBuffer();
private final StringBuilder string = new StringBuilder();

private static final String LS = System.getProperty( "line.separator", "\n" );

Expand All @@ -72,16 +73,6 @@ public String getOutput()

}

/**
* Number of milliseconds per second.
*/
private static final long MILLIS_PER_SECOND = 1000L;

/**
* Number of nanoseconds per second.
*/
private static final long NANOS_PER_SECOND = 1000000000L;

/**
* @param cl The command line {@link Commandline}
* @param systemOut {@link StreamConsumer}
Expand Down Expand Up @@ -288,31 +279,13 @@ public Integer call()
errorPumper.setName( "StreamPumper-systemErr" );
errorPumper.start();

int returnValue;
if ( timeoutInSeconds <= 0 )
if ( timeoutInSeconds > 0 && !p.waitFor( timeoutInSeconds, TimeUnit.SECONDS ) )
michael-o marked this conversation as resolved.
Show resolved Hide resolved
{
returnValue = p.waitFor();
throw new CommandLineTimeOutException(
String.format( "Process timed out after %d seconds.", timeoutInSeconds ) );
}
else
{
final long now = System.nanoTime();
final long timeout = now + NANOS_PER_SECOND * timeoutInSeconds;
while ( isAlive( p ) && ( System.nanoTime() < timeout ) )
{
// The timeout is specified in seconds. Therefore we must not sleep longer than one second
// but we should sleep as long as possible to reduce the number of iterations performed.
Thread.sleep( MILLIS_PER_SECOND - 1L );
}

if ( isAlive( p ) )
{
throw new InterruptedException( String.format( "Process timed out after %d seconds.",
timeoutInSeconds ) );

}

returnValue = p.exitValue();
}
int returnValue = p.waitFor();

// TODO Find out if waitUntilDone needs to be called using a try-finally construct. The method may throw an
// InterruptedException so that calls to waitUntilDone may be skipped.
Expand Down Expand Up @@ -342,14 +315,9 @@ public Integer call()
outputPumper.waitUntilDone();
errorPumper.waitUntilDone();

if ( inputFeeder != null )
if ( inputFeeder != null && inputFeeder.getException() != null )
{
inputFeeder.close();

if ( inputFeeder.getException() != null )
{
throw new CommandLineException( "Failure processing stdin.", inputFeeder.getException() );
}
throw new CommandLineException( "Failure processing stdin.", inputFeeder.getException() );
}

if ( outputPumper.getException() != null )
Expand All @@ -366,16 +334,12 @@ public Integer call()
}
catch ( InterruptedException ex )
{
Thread.currentThread().interrupt();
throw new CommandLineTimeOutException( "Error while executing external command, process killed.",
ex );

}
finally
{
if ( inputFeeder != null )
{
inputFeeder.disable();
}
if ( outputPumper != null )
{
outputPumper.disable();
Expand All @@ -395,18 +359,7 @@ public Integer call()
finally
{
ShutdownHookUtils.removeShutdownHook( processHook );

try
{
processHook.run();
}
finally
{
if ( inputFeeder != null )
{
inputFeeder.close();
}
}
processHook.run();
}
}
}
Expand Down Expand Up @@ -444,24 +397,6 @@ public static Properties getSystemEnvVars( boolean caseSensitive )
return ensureCaseSensitivity( envs, caseSensitive );
}

private static boolean isAlive( Process p )
{
if ( p == null )
{
return false;
}

try
{
p.exitValue();
return false;
}
catch ( IllegalThreadStateException e )
{
return true;
}
}

/**
* @param toProcess The command line to translate.
* @return The array of translated parts.
Expand All @@ -482,7 +417,7 @@ public static String[] translateCommandline( String toProcess ) throws CommandLi
boolean inEscape = false;
int state = normal;
final StringTokenizer tok = new StringTokenizer( toProcess, "\"\' \\", true );
List<String> tokens = new ArrayList<String>();
List<String> tokens = new ArrayList<>();
StringBuilder current = new StringBuilder();

while ( tok.hasMoreTokens() )
Expand Down
106 changes: 41 additions & 65 deletions src/main/java/org/apache/maven/shared/utils/cli/StreamFeeder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,94 +22,78 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.concurrent.atomic.AtomicReference;
import java.util.Objects;

/**
* Read from an InputStream and write the output to an OutputStream.
*
* @author <a href="mailto:[email protected]">Trygve Laugst&oslash;l</a>
*/
class StreamFeeder
extends AbstractStreamHandler
extends Thread
{

private final AtomicReference<InputStream> input;
private final InputStream input;

private final AtomicReference<OutputStream> output;
private final OutputStream output;

private volatile Throwable exception;
private Throwable exception;
private boolean done;

private final Object lock = new Object();

/**
* Create a new StreamFeeder
*
* @param input Stream to read from
* @param input Stream to read from
* @param output Stream to write to
*/
StreamFeeder( InputStream input, OutputStream output )
{
super();
this.input = new AtomicReference<InputStream>( input );
this.output = new AtomicReference<OutputStream>( output );
this.input = Objects.requireNonNull( input );
this.output = Objects.requireNonNull( output );
this.done = false;
}

@Override
@SuppressWarnings( "checkstyle:innerassignment" )
public void run()
{
try
{
feed();
}
catch ( Throwable e )
{
// Catch everything so the streams will be closed and flagged as done.
if ( this.exception != null )
for ( int data; !isInterrupted() && ( data = input.read() ) != -1; )
{
this.exception = e;
output.write( data );
}
output.flush();
}
catch ( IOException e )
{
exception = e;
}
finally
{
close();
}

synchronized ( this )
{
notifyAll();
}
synchronized ( lock )
{
done = true;
lock.notifyAll();
}
}

public void close()
private void close()
{
setDone();
final InputStream is = input.getAndSet( null );
if ( is != null )
try
{
try
{
is.close();
}
catch ( IOException ex )
{
if ( this.exception != null )
{
this.exception = ex;
}
}
output.close();
}

final OutputStream os = output.getAndSet( null );
if ( os != null )
catch ( IOException e )
{
try
{
os.close();
}
catch ( IOException ex )
if ( exception == null )
{
if ( this.exception != null )
{
this.exception = ex;
}
exception = e;
}
}
}
Expand All @@ -122,30 +106,22 @@ public Throwable getException()
return this.exception;
}

@SuppressWarnings( "checkstyle:innerassignment" )
private void feed()
throws IOException
public void waitUntilDone()
{
InputStream is = input.get();
OutputStream os = output.get();
boolean flush = false;

if ( is != null && os != null )
interrupt();
synchronized ( lock )
{
for ( int data; !isDone() && ( data = is.read() ) != -1; )
while ( !done )
{
if ( !isDisabled() )
try
{
os.write( data );
flush = true;
lock.wait();
}
catch ( InterruptedException e )
{
Thread.currentThread().interrupt();
}
}

if ( flush )
{
os.flush();
}
}
}

}
Loading