Merge lp:~kamstrup/dee/shared-model-txn-notify into lp:dee

[Mikkel Kamstrup Erlandsen]

Add two new signals to DeeSharedModel; begin-transaction and end-transaction.

Both signals will pass in the seqnums of the model before- and after the
transaction.

[Michal Hruby (mhr3)]

Looks quite good, although I'd prefer more comments, specifically:

144 + flush_revision_queue (DEE_MODEL (user_data));

Why is this needed now?

236 + /* Chain up to parent class impl */
237 + ((DeeModelIface*) g_type_interface_peek_parent (DEE_MODEL_GET_IFACE(model)))->clear (model);

Why the chainup?

review: Needs Fixing

[Mikkel Kamstrup Erlandsen (kamstrup)]

Pushed with comments

[lp:~kamstrup/dee/shared-model-txn-notify]

337. By Mikkel Kamstrup Erlandsen on 2012-01-06

Add some comments to clarify some of the more obscure logic

[Michal Hruby (mhr3)]

Great!

review: Approve

=== modified file 'src/dee-marshal.list'
--- src/dee-marshal.list	2010-05-06 13:45:44 +0000
+++ src/dee-marshal.list	2012-01-06 12:59:23 +0000
@@ -1,8 +1,12 @@
-# DbusSharedModel::RowsAdded, DbusSharedModel::RowsChanged
+# DeeSharedModel::RowsAdded, DbusSharedModel::RowsChanged
 VOID:BOXED,BOXED,BOXED
 
-# DbusSharedModel::RowsRemoved
+# DeeSharedModel::RowsRemoved
 VOID:BOXED,BOXED
 
-# DbusPeer::bye
+# DeeSharedModel
+VOID:UINT64,UINT64
+
+# DeePeer::bye
 VOID:STRING,STRING
+

=== modified file 'src/dee-shared-model.c'
--- src/dee-shared-model.c	2011-12-16 13:01:10 +0000
+++ src/dee-shared-model.c	2012-01-06 12:59:23 +0000
@@ -85,6 +85,7 @@
   GSList     *connections;
   gchar      *model_path;
   
+  guint64     last_committed_seqnum;
   /* Buffer of DeeSharedModelRevisions that we keep in order to batch
    * our DBus signals. The invariant is that all buffered revisions
    * are of the same type */
@@ -139,12 +140,13 @@
 enum
 {
   /* Public signal */
-  READY,
+  BEGIN_TRANSACTION,
+  END_TRANSACTION,
 
   LAST_SIGNAL
 };
 
-//static guint32 _signals[LAST_SIGNAL] = { 0 };
+static guint32 _signals[LAST_SIGNAL] = { 0 };
 
 /* Forwards */
 static void     on_connection_acquired                 (DeeSharedModel  *self,
@@ -315,7 +317,10 @@
   /* If we don't have anything queued up, just return. It's assumed beyond
    * this point that it is non-empty */
   if (priv->revision_queue == NULL)
-    return 0;
+    {
+      priv->last_committed_seqnum = dee_serializable_model_get_seqnum (self);
+      return 0;
+    }
 
   /* Since we always prepend to the queue we need to reverse it */
   priv->revision_queue = g_slist_reverse (priv->revision_queue);
@@ -326,7 +331,7 @@
    * first element and assume that the last seqnum before this transaction
    * started was the seqnum in the first revision - 1. */
   seqnum_end = ((DeeSharedModelRevision *) priv->revision_queue->data)->seqnum - 1;
-  seqnum_begin = seqnum_end;
+  seqnum_begin = priv->last_committed_seqnum;
 
   g_variant_builder_init (&aav, G_VARIANT_TYPE ("aav"));
   g_variant_builder_init (&au, G_VARIANT_TYPE ("au"));
@@ -415,6 +420,8 @@
    * we constructed the Commit message */
   g_slist_free (priv->revision_queue);
   priv->revision_queue = NULL;
+  
+  priv->last_committed_seqnum = seqnum_end;
 
   return seqnum_end - seqnum_begin; // Very theoretical overflow possible here...
 }
@@ -674,6 +681,42 @@
                                FALSE,
                                G_PARAM_READABLE);
   g_object_class_install_property (obj_class, PROP_SYNCHRONIZED, pspec);
