Fix #128 - [NS3] Test ns-3 with localhost

[nepi.git] / src / nepi / util / sshfuncs.py

diff --git a/src/nepi/util/sshfuncs.py b/src/nepi/util/sshfuncs.py
index f5c6400..d1952bc 100644 (file)
--- a/src/nepi/util/sshfuncs.py
+++ b/src/nepi/util/sshfuncs.py
@@ -18,6 +18,8 @@
 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
 #         Claudio Freire <claudio-daniel.freire@inria.fr>
 
 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
 #         Claudio Freire <claudio-daniel.freire@inria.fr>
 

+## TODO: This code needs reviewing !!!
+

 import base64
 import errno
 import hashlib
 import base64
 import errno
 import hashlib


@@ -44,7 +46,6 @@ def log(msg, level, out = None, err = None):
 
     logger.log(level, msg)
 
 
     logger.log(level, msg)
 

-

 if hasattr(os, "devnull"):
     DEV_NULL = os.devnull
 else:
 if hasattr(os, "devnull"):
     DEV_NULL = os.devnull
 else:


@@ -205,18 +206,17 @@ def eintr_retry(func):
     return rv
 
 def rexec(command, host, user, 
     return rv
 
 def rexec(command, host, user, 

-        port = None, 
+        port = None,
+        gwuser = None,
+        gw = None, 

         agent = True,
         sudo = False,
         agent = True,
         sudo = False,

-        stdin = None,

         identity = None,
         server_key = None,
         env = None,
         tty = False,
         identity = None,
         server_key = None,
         env = None,
         tty = False,

-        timeout = None,
-        retry = 3,
-        err_on_timeout = True,

         connect_timeout = 30,
         connect_timeout = 30,

+        retry = 3,

         persistent = True,
         forward_x11 = False,
         blocking = True,
         persistent = True,
         forward_x11 = False,
         blocking = True,


@@ -226,7 +226,9 @@ def rexec(command, host, user,
     """
     
     tmp_known_hosts = None
     """
     
     tmp_known_hosts = None

-    hostip = gethostbyname(host)
+    if not gw:
+        hostip = gethostbyname(host)
+    else: hostip = None

 
     args = ['ssh', '-C',
             # Don't bother with localhost. Makes test easier
 
     args = ['ssh', '-C',
             # Don't bother with localhost. Makes test easier


@@ -235,6 +237,7 @@ def rexec(command, host, user,
             '-o', 'ConnectionAttempts=3',
             '-o', 'ServerAliveInterval=30',
             '-o', 'TCPKeepAlive=yes',
             '-o', 'ConnectionAttempts=3',
             '-o', 'ServerAliveInterval=30',
             '-o', 'TCPKeepAlive=yes',

+            '-o', 'Batchmode=yes',

             '-l', user, hostip or host]
 
     if persistent and openssh_has_persist():
             '-l', user, hostip or host]
 
     if persistent and openssh_has_persist():


@@ -247,6 +250,13 @@ def rexec(command, host, user,
         # Do not check for Host key. Unsafe.
         args.extend(['-o', 'StrictHostKeyChecking=no'])
 
         # Do not check for Host key. Unsafe.
         args.extend(['-o', 'StrictHostKeyChecking=no'])
 

+    if gw:
+        if gwuser:
+            proxycommand = 'ProxyCommand=ssh %s@%s -W %%h:%%p' % (gwuser, gw)
+        else:
+            proxycommand = 'ProxyCommand=ssh %%r@%s -W %%h:%%p' % gw
+        args.extend(['-o', proxycommand])
+

     if agent:
         args.append('-A')
 
     if agent:
         args.append('-A')
 


@@ -254,6 +264,7 @@ def rexec(command, host, user,
         args.append('-p%d' % port)
 
     if identity:
         args.append('-p%d' % port)
 
     if identity:

+        identity = os.path.expanduser(identity)

         args.extend(('-i', identity))
 
     if tty:
         args.extend(('-i', identity))
 
     if tty:


@@ -272,65 +283,26 @@ def rexec(command, host, user,
         command = "sudo " + command
 
     args.append(command)
         command = "sudo " + command
 
     args.append(command)

-
-    for x in xrange(retry):
-        # connects to the remote host and starts a remote connection
-        proc = subprocess.Popen(args,
-                env = env,
-                stdout = subprocess.PIPE,
-                stdin = subprocess.PIPE, 
-                stderr = subprocess.PIPE)
-        
-        # attach tempfile object to the process, to make sure the file stays
-        # alive until the process is finished with it
-        proc._known_hosts = tmp_known_hosts

     
     

-        # by default, rexec calls _communicate which will block 
-        # until the process has exit. The argument block == False 
-        # forces to rexec to return immediately, without blocking 
-        try:
-            if blocking:
-                out, err = _communicate(proc, stdin, timeout, err_on_timeout)
-            else:
-                out = err = ""
-                if proc.poll():
-                    err = self._proc.stderr.read()
-
-            msg = " rexec - host %s - command %s " % (host, " ".join(args))
-            log(msg, logging.DEBUG, out, err)
+    log_msg = " rexec - host %s - command %s " % (host, " ".join(args))

 
 

-            if proc.poll():
-                skip = False
-
-                if err.strip().startswith('ssh: ') or err.strip().startswith('mux_client_hello_exchange: '):
-                    # SSH error, can safely retry
-                    skip = True 
-                elif retry:
-                    # Probably timed out or plain failed but can retry
-                    skip = True 
-                
-                if skip:
-                    t = x*2
-                    msg = "SLEEPING %d ... ATEMPT %d - host %s - command %s " % ( 
-                            t, x, host, " ".join(args))
-                    log(msg, logging.DEBUG)
+    stdout = stderr = stdin = subprocess.PIPE
+    if forward_x11:
+        stdout = stderr = stdin = None

 
 

-                    time.sleep(t)
-                    continue
-            break
-        except RuntimeError, e:
-            msg = " rexec EXCEPTION - host %s - command %s - TIMEOUT -> %s" % (host, " ".join(args), e.args)
-            log(msg, logging.DEBUG, out, err)
-
-            if retry <= 0:
-                raise
-            retry -= 1
-        
-    return ((out, err), proc)
+    return _retry_rexec(args, log_msg, 
+            stderr = stderr,
+            stdin = stdin,
+            stdout = stdout,
+            env = env, 
+            retry = retry, 
+            tmp_known_hosts = tmp_known_hosts,
+            blocking = blocking)

 
 def rcopy(source, dest,
 
 def rcopy(source, dest,

-        port = None, 
-        agent = True, 
+        port = None,
+        gwuser = None,
+        gw = None,

         recursive = False,
         identity = None,
         server_key = None,
         recursive = False,
         identity = None,
         server_key = None,


@@ -342,263 +314,81 @@ def rcopy(source, dest,
     Source and destination should have the user and host encoded
     as per scp specs.
     
     Source and destination should have the user and host encoded
     as per scp specs.
     

-    If source is a file object, a special mode will be used to
-    create the remote file with the same contents.
-    
-    If dest is a file object, the remote file (source) will be
-    read and written into dest.
-    
-    In these modes, recursive cannot be True.
-    
-    Source can be a list of files to copy to a single destination,
-    in which case it is advised that the destination be a folder.
+    Source can be a list of files to copy to a single destination, 
+    (in which case it is advised that the destination be a folder),
+    a single file in a string or a semi-colon separated list of files
+    in a string.

     """
     """

-    
-    if isinstance(source, file) and source.tell() == 0:
-        source = source.name
-    elif hasattr(source, 'read'):
-        tmp = tempfile.NamedTemporaryFile()
-        while True:
-            buf = source.read(65536)
-            if buf:
-                tmp.write(buf)
-            else:
-                break
-        tmp.seek(0)
-        source = tmp.name
-    
-    if isinstance(source, file) or isinstance(dest, file) \
-            or hasattr(source, 'read')  or hasattr(dest, 'write'):
-        assert not recursive
-        
-        # Parse source/destination as <user>@<server>:<path>
-        if isinstance(dest, basestring) and ':' in dest:
-            remspec, path = dest.split(':',1)
-        elif isinstance(source, basestring) and ':' in source:
-            remspec, path = source.split(':',1)
-        else:
-            raise ValueError, "Both endpoints cannot be local"
-        user,host = remspec.rsplit('@',1)
-        
-        tmp_known_hosts = None
-        hostip = gethostbyname(host)
-        
-        args = ['ssh', '-l', user, '-C',
-                # Don't bother with localhost. Makes test easier
-                '-o', 'NoHostAuthenticationForLocalhost=yes',
-                '-o', 'ConnectTimeout=60',
-                '-o', 'ConnectionAttempts=3',
-                '-o', 'ServerAliveInterval=30',
-                '-o', 'TCPKeepAlive=yes',
-                hostip or host ]

 
 

-        if openssh_has_persist():
-            args.extend([
-                '-o', 'ControlMaster=auto',
-                '-o', 'ControlPath=%s' % (make_control_path(agent, False),),
-                '-o', 'ControlPersist=60' ])
-
-        if port:
-            args.append('-P%d' % port)
-
-        if identity:
-            args.extend(('-i', identity))
-
-        if server_key:
-            # Create a temporary server key file
-            tmp_known_hosts = make_server_key_args(server_key, host, port)
-            args.extend(['-o', 'UserKnownHostsFile=%s' % (tmp_known_hosts.name,)])
-        
-        if isinstance(source, file) or hasattr(source, 'read'):
-            args.append('cat > %s' % (shell_escape(path),))
-        elif isinstance(dest, file) or hasattr(dest, 'write'):
-            args.append('cat %s' % (shell_escape(path),))
-        else:
-            raise AssertionError, "Unreachable code reached! :-Q"
-        
-        # connects to the remote host and starts a remote connection
-        if isinstance(source, file):
-            proc = subprocess.Popen(args, 
-                    stdout = open('/dev/null','w'),
-                    stderr = subprocess.PIPE,
-                    stdin = source)
-            err = proc.stderr.read()
-            proc._known_hosts = tmp_known_hosts
-            eintr_retry(proc.wait)()
-            return ((None,err), proc)
-        elif isinstance(dest, file):
-            proc = subprocess.Popen(args, 
-                    stdout = open('/dev/null','w'),
-                    stderr = subprocess.PIPE,
-                    stdin = source)
-            err = proc.stderr.read()
-            proc._known_hosts = tmp_known_hosts
-            eintr_retry(proc.wait)()
-            return ((None,err), proc)
-        elif hasattr(source, 'read'):
-            # file-like (but not file) source
-            proc = subprocess.Popen(args, 
-                    stdout = open('/dev/null','w'),
-                    stderr = subprocess.PIPE,
-                    stdin = subprocess.PIPE)
-            
-            buf = None
-            err = []
-            while True:
-                if not buf:
-                    buf = source.read(4096)
-                if not buf:
-                    #EOF
-                    break
-                
-                rdrdy, wrdy, broken = select.select(
-                    [proc.stderr],
-                    [proc.stdin],
-                    [proc.stderr,proc.stdin])
-                
-                if proc.stderr in rdrdy:
-                    # use os.read for fully unbuffered behavior
-                    err.append(os.read(proc.stderr.fileno(), 4096))
-                
-                if proc.stdin in wrdy:
-                    proc.stdin.write(buf)
-                    buf = None
-                
-                if broken:
-                    break
-            proc.stdin.close()
-            err.append(proc.stderr.read())
-                
-            proc._known_hosts = tmp_known_hosts
-            eintr_retry(proc.wait)()
-            return ((None,''.join(err)), proc)
-        elif hasattr(dest, 'write'):
-            # file-like (but not file) dest
-            proc = subprocess.Popen(args, 
-                    stdout = subprocess.PIPE,
-                    stderr = subprocess.PIPE,
-                    stdin = open('/dev/null','w'))
-            
-            buf = None
-            err = []
-            while True:
-                rdrdy, wrdy, broken = select.select(
-                    [proc.stderr, proc.stdout],
-                    [],
-                    [proc.stderr, proc.stdout])
-                
-                if proc.stderr in rdrdy:
-                    # use os.read for fully unbuffered behavior
-                    err.append(os.read(proc.stderr.fileno(), 4096))
-                
-                if proc.stdout in rdrdy:
-                    # use os.read for fully unbuffered behavior
-                    buf = os.read(proc.stdout.fileno(), 4096)
-                    dest.write(buf)
-                    
-                    if not buf:
-                        #EOF
-                        break
-                
-                if broken:
-                    break
-            err.append(proc.stderr.read())
-                
-            proc._known_hosts = tmp_known_hosts
-            eintr_retry(proc.wait)()
-            return ((None,''.join(err)), proc)
-        else:
-            raise AssertionError, "Unreachable code reached! :-Q"
+    # Parse destination as <user>@<server>:<path>
+    if isinstance(dest, str) and ':' in dest:
+        remspec, path = dest.split(':',1)
+    elif isinstance(source, str) and ':' in source:
+        remspec, path = source.split(':',1)

     else:
     else:

-        # Parse destination as <user>@<server>:<path>
-        if isinstance(dest, basestring) and ':' in dest:
-            remspec, path = dest.split(':',1)
-        elif isinstance(source, basestring) and ':' in source:
-            remspec, path = source.split(':',1)
-        else:
-            raise ValueError, "Both endpoints cannot be local"
-        user,host = remspec.rsplit('@',1)
-        
-        # plain scp
-        tmp_known_hosts = None
-
-        args = ['scp', '-q', '-p', '-C',
-                # Speed up transfer using blowfish cypher specification which is 
-                # faster than the default one (3des)
-                '-c', 'blowfish',
-                # Don't bother with localhost. Makes test easier
-                '-o', 'NoHostAuthenticationForLocalhost=yes',
-                '-o', 'ConnectTimeout=60',
-                '-o', 'ConnectionAttempts=3',
-                '-o', 'ServerAliveInterval=30',
-                '-o', 'TCPKeepAlive=yes' ]
-                
-        if port:
-            args.append('-P%d' % port)
-
-        if recursive:
-            args.append('-r')
-
-        if identity:
-            args.extend(('-i', identity))
-
-        if server_key:
-            # Create a temporary server key file
-            tmp_known_hosts = make_server_key_args(server_key, host, port)
-            args.extend(['-o', 'UserKnownHostsFile=%s' % (tmp_known_hosts.name,)])
+        raise ValueError, "Both endpoints cannot be local"
+    user,host = remspec.rsplit('@',1)
+    
+    # plain scp
+    tmp_known_hosts = None

 
 

-        if not strict_host_checking:
-            # Do not check for Host key. Unsafe.
-            args.extend(['-o', 'StrictHostKeyChecking=no'])
+    args = ['scp', '-q', '-p', '-C',
+            # Speed up transfer using blowfish cypher specification which is 
+            # faster than the default one (3des)
+            '-c', 'blowfish',
+            # Don't bother with localhost. Makes test easier
+            '-o', 'NoHostAuthenticationForLocalhost=yes',
+            '-o', 'ConnectTimeout=60',
+            '-o', 'ConnectionAttempts=3',
+            '-o', 'ServerAliveInterval=30',
+            '-o', 'TCPKeepAlive=yes' ]
+            
+    if port:
+        args.append('-P%d' % port)

 
 

-        if isinstance(source,list):
-            args.extend(source)
+    if gw:
+        if gwuser:
+            proxycommand = 'ProxyCommand=ssh %s@%s -W %%h:%%p' % (gwuser, gw)

         else:
         else:

-            if openssh_has_persist():
-                args.extend([
-                    '-o', 'ControlMaster=auto',
-                    '-o', 'ControlPath=%s' % (make_control_path(agent, False),)
-                    ])
-            args.append(source)
+            proxycommand = 'ProxyCommand=ssh %%r@%s -W %%h:%%p' % gw
+        args.extend(['-o', proxycommand])

 
 

-        args.append(dest)
+    if recursive:
+        args.append('-r')

 
 

-        for x in xrange(retry):
-            # connects to the remote host and starts a remote connection
-            proc = subprocess.Popen(args,
-                    stdout = subprocess.PIPE,
-                    stdin = subprocess.PIPE, 
-                    stderr = subprocess.PIPE)
-            
-            # attach tempfile object to the process, to make sure the file stays
-            # alive until the process is finished with it
-            proc._known_hosts = tmp_known_hosts
-        
-            try:
-                (out, err) = proc.communicate()
-                eintr_retry(proc.wait)()
-                msg = " rcopy - host %s - command %s " % (host, " ".join(args))
-                log(msg, logging.DEBUG, out, err)
+    if identity:
+        identity = os.path.expanduser(identity)
+        args.extend(('-i', identity))

 
 

-                if proc.poll():
-                    t = x*2
-                    msg = "SLEEPING %d ... ATEMPT %d - host %s - command %s " % ( 
-                            t, x, host, " ".join(args))
-                    log(msg, logging.DEBUG)
+    if server_key:
+        # Create a temporary server key file
+        tmp_known_hosts = make_server_key_args(server_key, host, port)
+        args.extend(['-o', 'UserKnownHostsFile=%s' % (tmp_known_hosts.name,)])

 
 

-                    time.sleep(t)
-                    continue
+    if not strict_host_checking:
+        # Do not check for Host key. Unsafe.
+        args.extend(['-o', 'StrictHostKeyChecking=no'])
+    
+    if isinstance(source, list):
+        args.extend(source)
+    else:
+        if openssh_has_persist():
+            args.extend([
+                '-o', 'ControlMaster=auto',
+                '-o', 'ControlPath=%s' % (make_control_path(False, False),)
+                ])
+        args.append(source)

 
 

-                break
-            except RuntimeError, e:
-                msg = " rcopy EXCEPTION - host %s - command %s - TIMEOUT -> %s" % (host, " ".join(args), e.args)
-                log(msg, logging.DEBUG, out, err)
+    if isinstance(dest, list):
+        args.extend(dest)
+    else:
+        args.append(dest)

 
 

-                if retry <= 0:
-                    raise
-                retry -= 1
-            
-        return ((out, err), proc)
+    log_msg = " rcopy - host %s - command %s " % (host, " ".join(args))
+    
+    return _retry_rexec(args, log_msg, env = None, retry = retry, 
+            tmp_known_hosts = tmp_known_hosts,
+            blocking = True)

 
 def rspawn(command, pidfile, 
         stdout = '/dev/null', 
 
 def rspawn(command, pidfile, 
         stdout = '/dev/null', 


@@ -610,6 +400,8 @@ def rspawn(command, pidfile,
         host = None, 
         port = None, 
         user = None, 
         host = None, 
         port = None, 
         user = None, 

+        gwuser = None,
+        gw = None,

         agent = None, 
         identity = None, 
         server_key = None,
         agent = None, 
         identity = None, 
         server_key = None,


@@ -683,6 +475,8 @@ def rspawn(command, pidfile,
         host = host,
         port = port,
         user = user,
         host = host,
         port = port,
         user = user,

+        gwuser = gwuser,
+        gw = gw,

         agent = agent,
         identity = identity,
         server_key = server_key,
         agent = agent,
         identity = identity,
         server_key = server_key,


@@ -699,6 +493,8 @@ def rgetpid(pidfile,
         host = None, 
         port = None, 
         user = None, 
         host = None, 
         port = None, 
         user = None, 

+        gwuser = None,
+        gw = None,

         agent = None, 
         identity = None,
         server_key = None):
         agent = None, 
         identity = None,
         server_key = None):


@@ -725,6 +521,8 @@ def rgetpid(pidfile,
         host = host,
         port = port,
         user = user,
         host = host,
         port = port,
         user = user,

+        gwuser = gwuser,
+        gw = gw,

         agent = agent,
         identity = identity,
         server_key = server_key
         agent = agent,
         identity = identity,
         server_key = server_key


@@ -745,6 +543,8 @@ def rstatus(pid, ppid,
         host = None, 
         port = None, 
         user = None, 
         host = None, 
         port = None, 
         user = None, 

+        gwuser = None,
+        gw = None,

         agent = None, 
         identity = None,
         server_key = None):
         agent = None, 
         identity = None,
         server_key = None):


@@ -769,6 +569,8 @@ def rstatus(pid, ppid,
         host = host,
         port = port,
         user = user,
         host = host,
         port = port,
         user = user,

+        gwuser = gwuser,
+        gw = gw,

         agent = agent,
         identity = identity,
         server_key = server_key
         agent = agent,
         identity = identity,
         server_key = server_key


@@ -792,6 +594,8 @@ def rkill(pid, ppid,
         host = None, 
         port = None, 
         user = None, 
         host = None, 
         port = None, 
         user = None, 

+        gwuser = None,
+        gw = None,

         agent = None, 
         sudo = False,
         identity = None, 
         agent = None, 
         sudo = False,
         identity = None, 


@@ -846,6 +650,8 @@ fi
         host = host,
         port = port,
         user = user,
         host = host,
         port = port,
         user = user,

+        gwuser = gwuser,
+        gw = gw,

         agent = agent,
         identity = identity,
         server_key = server_key
         agent = agent,
         identity = identity,
         server_key = server_key


@@ -856,116 +662,67 @@ fi
 
     return (out, err), proc
 
 
     return (out, err), proc
 

-# POSIX
-def _communicate(self, input, timeout=None, err_on_timeout=True):
-    read_set = []
-    write_set = []
-    stdout = None # Return
-    stderr = None # Return
-    
-    killed = False
-    
-    if timeout is not None:
-        timelimit = time.time() + timeout
-        killtime = timelimit + 4
-        bailtime = timelimit + 4
-
-    if self.stdin:
-        # Flush stdio buffer.  This might block, if the user has
-        # been writing to .stdin in an uncontrolled fashion.
-        self.stdin.flush()
-        if input:
-            write_set.append(self.stdin)
-        else:
-            self.stdin.close()
-    if self.stdout:
-        read_set.append(self.stdout)
-        stdout = []
-    if self.stderr:
-        read_set.append(self.stderr)
-        stderr = []
-
-    input_offset = 0
-    while read_set or write_set:
-        if timeout is not None:
-            curtime = time.time()
-            if timeout is None or curtime > timelimit:
-                if curtime > bailtime:
-                    break
-                elif curtime > killtime:
-                    signum = signal.SIGKILL
-                else:
-                    signum = signal.SIGTERM
-                # Lets kill it
-                os.kill(self.pid, signum)
-                select_timeout = 0.5
-            else:
-                select_timeout = timelimit - curtime + 0.1
-        else:
-            select_timeout = 1.0
+def _retry_rexec(args,
+        log_msg,
+        stdout = subprocess.PIPE,
+        stdin = subprocess.PIPE, 
+        stderr = subprocess.PIPE,
+        env = None,
+        retry = 3,
+        tmp_known_hosts = None,
+        blocking = True):
+
+    for x in xrange(retry):
+        # connects to the remote host and starts a remote connection
+        proc = subprocess.Popen(args,
+                env = env,
+                stdout = stdout,
+                stdin = stdin, 
+                stderr = stderr)

         
         

-        if select_timeout > 1.0:
-            select_timeout = 1.0
-            
+        # attach tempfile object to the process, to make sure the file stays
+        # alive until the process is finished with it
+        proc._known_hosts = tmp_known_hosts
+    
+        # The argument block == False forces to rexec to return immediately, 
+        # without blocking 

         try:
         try:

-            rlist, wlist, xlist = select.select(read_set, write_set, [], select_timeout)
-        except select.error,e:
-            if e[0] != 4:
-                raise
-            else:
-                continue
-        
-        if not rlist and not wlist and not xlist and self.poll() is not None:
-            # timeout and process exited, say bye
+            err = out = " "
+            if blocking:
+                (out, err) = proc.communicate()
+            elif stdout:
+                out = proc.stdout.read()
+                if proc.poll() and stderr:
+                    err = proc.stderr.read()
+
+            log(log_msg, logging.DEBUG, out, err)
+
+            if proc.poll():
+                skip = False
+
+                if err.strip().startswith('ssh: ') or err.strip().startswith('mux_client_hello_exchange: '):
+                    # SSH error, can safely retry
+                    skip = True 
+                elif retry:
+                    # Probably timed out or plain failed but can retry
+                    skip = True 
+                
+                if skip:
+                    t = x*2
+                    msg = "SLEEPING %d ... ATEMPT %d - command %s " % ( 
+                            t, x, " ".join(args))
+                    log(msg, logging.DEBUG)
+
+                    time.sleep(t)
+                    continue

             break
             break

+        except RuntimeError, e:
+            msg = " rexec EXCEPTION - TIMEOUT -> %s \n %s" % ( e.args, log_msg )
+            log(msg, logging.DEBUG, out, err)

 
 

-        if self.stdin in wlist:
-            # When select has indicated that the file is writable,
-            # we can write up to PIPE_BUF bytes without risk
-            # blocking.  POSIX defines PIPE_BUF >= 512
-            bytes_written = os.write(self.stdin.fileno(), buffer(input, input_offset, 512))
-            input_offset += bytes_written
-            if input_offset >= len(input):
-                self.stdin.close()
-                write_set.remove(self.stdin)
-
-        if self.stdout in rlist:
-            data = os.read(self.stdout.fileno(), 1024)
-            if data == "":
-                self.stdout.close()
-                read_set.remove(self.stdout)
-            stdout.append(data)
-
-        if self.stderr in rlist:
-            data = os.read(self.stderr.fileno(), 1024)
-            if data == "":
-                self.stderr.close()
-                read_set.remove(self.stderr)
-            stderr.append(data)
-    
-    # All data exchanged.  Translate lists into strings.
-    if stdout is not None:
-        stdout = ''.join(stdout)
-    if stderr is not None:
-        stderr = ''.join(stderr)
-
-    # Translate newlines, if requested.  We cannot let the file
-    # object do the translation: It is based on stdio, which is
-    # impossible to combine with select (unless forcing no
-    # buffering).
-    if self.universal_newlines and hasattr(file, 'newlines'):
-        if stdout:
-            stdout = self._translate_newlines(stdout)
-        if stderr:
-            stderr = self._translate_newlines(stderr)
-
-    if killed and err_on_timeout:
-        errcode = self.poll()
-        raise RuntimeError, ("Operation timed out", errcode, stdout, stderr)
-    else:
-        if killed:
-            self.poll()
-        else:
-            self.wait()
-        return (stdout, stderr)
+            if retry <= 0:
+                raise
+            retry -= 1
+        
+    return ((out, err), proc)