Merge lp:~pedronis/u1db/remote-sync-exchange-response-order into lp:u1db

[Samuele Pedroni]

reorganize things to send args first in remote sync exchange response

[John A Meinel (jameinel)]

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

On 10/31/2011 03:05 PM, Samuele Pedroni wrote:
> Samuele Pedroni has proposed merging
> lp:~pedronis/u1db/remote-sync-exchange-response-order into lp:u1db
> with lp:~pedronis/u1db/remote-sync-exchange as a prerequisite.
>
> Requested reviews: Ubuntu One hackers (ubuntuone-hackers)
>
> For more details, see:
> https://code.launchpad.net/~pedronis/u1db/remote-sync-exchange-response-order/+merge/80801
>
> reorganize things to send args first in remote sync exchange
> response

I think the overall change here is that 'send_response' doesn't
actually send a complete response back.

Instead, the act of getting 'e\0\0\0' from the client triggers us to
send the end of the response back to the client.

And that feels a bit wrong. If a client sends us an incomplete
request, or if we error out in the middle, the server should still
send the client back a complete record.

I'd be a lot happier with an api that had:

def start_response(...)

def finish_response(...)

def send_response(...):
  self.start_response(...)
  self.finish_response(...)

So that we have a convenience function for RPCs that just a have a
single chunk of response to send. And fuller api so if RPCs want to
control the start and finish of the response.

Doing it based on data from the client doesn't feel right.

I definitely like that the response header is sent before the stream
content.

> + my_gen = self._checkpoint_sync_exchange(from_replica_uid,
> +
> from_replica_generation, +
> last_known_generation) +

^- I'm not sure about the name here.

> === modified file 'u1db/remote/requests.py' ---
> u1db/remote/requests.py 2011-10-31 14:04:45 +0000 +++
> u1db/remote/requests.py 2011-10-31 14:04:45 +0000 @@ -168,11
> +168,16 @@
>
> def handle_end(self): def send_doc(doc_id, doc_rev, doc): -
> self.responder.send_stream_entry((doc_id, doc_rev, doc)) +
> self.responder.stream_entry((doc_id, doc_rev, doc)) +
> new_gen = self.target._checkpoint_sync_exchange( +
> self.from_replica_uid, +
> self.from_replica_generation, +
> self.last_known_generation) +
> self.responder.send_response(other_new_generation=new_gen) new_gen
> = self.target._finish_sync_exchange(self.from_replica_uid,
> self.from_replica_generation, self.last_known_generation,
> send_doc) - self._result(other_new_generation=new_gen) +
> self._close()
>

^- I understand how you are splitting things up here, but it feels a
bit wrong to have RPCSyncExchangeRequest poking at the private apis of
SyncTarget.

I think making it clear how to split up sync exchange into separate
steps is great. We should just make it a bit more part-of-an-api
rather than private collaboration between the classes.

I think if we moved to a state-of-sync object that might be a way to
go. Because it is true that we don't want to expose it as part of the
network RPC.

John
=:->
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.11 (GNU/Linux)
Comment: Using GnuPG with Mozilla - http://enigmail.mozdev.org/

iEYEARECAAYFAk6u/E8ACgkQJdeBCYSNAANalgCeJBmm5KYZkFj1YfvY++qan+wo
8nsAoI/jQD1HPwByEQK1YpECVfRG6SJp
=xjvG
-----END PGP SIGNATURE-----

[Samuele Pedroni (pedronis)]

some of this was addressed in:

https://code.launchpad.net/~pedronis/u1db/remote-sync-exchange-consolidated

