Wait for EOF on stderr before finishing SSH channel
JSch will allow us to close the connection and then just drop any late messages coming over the stderr stream for the command. This makes it easy to lose final output on a command, like from Gerrit Code Review's post receive hook. Instead spawn a background thread to copy data from JSch's pipe into our own buffer, and wait for that thread to receive EOF on the pipe before we declare the connection closed. This way we don't have a race condition between the stderr data arriving and JSch just tearing down the channel. Change-Id: Ica1ba40ed2b4b6efb7d5e4ea240efc0a56fb71f6 Signed-off-by: Shawn O. Pearce <spearce@spearce.org>
This commit is contained in:
parent
673b3984bd
commit
243b0d64a6
|
@ -1,4 +1,5 @@
|
||||||
/*
|
/*
|
||||||
|
* Copyright (C) 2008-2010, Google Inc.
|
||||||
* Copyright (C) 2008, Marek Zawirski <marek.zawirski@gmail.com>
|
* Copyright (C) 2008, Marek Zawirski <marek.zawirski@gmail.com>
|
||||||
* Copyright (C) 2008, Robin Rosenberg <robin.rosenberg@dewire.com>
|
* Copyright (C) 2008, Robin Rosenberg <robin.rosenberg@dewire.com>
|
||||||
* Copyright (C) 2008, Shawn O. Pearce <spearce@spearce.org>
|
* Copyright (C) 2008, Shawn O. Pearce <spearce@spearce.org>
|
||||||
|
@ -46,6 +47,7 @@
|
||||||
package org.eclipse.jgit.transport;
|
package org.eclipse.jgit.transport;
|
||||||
|
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
import java.io.InputStream;
|
||||||
import java.io.OutputStream;
|
import java.io.OutputStream;
|
||||||
import java.io.PipedInputStream;
|
import java.io.PipedInputStream;
|
||||||
import java.io.PipedOutputStream;
|
import java.io.PipedOutputStream;
|
||||||
|
@ -142,15 +144,13 @@ private String commandFor(final String exe) {
|
||||||
return cmd.toString();
|
return cmd.toString();
|
||||||
}
|
}
|
||||||
|
|
||||||
ChannelExec exec(final String exe, final OutputStream err)
|
ChannelExec exec(final String exe) throws TransportException {
|
||||||
throws TransportException {
|
|
||||||
initSession();
|
initSession();
|
||||||
|
|
||||||
final int tms = getTimeout() > 0 ? getTimeout() * 1000 : 0;
|
final int tms = getTimeout() > 0 ? getTimeout() * 1000 : 0;
|
||||||
try {
|
try {
|
||||||
final ChannelExec channel = (ChannelExec) sock.openChannel("exec");
|
final ChannelExec channel = (ChannelExec) sock.openChannel("exec");
|
||||||
channel.setCommand(commandFor(exe));
|
channel.setCommand(commandFor(exe));
|
||||||
channel.setErrStream(err);
|
|
||||||
channel.connect(tms);
|
channel.connect(tms);
|
||||||
return channel;
|
return channel;
|
||||||
} catch (JSchException je) {
|
} catch (JSchException je) {
|
||||||
|
@ -224,6 +224,8 @@ public void close() throws IOException {
|
||||||
class SshFetchConnection extends BasePackFetchConnection {
|
class SshFetchConnection extends BasePackFetchConnection {
|
||||||
private ChannelExec channel;
|
private ChannelExec channel;
|
||||||
|
|
||||||
|
private Thread errorThread;
|
||||||
|
|
||||||
private int exitStatus;
|
private int exitStatus;
|
||||||
|
|
||||||
SshFetchConnection() throws TransportException {
|
SshFetchConnection() throws TransportException {
|
||||||
|
@ -231,12 +233,16 @@ class SshFetchConnection extends BasePackFetchConnection {
|
||||||
try {
|
try {
|
||||||
final MessageWriter msg = new MessageWriter();
|
final MessageWriter msg = new MessageWriter();
|
||||||
setMessageWriter(msg);
|
setMessageWriter(msg);
|
||||||
channel = exec(getOptionUploadPack(), msg.getRawStream());
|
|
||||||
|
|
||||||
if (channel.isConnected())
|
channel = exec(getOptionUploadPack());
|
||||||
init(channel.getInputStream(), outputStream(channel));
|
if (!channel.isConnected())
|
||||||
else
|
throw new TransportException(uri, "connection failed");
|
||||||
throw new TransportException(uri, getMessages());
|
|
||||||
|
final InputStream upErr = channel.getErrStream();
|
||||||
|
errorThread = new StreamCopyThread(upErr, msg.getRawStream());
|
||||||
|
errorThread.start();
|
||||||
|
|
||||||
|
init(channel.getInputStream(), outputStream(channel));
|
||||||
|
|
||||||
} catch (TransportException err) {
|
} catch (TransportException err) {
|
||||||
close();
|
close();
|
||||||
|
@ -258,6 +264,16 @@ class SshFetchConnection extends BasePackFetchConnection {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void close() {
|
public void close() {
|
||||||
|
if (errorThread != null) {
|
||||||
|
try {
|
||||||
|
errorThread.join();
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
// Stop waiting and return anyway.
|
||||||
|
} finally {
|
||||||
|
errorThread = null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
super.close();
|
super.close();
|
||||||
|
|
||||||
if (channel != null) {
|
if (channel != null) {
|
||||||
|
@ -275,6 +291,8 @@ public void close() {
|
||||||
class SshPushConnection extends BasePackPushConnection {
|
class SshPushConnection extends BasePackPushConnection {
|
||||||
private ChannelExec channel;
|
private ChannelExec channel;
|
||||||
|
|
||||||
|
private Thread errorThread;
|
||||||
|
|
||||||
private int exitStatus;
|
private int exitStatus;
|
||||||
|
|
||||||
SshPushConnection() throws TransportException {
|
SshPushConnection() throws TransportException {
|
||||||
|
@ -282,12 +300,16 @@ class SshPushConnection extends BasePackPushConnection {
|
||||||
try {
|
try {
|
||||||
final MessageWriter msg = new MessageWriter();
|
final MessageWriter msg = new MessageWriter();
|
||||||
setMessageWriter(msg);
|
setMessageWriter(msg);
|
||||||
channel = exec(getOptionReceivePack(), msg.getRawStream());
|
|
||||||
|
|
||||||
if (channel.isConnected())
|
channel = exec(getOptionReceivePack());
|
||||||
init(channel.getInputStream(), outputStream(channel));
|
if (!channel.isConnected())
|
||||||
else
|
throw new TransportException(uri, "connection failed");
|
||||||
throw new TransportException(uri, getMessages());
|
|
||||||
|
final InputStream rpErr = channel.getErrStream();
|
||||||
|
errorThread = new StreamCopyThread(rpErr, msg.getRawStream());
|
||||||
|
errorThread.start();
|
||||||
|
|
||||||
|
init(channel.getInputStream(), outputStream(channel));
|
||||||
|
|
||||||
} catch (TransportException err) {
|
} catch (TransportException err) {
|
||||||
close();
|
close();
|
||||||
|
@ -309,6 +331,16 @@ class SshPushConnection extends BasePackPushConnection {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void close() {
|
public void close() {
|
||||||
|
if (errorThread != null) {
|
||||||
|
try {
|
||||||
|
errorThread.join();
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
// Stop waiting and return anyway.
|
||||||
|
} finally {
|
||||||
|
errorThread = null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
super.close();
|
super.close();
|
||||||
|
|
||||||
if (channel != null) {
|
if (channel != null) {
|
||||||
|
|
Loading…
Reference in New Issue