java calling rsync calling ssh: io threads blocked

This is a multi-part message in MIME format.
--------------060607010405020100020305
Content-Type: text/plain; charset=ISO-8859-1; format=flowed
Content-Transfer-Encoding: 7bit

Subject:
   java calling rsync calling ssh: io threads blocked.

I'm trying to run a java program, that starts rsync, that uses ssh to
send data (--rsh option).

When I run the java program, all my I/O threads are waiting; nothing
happens, and rsync timeouts after 300 seconds.

In Eclipse in Debug mode, I see that all my threads are "Running" .
After I suspend each thread, I can see in what line of my code the
thread is executing:

   Thread stdoutRunnable:
     line 187: int num = procStdout.read(buf);

   Thread stderrRunnable:
     line 166: int num = procStderr.read(buf);

   Thread stdinRunnable:
     line 114: stdinLock.wait();

These are all statements that block.

When I run the same rsync command from a cygwin command shell, the rsync
runs okay.

The rsync.exe and ssh.exe are part of a Cygwin environment installed on
my machine.
(Respectively
    rsync version 3.0.7 protocol version 30
  and
    OpenSSH_5.6p1, OpenSSL 0.9.8o 01 Jun 2010
)

The rsync command is:

C:/cygwin/bin/rsync \
--modify-window=2 \
--verbose \
--recursive \
--no-perms \
--no-group \
--times \
--fuzzy \
--compress \
--log-file="/cygdrive/c/Users/p.vanbemmelen/Documents/rsync/log/+logFile+" \
--rsync-path=rsync \
--stats \
--bwlimit="50" \
--progress \
--rsh="ssh -l p.vanbemmelen -i
/cygdrive/c/Users/p.vanbemmelen/.ssh/sshkey.ssh" \
--backup \
--human-readable \
--partial \
--timeout=300 \
--partial-dir=".rsync/partial/" \
--filter="exclude_p.vanbemmelen/downloads/" \
--chmod=u+rwx \
/cygdrive/c/cygwin/home/p.vanbemmelen \
p.vanbemmelen@backup2.redheadtech.nl:rbm/manualbackup \

(This command is hardcoded in the Java code; I can change a boolean in
that java code, and recompile, to print the command to stdout; so that
the same command can be issued from the cygwin command shell .)

I use ProcessBuilder from the Java library, and my own ProcessManager
class to listing to stdout, stderr of the rsync process.

I've attached the code.

The main() is in RsyncRunnableRemote.Test .

Note: The goal is to have a java program that can be installed on end
users' computers to give them an easy interface to rsync.
I'm perfectly able to run rsync on my own pc from the cygwin command
shell, but I really need to be able to run rsync from java.

Any help is appreciated,

Paul van Bemmelen.

--------------060607010405020100020305
Content-Type: text/plain;
 name="OutputStreamLogger.java"
Content-Transfer-Encoding: base64
Content-Disposition: attachment;
 filename="OutputStreamLogger.java"