> -----BEGIN PGP SIGNED MESSAGE-----
> Hash: SHA1
>
> On 10/31/2011 03:05 PM, Samuele Pedroni wrote:
> > Samuele Pedroni has proposed merging
> > lp:~pedronis/u1db/remote-sync-exchange-response-order into lp:u1db
> > with lp:~pedronis/u1db/remote-sync-exchange as a prerequisite.
> >
> > Requested reviews: Ubuntu One hackers (ubuntuone-hackers)
> >
> > For more details, see:
> > https://code.launchpad.net/~pedronis/u1db/remote-sync-exchange-response-
> order/+merge/80801
> >
> > reorganize things to send args first in remote sync exchange
> > response
>
>
> I think the overall change here is that 'send_response' doesn't
> actually send a complete response back.
>
> Instead, the act of getting 'e\0\0\0' from the client triggers us to
> send the end of the response back to the client.
>
> And that feels a bit wrong. If a client sends us an incomplete
> request, or if we error out in the middle, the server should still
> send the client back a complete record.
>
> I'd be a lot happier with an api that had:
>
> def start_response(...)
>
> def finish_response(...)
>
> def send_response(...):
> self.start_response(...)
> self.finish_response(...)

done this way

>
> So that we have a convenience function for RPCs that just a have a
> single chunk of response to send. And fuller api so if RPCs want to
> control the start and finish of the response.
>
> Doing it based on data from the client doesn't feel right.
>
> I definitely like that the response header is sent before the stream
> content.
>
>
>
> > + my_gen = self._checkpoint_sync_exchange(from_replica_uid,
> > +
> > from_replica_generation, +
> > last_known_generation) +
>
> ^- I'm not sure about the name here.
>
> > === modified file 'u1db/remote/requests.py' ---
> > u1db/remote/requests.py 2011-10-31 14:04:45 +0000 +++
> > u1db/remote/requests.py 2011-10-31 14:04:45 +0000 @@ -168,11
> > +168,16 @@
> >
> > def handle_end(self): def send_doc(doc_id, doc_rev, doc): -
> > self.responder.send_stream_entry((doc_id, doc_rev, doc)) +
> > self.responder.stream_entry((doc_id, doc_rev, doc)) +
> > new_gen = self.target._checkpoint_sync_exchange( +
> > self.from_replica_uid, +
> > self.from_replica_generation, +
> > self.last_known_generation) +
> > self.responder.send_response(other_new_generation=new_gen) new_gen
> > = self.target._finish_sync_exchange(self.from_replica_uid,
> > self.from_replica_generation, self.last_known_generation,
> > send_doc) - self._result(other_new_generation=new_gen) +
> > self._close()
> >
>
>
> ^- I understand how you are splitting things up here, but it feels a
> bit wrong to have RPCSyncExchangeRequest poking at the private apis of
> SyncTarget.
>
> I think making it clear how to split up sync exchange into separate
> steps is great. We should just make it a bit more part-of-an-api
> rather than private collaboration between the classes.
>
> I think if we moved to a state-of-sync object that might be a way to
> go. Because it is true that we don't want to expose it as part of the
> network RPC.

yes we need to extr...

Unmerged revisions

=== modified file 'u1db/backends/__init__.py'
--- u1db/backends/__init__.py	2011-10-31 14:04:45 +0000
+++ u1db/backends/__init__.py	2011-10-31 14:04:45 +0000
@@ -24,6 +24,8 @@
         self._db = db
         self.conflict_ids = set()
         self.seen_ids = set() # not superseded
+        self.my_gen = None
+        self.changed_doc_ids = None
         self._docs_trace = [] # for tests
 
     def _insert_other_doc(self, doc_id, doc_rev, doc):
@@ -51,16 +53,28 @@
                       last_known_generation, take_other_doc):
         for doc_id, doc_rev, doc in docs_info:
             self._insert_other_doc(doc_id, doc_rev, doc)