+  
+  /**
+   * DeeSharedModel::begin-transaction:
+   * @model: The shared model the signal is emitted on
+   * @begin_seqnum: The seqnum the model has now
+   * @end_seqnum: The seqnum the model will have after the transaction is applied
+   *
+   * Emitted right before a remote transaction will be committed to the model.
+   */
+  _signals[BEGIN_TRANSACTION] =
+    g_signal_new ("begin-transaction",
+                  DEE_TYPE_SHARED_MODEL,
+                  G_SIGNAL_RUN_LAST,
+                  0,
+                  NULL, NULL,
+                  _dee_marshal_VOID__UINT64_UINT64,
+                  G_TYPE_NONE, 2,
+                  G_TYPE_UINT64, G_TYPE_UINT64);
+
+  /**
+   * DeeSharedModel::end-transaction:
+   * @model: The shared model the signal is emitted on
+   * @begin_seqnum: The seqnum the model had before the transaction was applied
+   * @end_seqnum: The seqnum the model has now
+   *
+   * Emitted right after a remote transaction has been committed to the model.
+   */
+  _signals[END_TRANSACTION] =
+    g_signal_new ("end-transaction",
+                  DEE_TYPE_SHARED_MODEL,
+                  G_SIGNAL_RUN_LAST,
+                  0,
+                  NULL, NULL,
+                  _dee_marshal_VOID__UINT64_UINT64,
+                  G_TYPE_NONE, 2,
+                  G_TYPE_UINT64, G_TYPE_UINT64);
 
   /* Add private data */
   g_type_class_add_private (obj_class, sizeof (DeeSharedModelPrivate));
@@ -689,6 +732,7 @@
   priv->swarm = NULL;
   priv->model_path = NULL;
 
+  priv->last_committed_seqnum = 0;
   priv->revision_queue = NULL;
   priv->revision_queue_timeout_id = 0;
   priv->swarm_leader_handler = 0;