LyoqDQogKiANCiAqLw0KcGFja2FnZSBubC54czRhbGwucHZiZW1tZWwucnN5bmM7DQoNCmlt
cG9ydCBqYXZhLmlvLio7DQoNCmNsYXNzIE91dHB1dFN0cmVhbUxvZ2dlciBpbXBsZW1lbnRz
DQogICAgIFByb2Nlc3NNYW5hZ2VyLk91dHB1dFN0cmVhbUxpc3RlbmVyIHsNCiAgU3RyaW5n
IGxvZ0ZpbGU7DQogIE91dHB1dFN0cmVhbUxvZ2dlcihTdHJpbmcgbG9nRmlsZSkgew0KICAg
IHRoaXMubG9nRmlsZSA9IGxvZ0ZpbGU7DQogIH0NCiAgDQogIEBPdmVycmlkZQ0KICBwdWJs
aWMgdm9pZCBoYW5kbGUoYnl0ZVtdIGJ5dGVzLCBpbnQgbnVtKSB7DQogICAgU3RyaW5nIHMg
PSBuZXcgU3RyaW5nKGJ5dGVzLCAwLCBudW0pOw0KICAgIGhhbmRsZShzKTsNCiAgfSAgDQog
IHB1YmxpYyB2b2lkIGhhbmRsZShTdHJpbmcgcykgew0KICAgIFByaW50V3JpdGVyIG91dCA9
IG51bGw7DQogICAgdHJ5IHsNCiAgICAgIG91dCA9IG5ldyBQcmludFdyaXRlcihuZXcgRmls
ZVdyaXRlcihsb2dGaWxlLCB0cnVlKSwgdHJ1ZSk7DQogICAgICBvdXQud3JpdGUocyk7DQog
ICAgfSBjYXRjaCAoSU9FeGNlcHRpb24gZSkgew0KICAgICAgdGhyb3cgbmV3IElsbGVnYWxT
dGF0ZUV4Y2VwdGlvbihlLnRvU3RyaW5nKCkpOw0KICAgIH0NCiAgICBmaW5hbGx5IHsNCiAg
ICAgIHRyeSB7IG91dC5jbG9zZSgpOyB9IGNhdGNoKEV4Y2VwdGlvbiBlKSB7fQ0KICAgIH0N
CiAgfQ0KfQ==
--------------060607010405020100020305
Content-Type: text/plain;
 name="ProcessManager.java"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
 filename="ProcessManager.java"

package nl.xs4all.pvbemmel.rsync;

import java.io.*;
import java.util.*;

/**
 * Manages a process by creating Threads for i/o to stdin,stdout,stderr.
 * A client can send input to the process by setting the stdinReader.
 * A client can obtain stdout/stderr output from the process by adding an
 * OutputStreamListener. The listener's handle() will be called on the
 * corresponding StdoutRunnable c.q. StderrRunnable thread created by
 * ProcessManager.
 * <p>
 * If a listener's handle() takes a long time to return, then StdoutRunnable/
 * StderrRunnable will be unable to promptly read the stdout/stderr streams of
 * the subprocess. According to {@link java.lang.Process} this may cause
 * the subprocess to block, and even deadlock.<br />
 * <p>
 * If process ends, then the threads for stdout and stderr immediately end;
 * the thread for stdin will end when the next line from stdinReader is send
 * to the process.
 *
 * @author p.vanbemmelen
 */
public class ProcessManager {
  /* java.lang.Process:
   * Because some native platforms only provide limited buffer size for
   * standard input and output streams, failure to promptly write the input
   * stream or read the output stream of the subprocess may cause the
   * subprocess to block, and even deadlock.

   * public abstract InputStream getInputStream()
   * Gets the input stream of the subprocess. The stream obtains data piped
   * from the standard output stream of the process represented by this
   * Process object.
   * Implementation note: It is a good idea for the input stream to be
   * buffered.
   * Returns:
   * the input stream connected to the normal output of the subprocess.
   *
   * public abstract OutputStream getOutputStream()
   * Gets the output stream of the subprocess. Output to the stream is piped
   * into the standard input stream of the process represented by this
   * Process object.
   * Implementation note: It is a good idea for the output stream to be
   * buffered.
   * Returns:
   * the output stream connected to the normal input of the subprocess.
   */
  final private Process proc;
  final private BufferedWriter procStdin;
  final private BufferedInputStream procStdout;
  final private BufferedInputStream procStderr;
  final private List<OutputStreamListener> stdoutListeners;
  final private List<OutputStreamListener> stderrListeners;
  final private Object stdinLock;
  private BufferedReader stdinReader;
  private String newline;