-        return self._finish_sync_exchange(from_replica_uid,
-                                         from_replica_generation,
-                                         last_known_generation, take_other_doc)
+        my_gen = self._checkpoint_sync_exchange(from_replica_uid,
+                                                from_replica_generation,
+                                                last_known_generation)
+        self._finish_sync_exchange(from_replica_uid,
+                                   from_replica_generation,
+                                   last_known_generation, take_other_doc)
+        return my_gen
+
+    def _checkpoint_sync_exchange(self, from_replica_uid,
+                                  from_replica_generation,
+                                  last_known_generation):
+        my_gen, changed_doc_ids = self._db.whats_changed(last_known_generation)
+        self.my_gen = my_gen
+        self.changed_doc_ids = changed_doc_ids
+        return my_gen
 
     def _finish_sync_exchange(self, from_replica_uid, from_replica_generation,
                          last_known_generation, take_other_doc):
         seen_ids = self.seen_ids
         conflict_ids = self.conflict_ids
-        new_docs = []
-        my_gen, changed_doc_ids = self._db.whats_changed(last_known_generation)
+        my_gen = self.my_gen
+        changed_doc_ids = self.changed_doc_ids
         doc_ids_to_return = [doc_id for doc_id in changed_doc_ids
                              if doc_id not in seen_ids]
         new_docs = self._db.get_docs(doc_ids_to_return,

=== modified file 'u1db/remote/requests.py'
--- u1db/remote/requests.py	2011-10-31 14:04:45 +0000
+++ u1db/remote/requests.py	2011-10-31 14:04:45 +0000
@@ -168,11 +168,16 @@
 
     def handle_end(self):
         def send_doc(doc_id, doc_rev, doc):
-            self.responder.send_stream_entry((doc_id, doc_rev, doc))
+            self.responder.stream_entry((doc_id, doc_rev, doc))
+        new_gen = self.target._checkpoint_sync_exchange(
+                                                  self.from_replica_uid,
+                                                  self.from_replica_generation,
+                                                  self.last_known_generation)
+        self.responder.send_response(other_new_generation=new_gen)
         new_gen = self.target._finish_sync_exchange(self.from_replica_uid,
                                                self.from_replica_generation,
                                                self.last_known_generation,
                                                send_doc)
-        self._result(other_new_generation=new_gen)
+        self._close()
 
 RPCSyncExchange.register()

=== modified file 'u1db/remote/sync_server.py'
--- u1db/remote/sync_server.py	2011-10-31 14:04:45 +0000
+++ u1db/remote/sync_server.py	2011-10-31 14:04:45 +0000
@@ -170,10 +170,11 @@
 
     def received_end(self):
         self._request.handle_end()
-        if not self._responder._sent_response:
+        if not self._responder._started:
             raise errors.BadProtocolStream("Client sent end-of-message,"
                 " but the Request did not generate a response."
                 " for Request: %s" % (self._request,))
+        self._responder._finish_response()
 
 class Responder(object):
     """Encode responses from the server back to the client."""
@@ -184,7 +185,6 @@
             BUFFER_SIZE)
         self._encoder = protocol.ProtocolEncoderV1(self._out_buffer.write)
         self._started = False
-        self._sent_response = False
         self.request_name = ''
 
     def _write_to_client(self, content):
@@ -203,18 +203,17 @@
 
     # have a way to transmit an error
 
-    def send_stream_entry(self, entry):
-        "send stream entry as part of the response."
-        self._start_response()
-        self._encoder.encode_dict('x', entry)
-
     def send_response(self, **kwargs):
         """send/finalize response."""
         self._start_response()
         if kwargs:
             self._encoder.encode_dict('a', kwargs)
+
+    def stream_entry(self, entry):
+        "send stream entry as part of the response."
+        self._start_response()
+        self._encoder.encode_dict('x', entry)
+
+    def _finish_response(self):
         self._encoder.encode_end()
         self._out_buffer.flush()
-        self._sent_response = True
-
-

=== modified file 'u1db/tests/__init__.py'
--- u1db/tests/__init__.py	2011-10-31 14:04:45 +0000
+++ u1db/tests/__init__.py	2011-10-31 14:04:45 +0000
@@ -108,12 +108,16 @@
 
 class ResponderForTests(object):
     """Responder for tests."""
+    _started = False
     _sent_response = False
     status = 'success'
 
     def send_response(self, **kwargs):
