Merge lp:~jameinel/bzr/2.1-client-reconnect-819604 into lp:bzr/2.1

[John A Meinel]

Start of bug #819604, allow bzr clients to reconnect if the connection is gone when we go to write a new request.

This is some initial work for having clients reconnect when their bzr+ssh (or bzr://) connection gets closed underneath them.

To start with, the trickiest bit is that it is really quite hard to detect when things are closed. Mostly because of buffering, etc, at lots of different levels (internal memory buffering, socket/pipe buffering, 'ssh.exe' buffering, latency of close message from server back to client, etc.)

I'll also clarify that this isn't intended to handle all possible sorts of connection failures. What we really are trying to handle is allowing a server to be upgraded 'live'. Such that the server will gently disconnect us only between complete requests. Versus having requests terminated randomly in the middle of content. It may help with some of those cases, and we certainly don't want to cause corruption, etc if one of those happens. But it isn't the goal of *this* change.

This specific patch has a few aspects, and does help some cases in real-world testing.

1) Backport a minimum amount of the SmartSSHClientMedium proxies to _real_medium patch from Andrew. In bzr-2.2 Andrew updated the code so that when spawning an 'ssh' subprocess, we use a socketpair when possible instead of pipes. (That way we can read without blocking, allowing us to use large buffers, etc.)

It also meant that he shared the logic from SmartSimplePipeStreamMedium and SmartTCPClientMedium, which I wanted to do, to avoid having to re-implement the logic multiple times.

This is a pretty small part of the patch.

2) Update some of the lower level code so that we get ConnectionReset rather than various IOError and ValueError when writing to a closed connection.

3) Update the _SmartClient._send_request logic. If we get a ConnectionReset while we are trying to write the request, then we can safely retry the request. On the assumption that as long as the server doesn't see the final 'e' terminal byte, it will reject the request, because it is incomplete.

4) The one caveat is if there is a 'body_stream'. body_stream is an iterable, so we can't just rewind it and resume it.

On the plus side, there is only one caller RemoteStreamSink.insert_stream. In theory we could update that caller. In practice, it *also* takes a 'stream' object, which can't simply be rewound. Though we do a 'no-op' stream just before the real content to determine that the remote server actually supports the RPC we want to use.

That would allow us to close the gap a little bit. So we can detect ConnectionReset all the way to the point that we actually start streaming the content.

The other bit that would be possible, is to update ProtocolThreeRequester.call_with_body_stream to allow it to set a flag to indicate whether it has actually started consuming the stream yet. Then update it to flush just before it iterates the body stream to help force detecting closed connections. (otherwise we are likely to have buffered at least the first part of the body_stream in local memory, since the local buffer is 1MB and the first flush is only called after the first chunk of the body_stream.)

This just closes a window where we won't reconnect during 'bzr push'.

5) The next step is that (especially for sockets), we write out the request successfully, and then notice it is closed when we try to read back the response.

Here I'm a lot more concerned about non-idempotent requests. Because the server might have fully read the request, and sent back a response, but we didn't get to see the response.

However, our RPC design is meant to be stateless (all state is passed in the request, not assumed to be stored on the server side). Which means things *should* be idempotent. For something like 'put_bytes', we may write the same content twice, but we shouldn't corrupt anything.

The only one I can think of is 'append_bytes', and the associated open_write_stream() code. (RemoteTransport.open_write_stream uses AppendBasedFileStream as its implementation.)

And that code shouldn't really be used with up-to-date bzr clients and servers. So I'm much more ok with it just dying if it isn't safe to retry.

[John A Meinel (jameinel)]

I should note, I think it would be prudent to land this into bzr.dev first, but I wanted to make sure it was written against 2.1 since that is the oldest client that we want to backport support for.

Also, I expect there will be a modest amount of conflicts bringing this up from bzr-2.1 into bzr.dev. There has been a modest amount of changes inbetween, so I'll try to produce branches for each series.

[John A Meinel (jameinel)]

Small update, in working on the read-reconnect code, I realized passing all these arguments around was very clumsy. So I refactored the code into a helper class _SmartClientRequest. That way, the 6 arguments being passed around just end up as attributes, and the function call interplay is easier to follow.

[Vincent Ladeuil (vila)]

> I should note, I think it would be prudent to land this into bzr.dev first,
> but I wanted to make sure it was written against 2.1 since that is the oldest
> client that we want to backport support for.

+4 (+1 for each of 2.1, 2.2, 2.3, 2.4)

No need to ask anymore, good, keep reading my mind ;)

I don't know exactly how we should proceed but as much as possible, all the modifications related to this topic should be done in a separate branch which should be merged into whatever series we target.

Backport is hard, let's avoid it as much as possible (but no more ;)

> Also, I expect there will be a modest amount of conflicts bringing this up
> from bzr-2.1 into bzr.dev. There has been a modest amount of changes
> inbetween, so I'll try to produce branches for each series.

Yup, that sounds a lot like a loom with a thread by targeted series. Or any arrangement of branches that makes our life easier.

[Vincent Ladeuil (vila)]

> Small update, in working on the read-reconnect code, I realized passing all
> these arguments around was very clumsy. So I refactored the code into a helper
> class _SmartClientRequest. That way, the 6 arguments being passed around just
> end up as attributes, and the function call interplay is easier to follow.

From the description alone this part got my approval. If some helpers need to move there too, clarifying the class features, even better ( no pre-conceptions here, just saying).

[John A Meinel (jameinel)]

Some small code cleanups, and add -Dnoretry as a debugging flag that will disable the retry-on-write.

[Martin Pool (mbp)]

I'm going to (reluctantly) bump these out of the merge queue until this is landed and well tested in 2.5 - probably until after the 2.5.0 final release is out. Then we can come back and look at landing to earlier series.