  public interface OutputStreamListener {
    public void handle(byte[] bytes, int num);
  }
  public ProcessManager(Process proc) {
    this.proc = proc;
    newline = "\n";
    stdinLock = new Object();
    stdoutListeners = Collections.synchronizedList(
        new ArrayList<OutputStreamListener>());
    stderrListeners = Collections.synchronizedList(
        new ArrayList<OutputStreamListener>());
    // Confusing names:
    // To get access to the process's stdin, you need to call getOutputStream.
    // Below, I'll use procStdin for the writer connected to the
    // getOutputStream return value.
    procStdin = new BufferedWriter(new OutputStreamWriter(
        proc.getOutputStream()));
    procStdout = new BufferedInputStream(proc.getInputStream());
    procStderr = new BufferedInputStream(proc.getErrorStream());
    //
    Runnable stdinRunnable = new StdinRunnable();
    Runnable stdoutRunnable = new StdoutRunnable();
    Runnable stderrRunnable = new StderrRunnable();
    new Thread(stdoutRunnable, "stdoutRunnable").start();
    new Thread(stderrRunnable, "stderrRunnable").start();
    Thread stdinThread = new Thread(stdinRunnable, "stdinRunnable");
    stdinThread.setDaemon(true);
    stdinThread.start();
  }
  public void setStdin(BufferedReader reader) {
    synchronized(stdinLock) {
      stdinReader = reader;
      stdinLock.notify();
    }
  }
  /**
   * Each line send to process will end in <code>newline</code>.
   * Client may specify this if the process requires it.
   * @param newline default "\n".
   */
  public void setProcessInNewLine(String newline) {
    this.newline = newline;
  }
  public void addStdoutListener(OutputStreamListener listener) {
    stdoutListeners.add(listener);
  }
  public void addStderrListener(OutputStreamListener listener) {
    stderrListeners.add(listener);
  }
  private final class StdinRunnable implements Runnable {
    public void run() {
      while(stdinReader==null) {
        synchronized(stdinLock) {
          try {
            stdinLock.wait();
          }
          catch(InterruptedException ie) {
          }
        }
      };
      while(true) {
        try {
/*
 * String java.io.BufferedReader.readLine() throws IOException
 *
 * Reads a line of text. A line is considered to be terminated by any
 * one of a line feed ('\n'), a carriage return ('\r'), or a carriage
 * return followed immediately by a linefeed.
 *
 * Returns:
 * A String containing the contents of the line, not including any
 * line-termination characters, or null if the end of the stream has
 * been reached
 * Throws:
 * IOException - If an I/O error occurs
 */
          String line = stdinReader.readLine();
// Debug.println(line);
          procStdin.write(line + newline);
          procStdin.flush();
        }
        catch(IOException ioe) {
          Integer exitValue = null;
          try {
            exitValue = proc.exitValue();
          }
          catch(IllegalThreadStateException itse) {
          }
          // exitValue!=null <==> proc has exitted.
          
          if(exitValue==null) {
            throw new IllegalStateException(ioe.getMessage());
          }
          else {
            break; // : will exit infinite loop, and end thread.
          }
        }
      }
// Debug.println(Thread.currentThread().getName() + " about to return");
    }
  }
  private final class StderrRunnable implements Runnable {
    public void run() {
      while(true) {
        byte[] buf = new byte[1024];
        try {
          int num = procStderr.read(buf);
          if(num==-1) {
            break; // : will exit infinite loop, and end thread.
          }
          for(OutputStreamListener listener: stderrListeners) {
            listener.handle(buf, num);
          }
        }
        catch(IOException ioe) {
          throw new IllegalStateException(ioe.getMessage());
        }
      }
// Debug.println(Thread.currentThread().getName() + " about to return");
    }
  }

  private final class StdoutRunnable implements Runnable {
    public void run() {
      while(true) {
        byte[] buf = new byte[1024];
        try {
          int num = procStdout.read(buf);
          if(num==-1) {
            break; // : will exit infinite loop, and end thread.
          }
          for(OutputStreamListener listener: stdoutListeners) {
            listener.handle(buf, num);
          }
        }
        catch(IOException ioe) {
          throw new IllegalStateException(ioe.getMessage());
        }
      }
// Debug.println(Thread.currentThread().getName() + " about to return");
    }
  }