+        self._started = True
+        self.kwargs = kwargs
+
+    def _finish_response(self):
         self._sent_response = True
-        self.kwargs = kwargs
 
 
 class TestCaseWithSyncServer(TestCase):

=== modified file 'u1db/tests/test_remote_client.py'
--- u1db/tests/test_remote_client.py	2011-10-31 14:04:45 +0000
+++ u1db/tests/test_remote_client.py	2011-10-31 14:04:45 +0000
@@ -75,20 +75,17 @@
 
     def __init__(self, state, responder):
         super(WithStreamRequest, self).__init__(state, responder)
-        self._args = None
-        self._entries = []
 
     def handle_args(self, **kwargs):
-        self._args = kwargs
+        self.responder.send_response(**kwargs)
 
     def handle_stream_entry(self, entry):
-        self._entries.append(entry)
+        v = entry['outgoing'] * 5
+        self.responder.stream_entry({'incoming': v})
 
     def handle_end(self):
-        for entry in self._entries:
-            v = entry['outgoing'] * 5
-            self.responder.send_stream_entry({'incoming': v})
-        self.responder.send_response(**self._args)
+        pass
+
 
 
 class TestClient(tests.TestCase):
@@ -200,9 +197,9 @@
             'u1db-1\n'
             'h%s{"server_version": "%s", "request": "withstream"}'
             % (struct.pack('!L', 47 + len(_u1db_version)), _u1db_version)
+            + 'a\x00\x00\x00\x0a{"one": 1}'
             + 'x\x00\x00\x00\x10{"incoming": 50}'
             + 'x\x00\x00\x00\x11{"incoming": 100}'
-            + 'a\x00\x00\x00\x0a{"one": 1}'
             + 'e\x00\x00\x00\x00',
             content)
         entries = []

=== modified file 'u1db/tests/test_remote_sync_server.py'
--- u1db/tests/test_remote_sync_server.py	2011-10-31 14:04:45 +0000
+++ u1db/tests/test_remote_sync_server.py	2011-10-31 14:04:45 +0000
@@ -232,22 +232,22 @@
     def test_send_response_after_args(self):
         handler = self.makeStructToRequest()
         handler.received_header({'client_version': '1', 'request': 'arg'})
-        self.assertFalse(handler._responder._sent_response)
+        self.assertFalse(handler._responder._started)
         handler.received_args({'arg': 'value', 'foo': 1})
-        self.assertTrue(handler._responder._sent_response)
+        self.assertTrue(handler._responder._started)
         handler.received_end()
 
     def test_send_response_after_end(self):
         handler = self.makeStructToRequest()
         handler.received_header({'client_version': '1', 'request': 'end'})
-        self.assertFalse(handler._responder._sent_response)
+        self.assertFalse(handler._responder._started)
         handler.received_end()
         self.assertTrue(handler._responder._sent_response)
 
     def test_end_no_response(self):
         handler = self.makeStructToRequest()
         handler.received_header({'client_version': '1', 'request': 'arg'})
-        self.assertFalse(handler._responder._sent_response)
+        self.assertFalse(handler._responder._started)
         self.assertRaises(errors.BadProtocolStream,
                           handler.received_end)
 
@@ -279,6 +279,7 @@
         responder = sync_server.Responder(server_sock)
         responder.request_name = 'request'
         responder.send_response(value='success')
+        responder._finish_response()
         self.assertEqual(
             'u1db-1\n'
             'h%s{"server_version": "%s", "request": "request"}'
@@ -291,8 +292,9 @@
         server_sock, client_sock = tests.socket_pair()
         responder = sync_server.Responder(server_sock)
         responder.request_name = 'request'
-        responder.send_stream_entry({'entry': True})
         responder.send_response()
+        responder.stream_entry({'entry': True})
+        responder._finish_response()
         self.assertEqual(
             'u1db-1\n'
             'h%s{"server_version": "%s", "request": "request"}'