@@ -720,12 +764,20 @@
                          GDBusMethodInvocation *invocation,
                          gpointer               user_data)
 {
-  GVariant  *retval;
+  DeeSharedModelPrivate *priv;
+  GVariant              *retval;
 
   g_return_if_fail (DEE_IS_SHARED_MODEL (user_data));
+  
+  priv = DEE_SHARED_MODEL (user_data)->priv;
 
   if (g_strcmp0 ("Clone", method_name) == 0)
     {
+      /* If we have anything in the rev queue it wont validate against the
+       * seqnum for the cloned model. So flush the rev queue before answering
+       * the Clone call */
+      flush_revision_queue (DEE_MODEL (user_data));
+      
       /* We return a special error if we have no schema. It's legal for the
        * leader to expect the schema from the slaves */
       if (dee_model_get_n_columns (DEE_MODEL (user_data)) == 0)
@@ -931,6 +983,11 @@
    * but in that case we should still consider our selves synchronized */
   if (transaction != NULL)
     {
+      /* Guard against a race where we might inadvertedly have accepted a Commit
+       * before receiving the initial Clone */
+      if (dee_model_get_n_columns (DEE_MODEL (self)) > 0)
+        reset_model (DEE_MODEL (self));
+      
       /* We use the swarm name as sender_name here, because DBus passes us the
       * unique name of the swarm leader here and we want to indicate in the debug
       * messages that the transaction came from the leader */
@@ -1143,9 +1200,11 @@
   if (current_seqnum != 0 &&
       current_seqnum != seqnum_before)
   {
-      g_warning ("Transaction from %s is in the %s. Ignoring transaction.",
+      g_warning ("Transaction from %s is in the %s. Expected seqnum %"G_GUINT64_FORMAT
+                 ", but got %"G_GUINT64_FORMAT". Ignoring transaction.",
                  sender_name,
-                 current_seqnum < seqnum_before ? "future" : "past");
+                 current_seqnum < seqnum_before ? "future" : "past",
+                 current_seqnum, seqnum_before);
       if (dee_shared_model_is_leader (self))
         {
           g_warning ("Invalidating %s", sender_name);
@@ -1173,7 +1232,7 @@
                  sender_name);
       // FIXME cleanup
     }
-  if (n_rows != (seqnum_after - seqnum_before))
+  if (n_rows > (seqnum_after - seqnum_before))
     {
       g_warning ("Commit from %s has illegal seqnum count.",
                  sender_name);
@@ -1185,6 +1244,7 @@
   trace_object (self, "Applying transaction of %i rows", n_rows);
 
   /* Phew. Finally. We're ready to merge the changes */
+  g_signal_emit (self, _signals[BEGIN_TRANSACTION], 0, seqnum_before, seqnum_after);
   priv->suppress_remote_signals = TRUE;
   for (i = 0; i < n_rows; i++) /* Begin outer loop */
     {
@@ -1270,7 +1330,13 @@
   g_variant_unref (au);
   g_variant_unref (ay);
 
+  /* We must manually override the seqnum in case we started off from
+   * zero our selves, but the transaction was a later snapshot */
   dee_serializable_model_set_seqnum (DEE_MODEL (self), seqnum_after);
+  
+  priv->last_committed_seqnum = seqnum_after;
+  
+  g_signal_emit (self, _signals[END_TRANSACTION], 0, seqnum_before, seqnum_after);
 }
 
 static void
@@ -1650,6 +1716,7 @@
   DeeModel              *backend;
   gboolean               was_suppressing;
   guint64                seqnum;
+  guint                  n_rows;
 
   self = DEE_SHARED_MODEL (model);
   priv = self->priv;
@@ -1658,21 +1725,25 @@
 
   was_suppressing = priv->suppress_remote_signals;
   seqnum = dee_serializable_model_get_seqnum (model);
-  if (!was_suppressing && dee_model_get_n_rows (model) > 0)
+  n_rows = dee_model_get_n_rows (model);
+  
+  if (!was_suppressing && n_rows > 0)
     {
+      seqnum += n_rows;
       enqueue_revision (model,
                         CHANGE_TYPE_CLEAR,
                         0,
-                        ++seqnum,
+                        seqnum,
                         NULL);
     }
   /* make sure we don't enqueue lots of CHANGE_TYPE_REMOVE */
   priv->suppress_remote_signals = TRUE;
 
-  /* FIXME: should we set the seqnum after each removed row? */
-  dee_model_clear (backend);
+  /* Chain up to parent class impl. This handles the seqnums for us and the
+   * backend alike. We just hook in before it, really, to player clever
+   * tricks with the revision queue (inserting a CLEAR and not N*REMOVE) */
+  ((DeeModelIface*) g_type_interface_peek_parent (DEE_MODEL_GET_IFACE(model)))->clear (model);
 
-  dee_serializable_model_set_seqnum (model, seqnum);
   priv->suppress_remote_signals = was_suppressing;
 
   g_object_unref (backend);
@@ -1695,9 +1766,6 @@
   guint64                 last_seqnum;
   
   g_return_val_if_fail (DEE_IS_SHARED_MODEL (self), FALSE);
-  
-  trace_object (self, "Serializing %u rows",
-                dee_model_get_n_rows (DEE_MODEL (self)));
 
   _self = DEE_MODEL (self);
   n_columns = dee_model_get_n_columns (DEE_MODEL (self));
@@ -1732,7 +1800,7 @@
 
   /* Collect the seqnums */
   last_seqnum = dee_serializable_model_get_seqnum (_self);
-  tt = g_variant_new ("(tt)", last_seqnum - i, last_seqnum);
+  tt = g_variant_new ("(tt)", last_seqnum - i, last_seqnum);//  FIXME last_committed_seqnum
 
   /* Build the final clone */
   g_variant_builder_init (&clone, G_VARIANT_TYPE ("(sasaavauay(tt))"));
@@ -1743,6 +1811,10 @@
   g_variant_builder_add_value (&clone, g_variant_builder_end (&ay));
   g_variant_builder_add_value (&clone, tt);
 
+  trace_object (self, "Serialized %u rows. "
+                "Seqnum range %"G_GUINT64_FORMAT"-%"G_GUINT64_FORMAT,
+                i, last_seqnum - i, last_seqnum);
+
   return g_variant_builder_end (&clone);
 }
 

=== modified file 'tests/model-helper-add3rows.c'
--- tests/model-helper-add3rows.c	2011-12-14 14:09:58 +0000
+++ tests/model-helper-add3rows.c	2012-01-06 12:59:23 +0000
@@ -25,6 +25,23 @@
 #include <gtx.h>
 #include <dee.h>
 
+guint64 before_begin_seqnum, before_end_seqnum,
+        after_begin_seqnum, after_end_seqnum;
+
+static void
+_begin_txn (DeeSharedModel *model, guint64 begin_seqnum, guint64 end_seqnum)
+{
+  before_begin_seqnum = begin_seqnum;
+  before_end_seqnum = end_seqnum;
+}
+
+static void
+_end_txn (DeeSharedModel *model, guint64 begin_seqnum, guint64 end_seqnum)
+{
+  after_begin_seqnum = begin_seqnum;
+  after_end_seqnum = end_seqnum;
+}
+
 static void
 _row_added (DeeModel *model, DeeModelIter *iter, GSList **rows_so_far)
 {
@@ -48,6 +65,8 @@
   else
     model = dee_shared_model_new_for_peer ((DeePeer*) dee_client_new (argv[1]));
 
+  g_signal_connect (model, "begin-transaction", G_CALLBACK (_begin_txn), NULL);
+  g_signal_connect (model, "end-transaction", G_CALLBACK (_end_txn), NULL);
 
   /* Wait until we find the leader */
   if (gtx_wait_for_signal (G_OBJECT (model), 1000, "notify::synchronized", NULL))
@@ -84,5 +103,10 @@
   gtx_assert_last_unref (model);
   g_slist_free (rows_added);
   
+  g_assert (before_begin_seqnum == after_begin_seqnum);
+  g_assert (before_end_seqnum == after_end_seqnum);
+  g_assert_cmpint (0, ==, (guint) before_begin_seqnum);
+  g_assert_cmpint (3, ==, (guint) before_end_seqnum);
+  
   return 0;
 }

=== modified file 'tests/model-helper-clear3add5.c'
--- tests/model-helper-clear3add5.c	2011-12-14 14:09:58 +0000
+++ tests/model-helper-clear3add5.c	2012-01-06 12:59:23 +0000
@@ -25,6 +25,23 @@
 #include <gtx.h>
 #include <dee.h>
 
+guint64 before_begin_seqnum, before_end_seqnum,
+        after_begin_seqnum, after_end_seqnum;
+
+static void
+_begin_txn (DeeSharedModel *model, guint64 begin_seqnum, guint64 end_seqnum)
+{
+  before_begin_seqnum = begin_seqnum;
+  before_end_seqnum = end_seqnum;
+}
+
+static void
+_end_txn (DeeSharedModel *model, guint64 begin_seqnum, guint64 end_seqnum)
+{
+  after_begin_seqnum = begin_seqnum;
+  after_end_seqnum = end_seqnum;
+}
+
 static void
 _row_added (DeeModel *model, DeeModelIter *iter, GSList **added)
 {
@@ -47,6 +64,8 @@
   else
     model = dee_shared_model_new_for_peer ((DeePeer*) dee_client_new (argv[1]));
 
+  g_signal_connect (model, "begin-transaction", G_CALLBACK (_begin_txn), NULL);
+  g_signal_connect (model, "end-transaction", G_CALLBACK (_end_txn), NULL);
   
   if (gtx_wait_for_signal (G_OBJECT (model), 100000, "notify::synchronized", NULL))
     g_error ("Helper model timed out waiting for 'ready' signal");
@@ -65,6 +84,15 @@
   g_assert (g_slist_length (added) == 5 || g_slist_length (added) == 2);
   g_assert_cmpint (dee_model_get_n_rows (model), ==, 5);
   
+  /* Disregarding the hypothetical optimization mentioned above we need the
+   * correct seqnum */
+  g_assert_cmpint (11, ==, (guint) dee_serializable_model_get_seqnum (model));
+  
+  g_assert (before_begin_seqnum == after_begin_seqnum);
+  g_assert (before_end_seqnum == after_end_seqnum);
+  g_assert_cmpint (3, ==, (guint) before_begin_seqnum);
+  g_assert_cmpint (11, ==, (guint) before_end_seqnum);
+  
   gtx_assert_last_unref (model);
   g_slist_free (added);
   

=== modified file 'tests/model-helper-clear3rows.c'
--- tests/model-helper-clear3rows.c	2011-12-14 14:09:58 +0000
+++ tests/model-helper-clear3rows.c	2012-01-06 12:59:23 +0000
@@ -25,6 +25,23 @@
 #include <gtx.h>
 #include <dee.h>
 
+guint64 before_begin_seqnum, before_end_seqnum,
+        after_begin_seqnum, after_end_seqnum;
+
+static void
+_begin_txn (DeeSharedModel *model, guint64 begin_seqnum, guint64 end_seqnum)
+{
+  before_begin_seqnum = begin_seqnum;
+  before_end_seqnum = end_seqnum;
+}
+
+static void
+_end_txn (DeeSharedModel *model, guint64 begin_seqnum, guint64 end_seqnum)
+{
+  after_begin_seqnum = begin_seqnum;
+  after_end_seqnum = end_seqnum;
+}
+
 static void
 _row_removed (DeeModel *model, DeeModelIter *iter, GSList **removed)
 {
@@ -47,6 +64,8 @@
   else
     model = dee_shared_model_new_for_peer ((DeePeer*) dee_client_new (argv[1]));
 
+  g_signal_connect (model, "begin-transaction", G_CALLBACK (_begin_txn), NULL);
+  g_signal_connect (model, "end-transaction", G_CALLBACK (_end_txn), NULL);
   
   if (gtx_wait_for_signal (G_OBJECT (model), 100000, "notify::synchronized", NULL))
     g_error ("Helper model timed out waiting for 'ready' signal");
@@ -63,6 +82,13 @@
   g_assert_cmpint (g_slist_length (removed), ==, 3);
   g_assert_cmpint (dee_model_get_n_rows(model), ==, 0);
   
+  g_assert_cmpint (6, ==, (guint) dee_serializable_model_get_seqnum (model));
+  
+  g_assert (before_begin_seqnum == after_begin_seqnum);
+  g_assert (before_end_seqnum == after_end_seqnum);
+  g_assert_cmpint (3, ==, (guint) before_begin_seqnum);
+  g_assert_cmpint (6, ==, (guint) before_end_seqnum);
+  
   gtx_assert_last_unref (model);
   g_slist_free (removed);
   

=== modified file 'tests/server-helper-client.c'
--- tests/server-helper-client.c	2012-01-02 15:59:19 +0000
+++ tests/server-helper-client.c	2012-01-06 12:59:23 +0000
@@ -102,6 +102,10 @@
     gtx_yield_main_loop (200);
     if (finished_children == num_clients) break;
   }
+  
+  /* Give a window of opportunity for children to
+   * flush stdout/err before we exit */
+  gtx_yield_main_loop (200);
 
   g_assert_cmpint (num_clients, ==, finished_children);
 

=== modified file 'tests/test-model-seqnums.c'
--- tests/test-model-seqnums.c	2011-12-09 11:47:47 +0000
+++ tests/test-model-seqnums.c	2012-01-06 12:59:23 +0000
@@ -143,4 +143,13 @@
 
   dee_model_clear (fix->model);
   g_assert_cmpint (3, ==, dee_serializable_model_get_seqnum (model));
+  
+  dee_model_append (fix->model, 11);
+  dee_model_append (fix->model, 12);
+  dee_model_append (fix->model, 13);
+  g_assert_cmpint (6, ==, dee_serializable_model_get_seqnum (model));
+  
+  dee_model_clear (fix->model);
+  g_assert_cmpint (9, ==, dee_serializable_model_get_seqnum (model));
 }
+

=== modified file 'vapi/dee-1.0.vapi'
--- vapi/dee-1.0.vapi	2012-01-04 08:32:53 +0000
+++ vapi/dee-1.0.vapi	2012-01-06 12:59:23 +0000
@@ -150,6 +150,8 @@
 		public Dee.Peer peer { get; construct; }
 		[NoAccessorMethod]
 		public bool synchronized { get; }
+		public signal void begin_transaction (uint64 begin_seqnum, uint64 end_seqnum);
+		public signal void end_transaction (uint64 begin_seqnum, uint64 end_seqnum);
 	}
 	[CCode (cheader_filename = "dee.h", type_id = "dee_term_list_get_type ()")]
 	public class TermList : GLib.Object {