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

[John A Meinel]

Bug #819604, allow reconnecting with a body stream as long as we haven't touched the stream yet.

This layers on top of my 'client reconnect' patch for the case where we have a body stream.

1) It adds a flush just before we process the first entry of the body stream. This adds an extra write call to 'bzr push'. However, we write after every entry in the stream, and the first entry of the stream is actually very small. (bzrlib.repository._stream_to_byte_stream() the first object that gets flushed is 'pack_writer.begin()', which is just the pack header.)
I don't imagine writing out the ProtocolThree headers will be adding tons of real world overhead.

2) It tracks once we start consuming body_stream, so that we don't do the auto-retry after that point.

I felt it was clear and clean enough to do it this way. We could arguably do a little bit better higher up the stack, but not much, as I described in the bug. Specifically, RemoteRepository.insert_stream is the only caller that passes a body_stream. But it, itself, only gets a stream of records to pass. So if we've started passing records, we can't retry anyway. So the retry logic needs to go all the way up to Fetch code.

[Andrew Bennetts (spiv)]

[Just a comment, not a full review]

> 1) It adds a flush just before we process the first entry of the body
> stream. This adds an extra write call to 'bzr push'. However, we write
> after every entry in the stream, and the first entry of the stream is
> actually very small. (bzrlib.repository._stream_to_byte_stream() the
> first object that gets flushed is 'pack_writer.begin()', which is just
> the pack header.)
> I don't imagine writing out the ProtocolThree headers will be adding
> tons of real world overhead.

You may be right, but please check. Use the netem stuff under linux to
add latency to loopback (I like 500ms), it's documented on the wiki, and
try some pushes with and without.

> 2) It tracks once we start consuming body_stream, so that we don't do
> the auto-retry after that point.
>
> I felt it was clear and clean enough to do it this way. We could

The layering here sounds reasonable to me.

-Andrew.

[John A Meinel (jameinel)]

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

On 10/10/2011 01:37 AM, Andrew Bennetts wrote:
> [Just a comment, not a full review]
>
>> 1) It adds a flush just before we process the first entry of the body
>> stream. This adds an extra write call to 'bzr push'. However, we write
>> after every entry in the stream, and the first entry of the stream is
>> actually very small. (bzrlib.repository._stream_to_byte_stream() the
>> first object that gets flushed is 'pack_writer.begin()', which is just
>> the pack header.)
>> I don't imagine writing out the ProtocolThree headers will be adding
>> tons of real world overhead.
>
> You may be right, but please check. Use the netem stuff under linux to
> add latency to loopback (I like 500ms), it's documented on the wiki, and
> try some pushes with and without.

Short version: This can cause us to send an extra packet, but we don't
always.

I plugged in netem, adding a 500ms delay on the loopback.

I then create a shared repository, branch into it loggerhead -r -10, and
then push to it from the client. I added debugging code that says
'flushing' whenever we hit that '.flush()'[1] line.

bzr.dev 21.627 21.655 21.645
flush() 22.587 22.575 21.656 21.640

It looks like there are 2 tiers of times possible, one with the extra
round trip, and one without, but the extra round trip is not guaranteed.
(Maybe it depends on whether we get a context switch between the
.flush() call and consuming the first hunk from the iterator.)

I ran 3-4 times before I started this list, and all of them were at 21.6
tier, but the next 6 runs were about even between 21.6 and 22.5.

If I change it so that we just have an empty shared repository, and we
push all of loggerhead's history, I get:

bzr.dev 54.682 54.713 54.742 54.683
flush() 55.052 55.083 55.403 55.043

What is strange is that it is consistently slower for a 'write
everything' call, but it isn't even 500ms slower.

Note that without netem slowing things down, the full push times are:

bzr.dev 3.923 3.975 3.945
flush() 3.974 3.954 3.967

So certainly well within any noise margin.

Note that 'flush()' is really just saying 'actually write this to the
socket/pipe/etc', it isn't actually requesting a round trip be made.
Though I think we might have NODELAY set, if you look at the code, it is
going to be writing the actual header/content bytes to the socket in a
millisecond or so.

Arguably during the time we are sending the bulk data stream, we should
be setting NODELAY off.

>
>> 2) It tracks once we start consuming body_stream, so that we don't do
>> the auto-retry after that point.
>>
>> I felt it was clear and clean enough to do it this way. We could
>
> The layering here sounds reasonable to me.
>
> -Andrew.
>
>

We could leave in the check for body_stream_started without adding in
the flush(), though the chances of triggering it before the stream are
virtually nil because we buffer up to 1MB of data before auto-flushing.