[John A Meinel (jameinel)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

On 11/29/2011 7:32 AM, Martin Pool wrote:
> I'm going to (reluctantly) bump these out of the merge queue until
> this is landed and well tested in 2.5 - probably until after the
> 2.5.0 final release is out. Then we can come back and look at
> landing to earlier series.

I don't think it is something you should feel bad about :). That was
certainly the intent.

1) I wrote it against 2.1 so we would have a chance to land it there
   cleanly.

2) I tried to split it up into multiple patches to make it easier to
   review.

3) I merged it up through the stack because there are a fair number of
   changes to this code, and thus conflicts, etc.

4) It should certainly land on trunk and get a thorough shakeout there.
   I'm not sure how much time you think is enough. Certainly if enough
   time passes, 2.1 will be fully obsolete :).

Anyway, thanks for getting this landed in 2.5.

John
=:->

-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.9 (Cygwin)
Comment: Using GnuPG with Mozilla - http://enigmail.mozdev.org/

iEYEARECAAYFAk7UiEcACgkQJdeBCYSNAANSFwCgiwx+EeIpiYLJF84P0NPcLFd5
T90AoJDimVbLmeDiqn1FWFDmdWtMXaWq
=94bN
-----END PGP SIGNATURE-----

=== modified file 'bzrlib/help_topics/en/debug-flags.txt'
--- bzrlib/help_topics/en/debug-flags.txt	2010-01-05 04:30:07 +0000
+++ bzrlib/help_topics/en/debug-flags.txt	2012-09-11 12:31:20 +0000
@@ -24,6 +24,8 @@
 -Dindex           Trace major index operations.
 -Dknit            Trace knit operations.
 -Dlock            Trace when lockdir locks are taken or released.
+-Dnoretry         If a connection is reset, fail immediately rather than
+                  retrying the request.
 -Dprogress        Trace progress bar operations.
 -Dmerge           Emit information for debugging merges.
 -Dno_apport       Don't use apport to report crashes.

=== modified file 'bzrlib/osutils.py'
--- bzrlib/osutils.py	2010-05-27 04:00:01 +0000
+++ bzrlib/osutils.py	2012-09-11 12:31:20 +0000
@@ -40,6 +40,7 @@
     rmtree,
     )
 import signal
