Merge pull request #3195 from maartenbreddels/fix_out_of_sync

Kernel is the source of truth: support multiple clients and fixes out of sync bug

Author: SylvainCorlay

packages/base/src/widget.ts

@@ -115,6 +115,9 @@ export class WidgetModel extends Backbone.Model {
     attributes: Backbone.ObjectHash,
     options: IBackboneModelOptions
   ): void {
+    this.expectedEchoMsgIds = {};
+    this.attrsToUpdate = new Set<string>();
+
     super.initialize(attributes, options);
 
     // Attributes should be initialized here, since user initialization may depend on it
@@ -227,6 +230,38 @@ export class WidgetModel extends Backbone.Model {
             const buffer_paths = data.buffer_paths || [];
             const buffers = msg.buffers || [];
             utils.put_buffers(state, buffer_paths, buffers);
+            if (msg.parent_header && data.echo) {
+              const msgId = (msg.parent_header as any).msg_id;
+              // we may have echos coming from other clients, we only care about
+              // dropping echos for which we expected a reply
+              const expectedEcho = data.echo.filter((attrName: string) =>
+                Object.keys(this.expectedEchoMsgIds).includes(attrName)
+              );
+              expectedEcho.forEach((attrName: string) => {
+                // we don't care about the old messages, only the one send with the
+                // last msgId
+                const isOldMessage =
+                  this.expectedEchoMsgIds[attrName] !== msgId;
+                if (isOldMessage) {
+                  // get rid of old updates
+                  delete state[attrName];
+                } else {
+                  // we got our confirmation, from now on we accept everything
+                  delete this.expectedEchoMsgIds[attrName];
+                  // except, we plan to send out a new state for this soon, so we will
+                  // also ignore the update for this property
+                  if (
+                    this._msg_buffer !== null &&
+                    Object.prototype.hasOwnProperty.call(
+                      this._msg_buffer,
+                      attrName
+                    )
+                  ) {
+                    delete state[attrName];
+                  }
+                }
+              });
+            }
             return (this.constructor as typeof WidgetModel)._deserialize_state(
               state,
               this.widget_manager
@@ -300,7 +335,11 @@ export class WidgetModel extends Backbone.Model {
         this._pending_msgs--;
         // Send buffer if one is waiting and we are below the throttle.
         if (this._msg_buffer !== null && this._pending_msgs < 1) {
-          this.send_sync_message(this._msg_buffer, this._msg_buffer_callbacks);
+          const msgId = this.send_sync_message(
+            this._msg_buffer,
+            this._msg_buffer_callbacks
+          );
+          this.rememberLastUpdateFor(msgId);
           this._msg_buffer = null;
           this._msg_buffer_callbacks = null;
         }
@@ -415,6 +454,10 @@ export class WidgetModel extends Backbone.Model {
       }
     }
 
+    Object.keys(attrs).forEach((attrName: string) => {
+      this.attrsToUpdate.add(attrName);
+    });
+
     const msgState = this.serialize(attrs);
 
     if (Object.keys(msgState).length > 0) {
@@ -444,7 +487,8 @@ export class WidgetModel extends Backbone.Model {
       } else {
         // We haven't exceeded the throttle, send the message like
         // normal.
-        this.send_sync_message(attrs, callbacks);
+        const msgId = this.send_sync_message(attrs, callbacks);
+        this.rememberLastUpdateFor(msgId);
         // Since the comm is a one-way communication, assume the message
         // arrived and was processed successfully.
         // Don't call options.success since we don't have a model back from
@@ -453,6 +497,12 @@ export class WidgetModel extends Backbone.Model {
       }
     }
   }
+  rememberLastUpdateFor(msgId: string) {
+    [...this.attrsToUpdate].forEach((attrName) => {
+      this.expectedEchoMsgIds[attrName] = msgId;
+    });
+    this.attrsToUpdate = new Set<string>();
+  }
 
   /**
    * Serialize widget state.
@@ -488,9 +538,9 @@ export class WidgetModel extends Backbone.Model {
   /**
    * Send a sync message to the kernel.
    */
-  send_sync_message(state: JSONObject, callbacks: any = {}): void {
+  send_sync_message(state: JSONObject, callbacks: any = {}): string {
     if (!this.comm) {
-      return;
+      return '';
     }
     try {
       callbacks.iopub = callbacks.iopub || {};
@@ -504,7 +554,7 @@ export class WidgetModel extends Backbone.Model {
 
       // split out the binary buffers
       const split = utils.remove_buffers(state);
-      this.comm.send(
+      const msgId = this.comm.send(
         {
           method: 'update',
           state: split.state,
@@ -515,9 +565,11 @@ export class WidgetModel extends Backbone.Model {
         split.buffers
       );
       this._pending_msgs++;
+      return msgId;
     } catch (e) {
       console.error('Could not send widget sync message', e);
     }
+    return '';
   }
 
   /**
@@ -624,6 +676,12 @@ export class WidgetModel extends Backbone.Model {
   private _msg_buffer: any;
   private _msg_buffer_callbacks: any;
   private _pending_msgs: number;
+  // keep track of the msg id for each attr for updates we send out so
+  // that we can ignore old messages that we send in order to avoid
+  // 'drunken' sliders going back and forward
+  private expectedEchoMsgIds: any;
+  // because we don't know the attrs in _handle_status, we keep track of what we will send
+  private attrsToUpdate: Set<string>;
 }
 
 export class DOMWidgetModel extends WidgetModel {

python/ipywidgets/ipywidgets/widgets/tests/test_set_state.py

@@ -80,7 +80,7 @@ def test_set_state_simple():
         c=[False, True, False],
     ))
 
-    assert w.comm.messages == []
+    assert len(w.comm.messages) == 1
 
 
 def test_set_state_transformer():
@@ -94,7 +94,8 @@ def test_set_state_transformer():
         data=dict(
             buffer_paths=[],
             method='update',
-            state=dict(d=[False, True, False])
+            state=dict(d=[False, True, False]),
+            echo=['d'],
         )))]
 
 
@@ -105,7 +106,7 @@ def test_set_state_data():
         a=True,
         d={'data': data},
     ))
-    assert w.comm.messages == []
+    assert len(w.comm.messages) == 1
 
 
 def test_set_state_data_truncate():
@@ -122,9 +123,10 @@ def test_set_state_data_truncate():
     buffers = msg[1].pop('buffers')
     assert msg == ((), dict(
         data=dict(
-            buffer_paths=[['d', 'data']],
             method='update',
-            state=dict(d={})
+            state=dict(d={}, a=True),
+            buffer_paths=[['d', 'data']],
+            echo=['a', 'd'],
         )))
 
     # Sanity:
@@ -144,8 +146,8 @@ def test_set_state_numbers_int():
         i = 3,
         ci = 4,
     ))
-    # Ensure no update message gets produced
-    assert len(w.comm.messages) == 0
+    # Ensure one update message gets produced
+    assert len(w.comm.messages) == 1
 
 
 def test_set_state_numbers_float():
@@ -156,8 +158,8 @@ def test_set_state_numbers_float():
         cf = 2.0,
         ci = 4.0
     ))
-    # Ensure no update message gets produced
-    assert len(w.comm.messages) == 0
+    # Ensure one update message gets produced
+    assert len(w.comm.messages) == 1
 
 
 def test_set_state_float_to_float():
@@ -167,8 +169,8 @@ def test_set_state_float_to_float():
         f = 1.2,
         cf = 2.6,
     ))
-    # Ensure no update message gets produced
-    assert len(w.comm.messages) == 0
+    # Ensure one message gets produced
+    assert len(w.comm.messages) == 1
 
 
 def test_set_state_cint_to_float():
@@ -235,6 +237,7 @@ def _propagate_value(self, change):
     # this mimics a value coming from the front end
     widget.set_state({'value': 42})
     assert widget.value == 42
+    assert widget.stop is True
 
     # we expect no new state to be sent
     calls = []
@@ -263,8 +266,96 @@ def _propagate_value(self, change):
     assert widget.other == 11
 
     # we expect only single state to be sent, i.e. the {'value': 42.0} state
-    msg = {'method': 'update', 'state': {'value': 2.0, 'other': 11.0}, 'buffer_paths': []}
+    msg = {'method': 'update', 'state': {'value': 2.0, 'other': 11.0}, 'buffer_paths': [], 'echo': ['value']}
     call42 = mock.call(msg, buffers=[])
 
     calls = [call42]
     widget._send.assert_has_calls(calls)
+
+
+
+def test_echo():
+    # we always echo values back to the frontend
+    class ValueWidget(Widget):
+        value = Float().tag(sync=True)
+
+    widget = ValueWidget(value=1)
+    assert widget.value == 1
+
+    widget._send = mock.MagicMock()
+    # this mimics a value coming from the front end
+    widget.set_state({'value': 42})
+    assert widget.value == 42
+
+    # we expect this to be echoed
+    msg = {'method': 'update', 'state': {'value': 42.0}, 'buffer_paths': [], 'echo': ['value']}
+    call42 = mock.call(msg, buffers=[])
+
+    calls = [call42]
+    widget._send.assert_has_calls(calls)
+
+
+def test_echo_single():
+    # we always echo multiple changes back in 1 update
+    class ValueWidget(Widget):
+        value = Float().tag(sync=True)
+        square = Float().tag(sync=True)
+        @observe('value')
+        def _square(self, change):
+            self.square = self.value**2
+
+    widget = ValueWidget(value=1)
+    assert widget.value == 1
+
+    widget._send = mock.MagicMock()
+    # this mimics a value coming from the front end
+    widget._handle_msg({
+        'content': {
+            'data': {
+                'method': 'update',
+                'state': {
+                    'value': 8,
+                }
+            }
+        }
+    })
+    assert widget.value == 8
+    assert widget.square == 64
+
+    # we expect this to be echoed
+    # note that only value is echoed, not square
+    msg = {'method': 'update', 'state': {'square': 64, 'value': 8.0}, 'buffer_paths': [], 'echo': ['value']}
+    call = mock.call(msg, buffers=[])
+
+    calls = [call]
+    widget._send.assert_has_calls(calls)
+
+
+def test_no_echo():
+    # in cases where values coming fromt the frontend are 'heavy', we might want to opt out
+    class ValueWidget(Widget):
+        value = Float().tag(sync=True, no_echo=True)
+
+    widget = ValueWidget(value=1)
+    assert widget.value == 1
+
+    widget._send = mock.MagicMock()
+    # this mimics a value coming from the front end
+    widget._handle_msg({
+        'content': {
+            'data': {
+                'method': 'update',
+                'state': {
+                    'value': 42,
+                }
+            }
+        }
+    })
+    assert widget.value == 42
+
+    # widget._send.assert_not_called(calls)
+    widget._send.assert_not_called()
+
+    # a regular set should sync to the frontend
+    widget.value = 43
+    widget._send.assert_has_calls([mock.call({'method': 'update', 'state': {'value': 43.0}, 'buffer_paths': [], 'echo': ['value']}, buffers=[])])