[1] I actually see 'flushing' display 2 times. Probably that is the
'probe' insert? (The empty one to determine the remote side supports the
api.)

John
=:->
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.11 (GNU/Linux)
Comment: Using ...

[John A Meinel (jameinel)]

On 10/10/2011 01:37 AM, Andrew Bennetts wrote:
> [Just a comment, not a full review]
>
>> 1) It adds a flush just before we process the first entry of the body
>> stream. This adds an extra write call to 'bzr push'. However, we write
>> after every entry in the stream, and the first entry of the stream is
>> actually very small. (bzrlib.repository._stream_to_byte_stream() the
>> first object that gets flushed is 'pack_writer.begin()', which is just
>> the pack header.)
>> I don't imagine writing out the ProtocolThree headers will be adding
>> tons of real world overhead.
>
> You may be right, but please check. Use the netem stuff under linux to
> add latency to loopback (I like 500ms), it's documented on the wiki, and
> try some pushes with and without.

Short version: This can cause us to send an extra packet, but we don't
always.

I plugged in netem, adding a 500ms delay on the loopback.

I then create a shared repository, branch into it loggerhead -r -10, and
then push to it from the client. I added debugging code that says
'flushing' whenever we hit that '.flush()'[1] line.

bzr.dev 21.627 21.655 21.645
flush() 22.587 22.575 21.656 21.640

It looks like there are 2 tiers of times possible, one with the extra
round trip, and one without, but the extra round trip is not guaranteed.
(Maybe it depends on whether we get a context switch between the
.flush() call and consuming the first hunk from the iterator.)

I ran 3-4 times before I started this list, and all of them were at 21.6
tier, but the next 6 runs were about even between 21.6 and 22.5.

If I change it so that we just have an empty shared repository, and we
push all of loggerhead's history, I get:

bzr.dev 54.682 54.713 54.742 54.683
flush() 55.052 55.083 55.403 55.043

What is strange is that it is consistently slower for a 'write
everything' call, but it isn't even 500ms slower.

Note that without netem slowing things down, the full push times are:

bzr.dev 3.923 3.975 3.945
flush() 3.974 3.954 3.967

So certainly well within any noise margin.

Note that 'flush()' is really just saying 'actually write this to the
socket/pipe/etc', it isn't actually requesting a round trip be made.
Though I think we might have NODELAY set, if you look at the code, it is
going to be writing the actual header/content bytes to the socket in a
millisecond or so.

Arguably during the time we are sending the bulk data stream, we should
be setting NODELAY off.

>
>> 2) It tracks once we start consuming body_stream, so that we don't do
>> the auto-retry after that point.
>>
>> I felt it was clear and clean enough to do it this way. We could
>
> The layering here sounds reasonable to me.
>
> -Andrew.
>
>

We could leave in the check for body_stream_started without adding in
the flush(), though the chances of triggering it before the stream are
virtually nil because we buffer up to 1MB of data before auto-flushing.

[1] I actually see 'flushing' display 2 times. Probably that is the
'probe' insert? (The empty one to determine the remote side supports the
api.)

John
=:->

=== modified file 'bzrlib/smart/client.py'
--- bzrlib/smart/client.py	2011-10-10 12:59:12 +0000
+++ bzrlib/smart/client.py	2011-10-10 12:59:12 +0000
@@ -239,9 +239,13 @@
             # 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):
+                or (self.body_stream is not None
+                    and encoder.body_stream_started)):
                 # We can't restart a body_stream that has been partially
                 # consumed, so we don't retry.
+                # Note: We don't have to worry about
+                #   SmartClientRequestProtocolOne or Two, because they don't
+                #   support client-side body streams.
                 raise
             trace.warning('ConnectionReset calling %r, retrying'
                           % (self.method,))
 
=== modified file 'bzrlib/smart/medium.py'
--- bzrlib/smart/medium.py	2011-10-10 12:59:12 +0000
+++ bzrlib/smart/medium.py	2011-10-10 12:59:12 +0000
@@ -739,8 +739,7 @@
         except IOError, e:
             if e.errno in (errno.EINVAL, errno.EPIPE):
                 raise errors.ConnectionReset(
-                    "Error trying to write to subprocess:\n%s"
-                    % (e,))
+                    "Error trying to write to subprocess:\n%s" % (e,))
             raise
         self._report_activity(len(bytes), 'write')
 
 
=== modified file 'bzrlib/smart/protocol.py'
--- bzrlib/smart/protocol.py	2011-10-10 12:59:12 +0000
+++ bzrlib/smart/protocol.py	2011-10-10 12:59:12 +0000
@@ -1282,6 +1282,7 @@
         _ProtocolThreeEncoder.__init__(self, medium_request.accept_bytes)
         self._medium_request = medium_request
         self._headers = {}