+import socket
 import subprocess
 import tempfile
 from tempfile import (
@@ -1929,6 +1930,20 @@
         return socket.gethostname().decode(get_user_encoding())
 
 
+# We must not read/write any more than 64k at a time from/to a socket so we
+# don't risk "no buffer space available" errors on some platforms.  Windows in
+# particular is likely to throw WSAECONNABORTED or WSAENOBUFS if given too much
+# data at once.
+MAX_SOCKET_CHUNK = 64 * 1024
+
+_end_of_stream_errors = [errno.ECONNRESET, errno.EPIPE, errno.EINVAL]
+for _eno in ['WSAECONNRESET', 'WSAECONNABORTED']:
+    _eno = getattr(errno, _eno, None)
+    if _eno is not None:
+        _end_of_stream_errors.append(_eno)
+del _eno
+
+
 def recv_all(socket, bytes):
     """Receive an exact number of bytes.
 
@@ -1948,21 +1963,37 @@
     return b
 
 
-def send_all(socket, bytes, report_activity=None):
+def send_all(sock, bytes, report_activity=None):
     """Send all bytes on a socket.
 
-    Regular socket.sendall() can give socket error 10053 on Windows.  This
-    implementation sends no more than 64k at a time, which avoids this problem.
+    Breaks large blocks in smaller chunks to avoid buffering limitations on
+    some platforms, and catches EINTR which may be thrown if the send is
+    interrupted by a signal.
+
+    This is preferred to socket.sendall(), because it avoids portability bugs
+    and provides activity reporting.
 
     :param report_activity: Call this as bytes are read, see
         Transport._report_activity
     """
-    chunk_size = 2**16
-    for pos in xrange(0, len(bytes), chunk_size):
-        block = bytes[pos:pos+chunk_size]
-        if report_activity is not None:
-            report_activity(len(block), 'write')
-        until_no_eintr(socket.sendall, block)
+    sent_total = 0
+    byte_count = len(bytes)
+    while sent_total < byte_count:
+        try:
+            sent = sock.send(buffer(bytes, sent_total, MAX_SOCKET_CHUNK))
+        except (socket.error, IOError), e:
+            if e.args[0] in _end_of_stream_errors:
+                raise errors.ConnectionReset(
+                    "Error trying to write to socket", e)
+            if e.args[0] != errno.EINTR:
+                raise
+        else:
+            if sent == 0:
+                raise errors.ConnectionReset('Sending to %s returned 0 bytes'
+                                             % (sock,))
+            sent_total += sent
+            if report_activity is not None:
+                report_activity(sent, 'write')
 
 
 def dereference_path(path):

=== modified file 'bzrlib/smart/client.py'
--- bzrlib/smart/client.py	2010-02-17 17:11:16 +0000
+++ bzrlib/smart/client.py	2012-09-11 12:31:20 +0000
@@ -16,10 +16,11 @@
 
 import bzrlib
 from bzrlib.smart import message, protocol
-from bzrlib.trace import warning
 from bzrlib import (
+    debug,
     errors,
     hooks,
+    trace,
     )
 
 
@@ -39,93 +40,12 @@
     def __repr__(self):
         return '%s(%r)' % (self.__class__.__name__, self._medium)
 
-    def _send_request(self, protocol_version, method, args, body=None,
-                      readv_body=None, body_stream=None):
-        encoder, response_handler = self._construct_protocol(
-            protocol_version)
-        encoder.set_headers(self._headers)
-        if body is not None:
-            if readv_body is not None:
-                raise AssertionError(
-                    "body and readv_body are mutually exclusive.")
-            if body_stream is not None:
-                raise AssertionError(
-                    "body and body_stream are mutually exclusive.")
-            encoder.call_with_body_bytes((method, ) + args, body)
-        elif readv_body is not None:
-            if body_stream is not None:
-                raise AssertionError(
-                    "readv_body and body_stream are mutually exclusive.")
-            encoder.call_with_body_readv_array((method, ) + args, readv_body)
-        elif body_stream is not None:
-            encoder.call_with_body_stream((method, ) + args, body_stream)
-        else:
-            encoder.call(method, *args)
-        return response_handler
-
-    def _run_call_hooks(self, method, args, body, readv_body):
-        if not _SmartClient.hooks['call']:
-            return
-        params = CallHookParams(method, args, body, readv_body, self._medium)
-        for hook in _SmartClient.hooks['call']:
-            hook(params)
-
     def _call_and_read_response(self, method, args, body=None, readv_body=None,
             body_stream=None, expect_response_body=True):
-        self._run_call_hooks(method, args, body, readv_body)
-        if self._medium._protocol_version is not None:
-            response_handler = self._send_request(
-                self._medium._protocol_version, method, args, body=body,
-                readv_body=readv_body, body_stream=body_stream)
-            return (response_handler.read_response_tuple(
-                        expect_body=expect_response_body),
-                    response_handler)
-        else:
-            for protocol_version in [3, 2]:
-                if protocol_version == 2:
-                    # If v3 doesn't work, the remote side is older than 1.6.
-                    self._medium._remember_remote_is_before((1, 6))
-                response_handler = self._send_request(
-                    protocol_version, method, args, body=body,
-                    readv_body=readv_body, body_stream=body_stream)
-                try:
-                    response_tuple = response_handler.read_response_tuple(
-                        expect_body=expect_response_body)
-                except errors.UnexpectedProtocolVersionMarker, err:
-                    # TODO: We could recover from this without disconnecting if
-                    # we recognise the protocol version.
-                    warning(
-                        'Server does not understand Bazaar network protocol %d,'
-                        ' reconnecting.  (Upgrade the server to avoid this.)'
-                        % (protocol_version,))
-                    self._medium.disconnect()
-                    continue
-                except errors.ErrorFromSmartServer:
-                    # If we received an error reply from the server, then it
-                    # must be ok with this protocol version.
-                    self._medium._protocol_version = protocol_version
-                    raise
-                else:
-                    self._medium._protocol_version = protocol_version
-                    return response_tuple, response_handler
-            raise errors.SmartProtocolError(
-                'Server is not a Bazaar server: ' + str(err))
-
-    def _construct_protocol(self, version):
-        request = self._medium.get_request()
-        if version == 3:
-            request_encoder = protocol.ProtocolThreeRequester(request)
-            response_handler = message.ConventionalResponseHandler()
-            response_proto = protocol.ProtocolThreeDecoder(
-                response_handler, expect_version_marker=True)
-            response_handler.setProtoAndMediumRequest(response_proto, request)
-        elif version == 2:
-            request_encoder = protocol.SmartClientRequestProtocolTwo(request)
-            response_handler = request_encoder
-        else:
-            request_encoder = protocol.SmartClientRequestProtocolOne(request)
-            response_handler = request_encoder
-        return request_encoder, response_handler
+        request = _SmartClientRequest(self, method, args, body=body,
+            readv_body=readv_body, body_stream=body_stream,
+            expect_response_body=expect_response_body)
+        return request.call_and_read_response()
 
     def call(self, method, *args):
         """Call a method on the remote server."""
@@ -191,6 +111,170 @@
         return self._medium.remote_path_from_transport(transport)
 
 
+class _SmartClientRequest(object):
+    """Encapsulate the logic for a single request.
+
+    This class handles things like reconnecting and sending the request a
+    second time when the connection is reset in the middle. It also handles the
+    multiple requests that get made if we don't know what protocol the server
+    supports yet.
+
+    Generally, you build up one of these objects, passing in the arguments that
+    you want to send to the server, and then use 'call_and_read_response' to
+    get the response from the server.
+    """
+
+    def __init__(self, client, method, args, body=None, readv_body=None,
+                 body_stream=None, expect_response_body=True):
+        self.client = client
+        self.method = method
+        self.args = args
+        self.body = body
+        self.readv_body = readv_body
+        self.body_stream = body_stream
+        self.expect_response_body = expect_response_body
+
+    def call_and_read_response(self):
+        """Send the request to the server, and read the initial response.
+
+        This doesn't read all of the body content of the response, instead it
+        returns (response_tuple, response_handler). response_tuple is the 'ok',
+        or 'error' information, and 'response_handler' can be used to get the
+        content stream out.
+        """
+        self._run_call_hooks()
+        protocol_version = self.client._medium._protocol_version
+        if protocol_version is None:
+            return self._call_determining_protocol_version()
+        else:
+            return self._call(protocol_version)
+
+    def _run_call_hooks(self):
+        if not _SmartClient.hooks['call']:
+            return
+        params = CallHookParams(self.method, self.args, self.body,
+                                self.readv_body, self.client._medium)
+        for hook in _SmartClient.hooks['call']:
+            hook(params)
+
+    def _call(self, protocol_version):
+        """We know the protocol version.
+
+        So this just sends the request, and then reads the response. This is
+        where the code will be to retry requests if the connection is closed.
+        """
+        response_handler = self._send(protocol_version)
+        response_tuple = response_handler.read_response_tuple(
+            expect_body=self.expect_response_body)
+        return (response_tuple, response_handler)
+
+    def _call_determining_protocol_version(self):
+        """Determine what protocol the remote server supports.
+
+        We do this by placing a request in the most recent protocol, and
+        handling the UnexpectedProtocolVersionMarker from the server.
+        """
+        for protocol_version in [3, 2]:
+            if protocol_version == 2:
+                # If v3 doesn't work, the remote side is older than 1.6.
+                self.client._medium._remember_remote_is_before((1, 6))
+            try:
+                response_tuple, response_handler = self._call(protocol_version)
+            except errors.UnexpectedProtocolVersionMarker, err:
+                # TODO: We could recover from this without disconnecting if
+                # we recognise the protocol version.
+                trace.warning(
+                    'Server does not understand Bazaar network protocol %d,'
+                    ' reconnecting.  (Upgrade the server to avoid this.)'
+                    % (protocol_version,))
+                self.client._medium.disconnect()
+                continue
+            except errors.ErrorFromSmartServer:
+                # If we received an error reply from the server, then it
+                # must be ok with this protocol version.
+                self.client._medium._protocol_version = protocol_version
+                raise
+            else:
+                self.client._medium._protocol_version = protocol_version
+                return response_tuple, response_handler
+        raise errors.SmartProtocolError(
+            'Server is not a Bazaar server: ' + str(err))
+
+    def _construct_protocol(self, version):
+        """Build the encoding stack for a given protocol version."""
+        request = self.client._medium.get_request()
+        if version == 3:
+            request_encoder = protocol.ProtocolThreeRequester(request)
+            response_handler = message.ConventionalResponseHandler()
+            response_proto = protocol.ProtocolThreeDecoder(
+                response_handler, expect_version_marker=True)
+            response_handler.setProtoAndMediumRequest(response_proto, request)
+        elif version == 2:
+            request_encoder = protocol.SmartClientRequestProtocolTwo(request)
+            response_handler = request_encoder
+        else:
+            request_encoder = protocol.SmartClientRequestProtocolOne(request)
+            response_handler = request_encoder
+        return request_encoder, response_handler
+
+    def _send(self, protocol_version):
+        """Encode the request, and send it to the server.
+
+        This will retry a request if we get a ConnectionReset while sending the
+        request to the server. (Unless we have a body_stream that we have
+        already started consuming, since we can't restart body_streams)
+
+        :return: response_handler as defined by _construct_protocol
+        """
+        encoder, response_handler = self._construct_protocol(protocol_version)
+        try:
+            self._send_no_retry(encoder)
+        except errors.ConnectionReset, e:
+            # If we fail during the _send_no_retry phase, then we can
+            # be confident that the server did not get our request, because we
+            # haven't started waiting for the reply yet. So try the request
+            # again. We only issue a single retry, because if the connection
+            # really is down, there is no reason to loop endlessly.
+
+            # Connection is dead, so close our end of it.
+            self.client._medium.reset()
+            if (('noretry' in debug.debug_flags)
+                or self.body_stream is not None):
+                # We can't restart a body_stream that has been partially
+                # consumed, so we don't retry.
+                raise
+            trace.warning('ConnectionReset calling %r, retrying'
+                          % (self.method,))
+            trace.log_exception_quietly()
+            encoder, response_handler = self._construct_protocol(
+                protocol_version)
+            self._send_no_retry(encoder)
+        return response_handler
+
+    def _send_no_retry(self, encoder):
+        """Just encode the request and try to send it."""
+        encoder.set_headers(self.client._headers)
+        if self.body is not None:
+            if self.readv_body is not None:
+                raise AssertionError(
+                    "body and readv_body are mutually exclusive.")
+            if self.body_stream is not None:
+                raise AssertionError(
+                    "body and body_stream are mutually exclusive.")
+            encoder.call_with_body_bytes((self.method, ) + self.args, self.body)
+        elif self.readv_body is not None:
+            if self.body_stream is not None:
+                raise AssertionError(
+                    "readv_body and body_stream are mutually exclusive.")
+            encoder.call_with_body_readv_array((self.method, ) + self.args,
+                                               self.readv_body)
+        elif self.body_stream is not None:
+            encoder.call_with_body_stream((self.method, ) + self.args,
+                                          self.body_stream)
+        else:
+            encoder.call(self.method, *self.args)
+
+
 class SmartClientHooks(hooks.Hooks):
 
     def __init__(self):

=== modified file 'bzrlib/smart/medium.py'
--- bzrlib/smart/medium.py	2010-04-27 06:40:37 +0000
+++ bzrlib/smart/medium.py	2012-09-11 12:31:20 +0000
@@ -712,6 +712,14 @@
         """
         return SmartClientStreamMediumRequest(self)
 
+    def reset(self):
+        """We have been disconnected, reset current state.
+
+        This resets things like _current_request and connected state.
+        """
+        self.disconnect()
+        self._current_request = None
+
 
 class SmartSimplePipesClientMedium(SmartClientStreamMedium):
     """A client medium using simple pipes.
@@ -726,11 +734,21 @@
 
     def _accept_bytes(self, bytes):
         """See SmartClientStreamMedium.accept_bytes."""
-        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
+        try:
+            osutils.until_no_eintr(self._writeable_pipe.write, bytes)
+        except IOError, e:
+            if e.errno in (errno.EINVAL, errno.EPIPE):
+                raise errors.ConnectionReset(
+                    "Error trying to write to subprocess:\n%s"
+                    % (e,))
+            raise
         self._report_activity(len(bytes), 'write')
 
     def _flush(self):
         """See SmartClientStreamMedium._flush()."""
+        # Note: If flush were to fail, we'd like to raise ConnectionReset, etc.
+        #       However, testing shows that even when the child process is
+        #       gone, this doesn't error.
         osutils.until_no_eintr(self._writeable_pipe.flush)
 
     def _read_bytes(self, count):
@@ -741,7 +759,10 @@
 
 
 class SmartSSHClientMedium(SmartClientStreamMedium):
-    """A client medium using SSH."""
+    """A client medium using SSH.
+
+    It delegates IO to a SmartSimplePipesClientMedium.
+    """
 
     def __init__(self, host, port=None, username=None, password=None,
             base=None, vendor=None, bzr_remote_path=None):
@@ -750,11 +771,11 @@
         :param vendor: An optional override for the ssh vendor to use. See
             bzrlib.transport.ssh for details on ssh vendors.
         """
-        self._connected = False
         self._host = host
         self._password = password
         self._port = port
         self._username = username
+        self._real_medium = None
         # for the benefit of progress making a short description of this
         # transport
         self._scheme = 'bzr+ssh'
@@ -762,10 +783,8 @@
         # _DebugCounter so we have to store all the values used in our repr
         # method before calling the super init.
         SmartClientStreamMedium.__init__(self, base)
-        self._read_from = None
         self._ssh_connection = None
         self._vendor = vendor
-        self._write_to = None
         self._bzr_remote_path = bzr_remote_path
 
     def __repr__(self):
@@ -783,21 +802,20 @@
     def _accept_bytes(self, bytes):
         """See SmartClientStreamMedium.accept_bytes."""
         self._ensure_connection()
-        osutils.until_no_eintr(self._write_to.write, bytes)
-        self._report_activity(len(bytes), 'write')
+        self._real_medium.accept_bytes(bytes)
 
     def disconnect(self):
         """See SmartClientMedium.disconnect()."""
-        if not self._connected:
-            return
-        osutils.until_no_eintr(self._read_from.close)
-        osutils.until_no_eintr(self._write_to.close)
-        self._ssh_connection.close()
-        self._connected = False
+        if self._real_medium is not None:
+            self._real_medium.disconnect()
+            self._real_medium = None
+        if self._ssh_connection is not None:
+            self._ssh_connection.close()
+            self._ssh_connection = None
 
     def _ensure_connection(self):
         """Connect this medium if not already connected."""
-        if self._connected:
+        if self._real_medium is not None:
             return
         if self._vendor is None:
             vendor = ssh._get_ssh_vendor()
@@ -807,22 +825,19 @@
                 self._password, self._host, self._port,
                 command=[self._bzr_remote_path, 'serve', '--inet',
                          '--directory=/', '--allow-writes'])
-        self._read_from, self._write_to = \
-            self._ssh_connection.get_filelike_channels()
-        self._connected = True
+        read_from, write_to = self._ssh_connection.get_filelike_channels()
+        self._real_medium = SmartSimplePipesClientMedium(
+            read_from, write_to, self.base)
 
     def _flush(self):
         """See SmartClientStreamMedium._flush()."""
-        self._write_to.flush()
+        self._real_medium._flush()
 
     def _read_bytes(self, count):
         """See SmartClientStreamMedium.read_bytes."""
-        if not self._connected:
+        if self._real_medium is None:
             raise errors.MediumNotConnected(self)
-        bytes_to_read = min(count, _MAX_READ_SIZE)
-        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
-        self._report_activity(len(bytes), 'read')
-        return bytes
+        return self._real_medium.read_bytes(count)
 
 
 # Port 4155 is the default port for bzr://, registered with IANA.
@@ -948,13 +963,17 @@
         self._medium._flush()
 
 
+WSAECONNABORTED = 10053
+WSAECONNRESET = 10054
+
 def _read_bytes_from_socket(sock, desired_count, report_activity):
     # We ignore the desired_count because on sockets it's more efficient to
     # read large chunks (of _MAX_READ_SIZE bytes) at a time.
     try:
         bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
     except socket.error, e:
-        if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
+        if len(e.args) and e.args[0] in (errno.ECONNRESET, WSAECONNABORTED,
+                                         WSAECONNRESET):
             # The connection was closed by the other side.  Callers expect an
             # empty string to signal end-of-stream.
             bytes = ''

=== modified file 'bzrlib/smart/protocol.py'
--- bzrlib/smart/protocol.py	2010-02-17 17:11:16 +0000
+++ bzrlib/smart/protocol.py	2012-09-11 12:31:20 +0000
@@ -1075,9 +1075,6 @@
         self._real_write_func = write_func
 
     def _write_func(self, bytes):
-        # TODO: It is probably more appropriate to use sum(map(len, _buf))
-        #       for total number of bytes to write, rather than buffer based on
-        #       the number of write() calls
         # TODO: Another possibility would be to turn this into an async model.
         #       Where we let another thread know that we have some bytes if
         #       they want it, but we don't actually block for it

=== modified file 'bzrlib/tests/test_osutils.py'
--- bzrlib/tests/test_osutils.py	2010-11-30 20:42:42 +0000
+++ bzrlib/tests/test_osutils.py	2012-09-11 12:31:20 +0000
@@ -801,6 +801,45 @@
         self.assertEqual(None, osutils.safe_file_id(None))
 
 
+class TestSendAll(tests.TestCase):
+
+    def test_send_with_disconnected_socket(self):
+        class DisconnectedSocket(object):
+            def __init__(self, err):
+                self.err = err
+            def send(self, content):
+                raise self.err
+            def close(self):
+                pass
+        # All of these should be treated as ConnectionReset
+        errs = []
+        for err_cls in (IOError, socket.error):
+            for errnum in osutils._end_of_stream_errors:
+                errs.append(err_cls(errnum))
+        for err in errs:
+            sock = DisconnectedSocket(err)
+            self.assertRaises(errors.ConnectionReset,
+                osutils.send_all, sock, 'some more content')
+
+    def test_send_with_no_progress(self):
+        # See https://bugs.launchpad.net/bzr/+bug/1047309
+        # It seems that paramiko can get into a state where it doesn't error,
+        # but it returns 0 bytes sent for requests over and over again.
+        class NoSendingSocket(object):
+            def __init__(self):
+                self.call_count = 0
+            def send(self, bytes):
+                self.call_count += 1
+                if self.call_count > 100:
+                    # Prevent the test suite from hanging
+                    raise RuntimeError('too many calls')
+                return 0
+        sock = NoSendingSocket()
+        self.assertRaises(errors.ConnectionReset,
+                          osutils.send_all, sock, 'content')
+        self.assertEqual(1, sock.call_count)
+
+
 class TestWin32Funcs(tests.TestCase):
     """Test that _win32 versions of os utilities return appropriate paths."""
 

=== modified file 'bzrlib/tests/test_smart_transport.py'
--- bzrlib/tests/test_smart_transport.py	2010-02-17 17:11:16 +0000
+++ bzrlib/tests/test_smart_transport.py	2012-09-11 12:31:20 +0000
@@ -18,13 +18,17 @@
 
 # all of this deals with byte strings so this is safe
 from cStringIO import StringIO
+import errno
 import os
 import socket
+import subprocess
+import sys
 import threading
 
 import bzrlib
 from bzrlib import (
         bzrdir,
+        debug,
         errors,
         osutils,
         tests,
@@ -49,6 +53,29 @@
 from bzrlib.transport.http import SmartClientHTTPMediumRequest
 
 
+def create_file_pipes():
+    r, w = os.pipe()
+    # These must be opened without buffering, or we get undefined results
+    rf = os.fdopen(r, 'rb', 0)
+    wf = os.fdopen(w, 'wb', 0)
+    return rf, wf
+
+
+def portable_socket_pair():
+    """Return a pair of TCP sockets connected to each other.
+
+    Unlike socket.socketpair, this should work on Windows.
+    """
+    listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listen_sock.bind(('127.0.0.1', 0))
+    listen_sock.listen(1)
+    client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    client_sock.connect(listen_sock.getsockname())
+    server_sock, addr = listen_sock.accept()
+    listen_sock.close()
+    return server_sock, client_sock
+
+
 class StringIOSSHVendor(object):
     """A SSH vendor that uses StringIO to buffer writes and answer reads."""
 
@@ -63,6 +90,27 @@
         return StringIOSSHConnection(self)
 
 
+class FirstRejectedStringIOSSHVendor(StringIOSSHVendor):
+    """The first connection will be considered closed.
+
+    The second connection will succeed normally.
+    """
+
+    def __init__(self, read_from, write_to, fail_at_write=True):
+        super(FirstRejectedStringIOSSHVendor, self).__init__(read_from,
+            write_to)
+        self.fail_at_write = fail_at_write
+        self._first = True
+
+    def connect_ssh(self, username, password, host, port, command):
+        self.calls.append(('connect_ssh', username, password, host, port,
+            command))
+        if self._first:
+            self._first = False
+            return ClosedSSHConnection(self)
+        return StringIOSSHConnection(self)
+
+
 class StringIOSSHConnection(object):
     """A SSH connection that uses StringIO to buffer writes and answer reads."""
 
@@ -71,11 +119,36 @@
 
     def close(self):
         self.vendor.calls.append(('close', ))
+        self.vendor.read_from.close()
+        self.vendor.write_to.close()
 
     def get_filelike_channels(self):
         return self.vendor.read_from, self.vendor.write_to
 
 
+class ClosedSSHConnection(object):
+    """An SSH connection that just has closed channels."""
+
+    def __init__(self, vendor):
+        self.vendor = vendor
+
+    def close(self):
+        self.vendor.calls.append(('close', ))
+
+    def get_filelike_channels(self):
+        # We create matching pipes, and then close the ssh side
+        bzr_read, ssh_write = create_file_pipes()
+        # We always fail when bzr goes to read
+        ssh_write.close()
+        if self.vendor.fail_at_write:
+            # If set, we'll also fail when bzr goes to write
+            ssh_read, bzr_write = create_file_pipes()
+            ssh_read.close()
+        else:
+            bzr_write = self.vendor.write_to
+        return bzr_read, bzr_write
+
+
 class _InvalidHostnameFeature(tests.Feature):
     """Does 'non_existent.invalid' fail to resolve?
 
@@ -171,6 +244,91 @@
         client_medium._accept_bytes('abc')
         self.assertEqual('abc', output.getvalue())
 
+    def test_simple_pipes__accept_bytes_subprocess_closed(self):
+        # It is unfortunate that we have to use Popen for this. However,
+        # os.pipe() does not behave the same as subprocess.Popen().
+        # On Windows, if you use os.pipe() and close the write side,
+        # read.read() hangs. On Linux, read.read() returns the empty string.
+        p = subprocess.Popen([sys.executable, '-c',
+            'import sys\n'
+            'sys.stdout.write(sys.stdin.read(4))\n'
+            'sys.stdout.close()\n'],
+            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+        client_medium = medium.SmartSimplePipesClientMedium(
+            p.stdout, p.stdin, 'base')
+        client_medium._accept_bytes('abc\n')
+        self.assertEqual('abc', client_medium._read_bytes(3))
+        p.wait()
+        # While writing to the underlying pipe,
+        #   Windows py2.6.6 we get IOError(EINVAL)
+        #   Lucid py2.6.5, we get IOError(EPIPE)
+        # In both cases, it should be wrapped to ConnectionReset
+        self.assertRaises(errors.ConnectionReset,
+                          client_medium._accept_bytes, 'more')
+
+    def test_simple_pipes__accept_bytes_pipe_closed(self):
+        child_read, client_write = create_file_pipes()
+        client_medium = medium.SmartSimplePipesClientMedium(
+            None, client_write, 'base')
+        client_medium._accept_bytes('abc\n')
+        self.assertEqual('abc\n', child_read.read(4))
+        # While writing to the underlying pipe,
+        #   Windows py2.6.6 we get IOError(EINVAL)
+        #   Lucid py2.6.5, we get IOError(EPIPE)
+        # In both cases, it should be wrapped to ConnectionReset
+        child_read.close()
+        self.assertRaises(errors.ConnectionReset,
+                          client_medium._accept_bytes, 'more')
+
+    def test_simple_pipes__flush_pipe_closed(self):
+        child_read, client_write = create_file_pipes()
+        client_medium = medium.SmartSimplePipesClientMedium(
+            None, client_write, 'base')
+        client_medium._accept_bytes('abc\n')
+        child_read.close()
+        # Even though the pipe is closed, flush on the write side seems to be a
+        # no-op, rather than a failure.
+        client_medium._flush()
+
+    def test_simple_pipes__flush_subprocess_closed(self):
+        p = subprocess.Popen([sys.executable, '-c',
+            'import sys\n'
+            'sys.stdout.write(sys.stdin.read(4))\n'
+            'sys.stdout.close()\n'],
+            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+        client_medium = medium.SmartSimplePipesClientMedium(
+            p.stdout, p.stdin, 'base')
+        client_medium._accept_bytes('abc\n')
+        p.wait()
+        # Even though the child process is dead, flush seems to be a no-op.
+        client_medium._flush()
+
+    def test_simple_pipes__read_bytes_pipe_closed(self):
+        child_read, client_write = create_file_pipes()
+        client_medium = medium.SmartSimplePipesClientMedium(
+            child_read, client_write, 'base')
+        client_medium._accept_bytes('abc\n')
+        client_write.close()
+        self.assertEqual('abc\n', client_medium._read_bytes(4))
+        self.assertEqual('', client_medium._read_bytes(4))
+
+    def test_simple_pipes__read_bytes_subprocess_closed(self):
+        p = subprocess.Popen([sys.executable, '-c',
+            'import sys\n'
+            'if sys.platform == "win32":\n'
+            '    import msvcrt, os\n'
+            '    msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)\n'
+            '    msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)\n'
+            'sys.stdout.write(sys.stdin.read(4))\n'
+            'sys.stdout.close()\n'],
+            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+        client_medium = medium.SmartSimplePipesClientMedium(
+            p.stdout, p.stdin, 'base')
+        client_medium._accept_bytes('abc\n')
+        p.wait()
+        self.assertEqual('abc\n', client_medium._read_bytes(4))
+        self.assertEqual('', client_medium._read_bytes(4))
+
     def test_simple_pipes_client_disconnect_does_nothing(self):
         # calling disconnect does nothing.
         input = StringIO()
@@ -556,6 +714,28 @@
         request.finished_reading()
         self.assertRaises(errors.ReadingCompleted, request.read_bytes, None)
 
+    def test_reset(self):
+        server_sock, client_sock = portable_socket_pair()
+        # TODO: Use SmartClientAlreadyConnectedSocketMedium for the versions of
+        #       bzr where it exists.
+        client_medium = medium.SmartTCPClientMedium(None, None, None)
+        client_medium._socket = client_sock
+        client_medium._connected = True
+        req = client_medium.get_request()
+        self.assertRaises(errors.TooManyConcurrentRequests,
+            client_medium.get_request)
+        client_medium.reset()
+        # The stream should be reset, marked as disconnected, though ready for
+        # us to make a new request
+        self.assertFalse(client_medium._connected)
+        self.assertIs(None, client_medium._socket)
+        try:
+            self.assertEqual('', client_sock.recv(1))
+        except socket.error, e:
+            if e.errno not in (errno.EBADF,):
+                raise
+        req = client_medium.get_request()
+
 
 class RemoteTransportTests(TestCaseWithSmartMedium):
 
@@ -609,20 +789,6 @@
         super(TestSmartServerStreamMedium, self).setUp()
         self._captureVar('BZR_NO_SMART_VFS', None)
 
-    def portable_socket_pair(self):
-        """Return a pair of TCP sockets connected to each other.
-
-        Unlike socket.socketpair, this should work on Windows.
-        """
-        listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        listen_sock.bind(('127.0.0.1', 0))
-        listen_sock.listen(1)
-        client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        client_sock.connect(listen_sock.getsockname())
-        server_sock, addr = listen_sock.accept()
-        listen_sock.close()
-        return server_sock, client_sock
-
     def test_smart_query_version(self):
         """Feed a canned query version to a server"""
         # wire-to-wire, using the whole stack
@@ -687,7 +853,7 @@
 
     def test_socket_stream_with_bulk_data(self):
         sample_request_bytes = 'command\n9\nbulk datadone\n'
-        server_sock, client_sock = self.portable_socket_pair()
+        server_sock, client_sock = portable_socket_pair()
         server = medium.SmartServerSocketStreamMedium(
             server_sock, None)
         sample_protocol = SampleRequest(expected_bytes=sample_request_bytes)
@@ -706,7 +872,7 @@
         self.assertTrue(server.finished)
 
     def test_socket_stream_shutdown_detection(self):
-        server_sock, client_sock = self.portable_socket_pair()
+        server_sock, client_sock = portable_socket_pair()
         client_sock.close()
         server = medium.SmartServerSocketStreamMedium(
             server_sock, None)
@@ -726,7 +892,7 @@
         rest_of_request_bytes = 'lo\n'
         expected_response = (
             protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n')
-        server_sock, client_sock = self.portable_socket_pair()
+        server_sock, client_sock = portable_socket_pair()
         server = medium.SmartServerSocketStreamMedium(
             server_sock, None)
         client_sock.sendall(incomplete_request_bytes)
@@ -802,7 +968,7 @@
         # _serve_one_request should still process both of them as if they had
         # been received separately.
         sample_request_bytes = 'command\n'
-        server_sock, client_sock = self.portable_socket_pair()
+        server_sock, client_sock = portable_socket_pair()
         server = medium.SmartServerSocketStreamMedium(
             server_sock, None)
         first_protocol = SampleRequest(expected_bytes=sample_request_bytes)
@@ -839,7 +1005,7 @@
         self.assertTrue(server.finished)
 
     def test_socket_stream_error_handling(self):
-        server_sock, client_sock = self.portable_socket_pair()
+        server_sock, client_sock = portable_socket_pair()
         server = medium.SmartServerSocketStreamMedium(
             server_sock, None)
         fake_protocol = ErrorRaisingProtocol(Exception('boom'))
@@ -860,7 +1026,7 @@
         self.assertEqual('', from_server.getvalue())
 
     def test_socket_stream_keyboard_interrupt_handling(self):
-        server_sock, client_sock = self.portable_socket_pair()
+        server_sock, client_sock = portable_socket_pair()
         server = medium.SmartServerSocketStreamMedium(
             server_sock, None)
         fake_protocol = ErrorRaisingProtocol(KeyboardInterrupt('boom'))
@@ -877,7 +1043,7 @@
         return server._build_protocol()
 
     def build_protocol_socket(self, bytes):
-        server_sock, client_sock = self.portable_socket_pair()
+        server_sock, client_sock = portable_socket_pair()
         server = medium.SmartServerSocketStreamMedium(
             server_sock, None)
         client_sock.sendall(bytes)
@@ -3214,6 +3380,114 @@
         # encoder.
 
 
+class Test_SmartClientRequest(tests.TestCase):
+
+    def make_client_with_failing_medium(self, fail_at_write=True):
+        response = StringIO()
+        output = StringIO()
+        vendor = FirstRejectedStringIOSSHVendor(response, output,
+                    fail_at_write=fail_at_write)
+        client_medium = medium.SmartSSHClientMedium(
+            'a host', 'a port', 'a user', 'a pass', 'base', vendor,
+            'bzr')
+        smart_client = client._SmartClient(client_medium, headers={})
+        return output, vendor, smart_client
+
+    def test__send_no_retry_pipes(self):
+        client_read, server_write = create_file_pipes()
+        server_read, client_write = create_file_pipes()
+        client_medium = medium.SmartSimplePipesClientMedium(client_read,
+            client_write, base='/')
+        smart_client = client._SmartClient(client_medium)
+        smart_request = client._SmartClientRequest(smart_client,
+            'hello', ())
+        # Close the server side
+        server_read.close()
+        encoder, response_handler = smart_request._construct_protocol(3)
+        self.assertRaises(errors.ConnectionReset,
+            smart_request._send_no_retry, encoder)
+
+    def test__send_read_response_sockets(self):
+        listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        listen_sock.bind(('127.0.0.1', 0))
+        listen_sock.listen(1)
+        host, port = listen_sock.getsockname()
+        client_medium = medium.SmartTCPClientMedium(host, port, '/')
+        client_medium._ensure_connection()
+        smart_client = client._SmartClient(client_medium)
+        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
+        # Accept the connection, but don't actually talk to the client.
+        server_sock, _ = listen_sock.accept()
+        server_sock.close()
+        # Sockets buffer and don't really notice that the server has closed the
+        # connection until we try to read again.
+        handler = smart_request._send(3)
+        self.assertRaises(errors.ConnectionReset,
+            handler.read_response_tuple, expect_body=False)
+
+    def test__send_retries_on_write(self):
+        output, vendor, smart_client = self.make_client_with_failing_medium()
+        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
+        handler = smart_request._send(3)
+        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
+                         '\x00\x00\x00\x02de'   # empty headers
+                         's\x00\x00\x00\tl5:helloee',
+                         output.getvalue())
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+             ('close',),
+             ('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+            ],
+            vendor.calls)
+
+    def test__send_doesnt_retry_read_failure(self):
+        output, vendor, smart_client = self.make_client_with_failing_medium(
+            fail_at_write=False)
+        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
+        handler = smart_request._send(3)
+        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
+                         '\x00\x00\x00\x02de'   # empty headers
+                         's\x00\x00\x00\tl5:helloee',
+                         output.getvalue())
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+            ],
+            vendor.calls)
+        self.assertRaises(errors.ConnectionReset, handler.read_response_tuple)
+
+    def test__send_doesnt_retry_body_stream(self):
+        # We don't know how much of body_stream would get iterated as part of
+        # _send before it failed to actually send the request, so we
+        # just always fail in this condition.
+        output, vendor, smart_client = self.make_client_with_failing_medium()
+        smart_request = client._SmartClientRequest(smart_client, 'hello', (),
+            body_stream=['a', 'b'])
+        self.assertRaises(errors.ConnectionReset, smart_request._send, 3)
+        # We got one connect, but it fails, so we disconnect, but we don't
+        # retry it
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+             ('close',),
+            ],
+            vendor.calls)
+
+    def test__send_disabled_retry(self):
+        debug.debug_flags.add('noretry')
+        output, vendor, smart_client = self.make_client_with_failing_medium()
+        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
+        self.assertRaises(errors.ConnectionReset, smart_request._send, 3)
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+             ('close',),
+            ],
+            vendor.calls)
+
+
 class LengthPrefixedBodyDecoder(tests.TestCase):
 
     # XXX: TODO: make accept_reading_trailer invoke translate_response or