  public static class Test {
    public static void main(String[] args) throws IOException {
      boolean simpleShell = false;
      String[] cmdarray = null;
      File dir = null;
      String path = null;
      if(simpleShell) {
        cmdarray = new String[] { "C:/cygwin/bin/bash.exe" };
        dir = new File("C:/cygwin/bin");
        path = "C:\\cygwin\\bin";
      }
      else {
        cmdarray = new String[] {
          "C:/Program Files/Redhead Online Backup/Bin/ssh",
          "-l",
          "p.vanbemmelen",
          "-i",
          "/cygdrive/c/Users/p.vanbemmelen/.ssh/sshkey.ssh",
          "p.vanbemmelen@backup2.redheadtech.nl",
          "rsync --version"
        };
        dir = new File("C:/Program Files/Redhead Online Backup/Bin");
        path = "C:\\Program Files\\Redhead Online Backup\\Bin";
      }
      //
      ProcessBuilder procB = new ProcessBuilder(cmdarray);
      procB.directory(dir);
      Map<String, String> procEnv = procB.environment();
      procEnv.put("Path", path);
      Process proc = procB.start();
      ProcessManager procMgr = new ProcessManager(proc);
      
      procMgr.addStdoutListener(new OutputStreamListener() {
        public void handle(byte[] bytes, int num) {
          System.out.write(bytes, 0, num);
        }
      });
      procMgr.addStderrListener(new OutputStreamListener() {
        public void handle(byte[] bytes, int num) {
          System.out.write(bytes, 0, num);
        }
      });
      procMgr.setStdin(new BufferedReader(new InputStreamReader(System.in)));
    }
  }
}

--------------060607010405020100020305
Content-Type: text/plain;
 name="RsyncRunnableRemote.java"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
 filename="RsyncRunnableRemote.java"

package nl.xs4all.pvbemmel.rsync;

import java.io.*;
import java.util.*;
import static java.lang.System.out;

public class RsyncRunnableRemote {
  private final static String nl = "\n";

  public void run() {
    OutputStreamLogger logger = new OutputStreamLogger("rsyncRemote.log");
    try {
      
      String[] cmdarray = new String[] {
          "C:/cygwin/bin/rsync",
          "--modify-window=2" ,
          "--verbose" ,
          "--recursive" ,
          "--no-perms" ,
          "--no-group" ,
          "--times" ,
          "--fuzzy" ,
          "--compress" ,
          "--log-file=\"/cygdrive/c/Users/p.vanbemmelen/Documents/rsync/log/+logFile+\"" ,
          "--rsync-path=rsync" ,
          "--stats" ,
          "--bwlimit=\"50\"" ,
          "--progress" ,
          "--rsh=\"ssh -l p.vanbemmelen -i /cygdrive/c/Users/p.vanbemmelen/.ssh/sshkey.ssh\"" ,
          "--backup" ,
          "--human-readable" ,
          "--partial" ,
          "--timeout=300" ,
          "--partial-dir=\".rsync/partial/\"" ,
          "--filter=\"exclude_p.vanbemmelen/downloads/\"" ,
          "--chmod=u+rwx",
          "/cygdrive/c/cygwin/home/p.vanbemmelen" ,
          "p.vanbemmelen@backup2.redheadtech.nl:rbm/manualbackup"
      };
      boolean printOnly = false;
      if(printOnly) {
        for(String s : cmdarray) {
          out.println(s + " \\");
        }
        return;
      }
      File dir = new File("C:/cygwin/bin");

      ProcessBuilder procB = new ProcessBuilder(cmdarray);
      procB.directory(dir);
      Map<String,String> procEnv = procB.environment();
      String path = procEnv.get("Path");
      procEnv.put("Path", "C:/cygwin/bin" + ";" + path);

      Process proc = procB.start();

      ProcessManager procMgr = new ProcessManager(proc);
      procMgr.addStdoutListener(logger);
      procMgr.addStderrListener(logger);
    }
    catch(IOException ioe) {
      logger.handle(ioe.toString() + nl );
      throw new RuntimeException(ioe.toString());
    }
  }
  public static class Test {
    public static void main(String[] args) {
      new RsyncRunnableRemote().run();
    }
  }
}

--------------060607010405020100020305--