+        self.body_stream_started = None
 
     def set_headers(self, headers):
         self._headers = headers.copy()
@@ -1347,6 +1348,7 @@
             if path is not None:
                 mutter('                  (to %s)', path)
             self._request_start_time = osutils.timer_func()
+        self.body_stream_started = False
         self._write_protocol_version()
         self._write_headers(self._headers)
         self._write_structure(args)
@@ -1354,6 +1356,9 @@
         #       have finished sending the stream.  We would notice at the end
         #       anyway, but if the medium can deliver it early then it's good
         #       to short-circuit the whole request...
+        # Provoke any ConnectionReset failures before we start the body stream.
+        self.flush()
+        self.body_stream_started = True
         for exc_info, part in _iter_with_errors(stream):
             if exc_info is not None:
                 # Iterating the stream failed.  Cleanly abort the request.
 
=== modified file 'bzrlib/tests/test_smart_transport.py'
--- bzrlib/tests/test_smart_transport.py	2011-10-10 12:59:12 +0000
+++ bzrlib/tests/test_smart_transport.py	2011-10-10 12:59:12 +0000
@@ -2944,6 +2944,33 @@
             'e', # end
             output.getvalue())
 
+    def test_records_start_of_body_stream(self):
+        requester, output = self.make_client_encoder_and_output()
+        requester.set_headers({})
+        in_stream = [False]
+        def stream_checker():
+            self.assertTrue(requester.body_stream_started)
+            in_stream[0] = True
+            yield 'content'
+        flush_called = []
+        orig_flush = requester.flush
+        def tracked_flush():
+            flush_called.append(in_stream[0])
+            if in_stream[0]:
+                self.assertTrue(requester.body_stream_started)
+            else:
+                self.assertFalse(requester.body_stream_started)
+            return orig_flush()
+        requester.flush = tracked_flush
+        requester.call_with_body_stream(('one arg',), stream_checker())
+        self.assertEqual(
+            'bzr message 3 (bzr 1.6)\n' # protocol version
+            '\x00\x00\x00\x02de' # headers
+            's\x00\x00\x00\x0bl7:one arge' # args
+            'b\x00\x00\x00\x07content' # body
+            'e', output.getvalue())
+        self.assertEqual([False, True, True], flush_called)
+
 
 class StubMediumRequest(object):
     """A stub medium request that tracks the number of times accept_bytes is
@@ -3458,22 +3485,68 @@
             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.
+    def test__send_request_retries_body_stream_if_not_started(self):
         output, vendor, smart_client = self.make_client_with_failing_medium()
         smart_request = client._SmartClientRequest(smart_client, 'hello', (),
             body_stream=['a', 'b'])
+        response_handler = smart_request._send(3)
+        # We connect, get disconnected, and notice before consuming the stream,
+        # so we try again one time and succeed.
+        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)
+        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
+                         '\x00\x00\x00\x02de'   # empty headers
+                         's\x00\x00\x00\tl5:helloe'
+                         'b\x00\x00\x00\x01a'
+                         'b\x00\x00\x00\x01b'
+                         'e',
+                         output.getvalue())
+
+    def test__send_request_stops_if_body_started(self):
+        # We intentionally use the python StringIO so that we can subclass it.
+        from StringIO import StringIO
+        response = StringIO()
+
+        class FailAfterFirstWrite(StringIO):
+            """Allow one 'write' call to pass, fail the rest"""
+            def __init__(self):
+                StringIO.__init__(self)
+                self._first = True
+
+            def write(self, s):
+                if self._first:
+                    self._first = False
+                    return StringIO.write(self, s)
+                raise IOError(errno.EINVAL, 'invalid file handle')
+        output = FailAfterFirstWrite()
+
+        vendor = FirstRejectedStringIOSSHVendor(response, output,
+            fail_at_write=False)
+        client_medium = medium.SmartSSHClientMedium(
+            'a host', 'a port', 'a user', 'a pass', 'base', vendor,
+            'bzr')
+        smart_client = client._SmartClient(client_medium, headers={})
+        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
+        # We connect, and manage to get to the point that we start consuming
+        # the body stream. The next write fails, so we just stop.
         self.assertEqual(
             [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
               ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
              ('close',),
             ],
             vendor.calls)
+        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
+                         '\x00\x00\x00\x02de'   # empty headers
+                         's\x00\x00\x00\tl5:helloe',
+                         output.getvalue())
 
     def test__send_disabled_retry(self):
         debug.debug_flags.add('noretry')