From a4056eea7d35ee1d87218316d43618a7731b08c8 Mon Sep 17 00:00:00 2001 From: Keith Bruce Date: Mon, 26 Oct 2026 13:34:08 +0000 Subject: [PATCH] VersionedValue: never refuse or drop an owner's change VersionedValueListener.setFromOwner and VersionedValue.set(T) took a new version and then stored the value as two separate steps. A set on another thread, for example a BitProducerConsumer applying a message from the network, could take a later version in between and be stored first. The earlier set was then refused as out of date: nothing was stored, no listener was told, and for a BitProducerConsumer no event report was sent for the owner's change. In JMRI this can leave an OlcbSensor in a state that differs from the bit and from the bus after quick changes, because OlcbSensor.setKnownState shows the requested state after calling setFromOwner, even when the set was refused. It is the likely cause of intermittent failures of OlcbSensorTest.testSensor. Take the version and store the value as one step, under the lock of the VersionedValue. Listeners are still called after the lock is released. A second, smaller window remains once that is fixed. The listener that sends a BitProducerConsumer's event reports also sets the values received from the network. If such a value is stored after the owner's change but before the listener is told of it, the listener sees a value newer than the owner's and skips the owner's change, so again no report is sent. Add an opt-in mode, VersionedValueListener(parent, true), in which the listener is told of every change made by someone else, with the value of that change, even if it has set a newer value since. Notifications now carry the version and the listener that made the change (VersionedValue.Change) so the listener can tell its own changes apart. BitProducerConsumer uses this mode for its reporting listener. Messages a BitProducerConsumer sends still come back to it and set the value, as before. They come back as they leave the send queue, which is where they take their place among the messages heard from the bus, so applying them keeps this node in step with the order of the messages on the bus. Tests: - three VersionedValue tests force the interleaving (the other set runs after the owner has taken its version, decided by thread state, not by time); each fails without this change; - a test shows a listener in the new mode is told of an overtaken change, and one shows the default mode still skips it; - BitProducerConsumer tests check that a report heard from another node while ours is queued, and our own reports, set the value in bus order. --- .../implementations/BitProducerConsumer.java | 6 +- .../implementations/VersionedValue.java | 100 +++++++-- .../VersionedValueListener.java | 77 +++++-- .../BitProducerConsumerTest.java | 99 +++++++++ .../MockVersionedValueListener.java | 4 + .../implementations/VersionedValueTest.java | 190 ++++++++++++++++++ 6 files changed, 437 insertions(+), 39 deletions(-) diff --git a/src/org/openlcb/implementations/BitProducerConsumer.java b/src/org/openlcb/implementations/BitProducerConsumer.java index 6cc7cb39..278291ff 100644 --- a/src/org/openlcb/implementations/BitProducerConsumer.java +++ b/src/org/openlcb/implementations/BitProducerConsumer.java @@ -64,7 +64,11 @@ public BitProducerConsumer(OlcbInterface iface, EventID eventOn, EventID eventOf this.eventOff = eventOff; this.flags = flags; value = new VersionedValue<>((flags & DEFAULT_TRUE) != 0); - valueListener = new VersionedValueListener(value) { + // Reports every change made by an owner, even one that a value from the network (which + // this listener sets) has overtaken before the report was made. The report is then + // sent after that network message, and when it comes back to us it sets the value to + // match the order of the messages on the bus. + valueListener = new VersionedValueListener(value, true) { @Override public void update(Boolean newValue) { if ((flags & IS_PRODUCER) == 0) return; diff --git a/src/org/openlcb/implementations/VersionedValue.java b/src/org/openlcb/implementations/VersionedValue.java index b3132c7d..777e5984 100644 --- a/src/org/openlcb/implementations/VersionedValue.java +++ b/src/org/openlcb/implementations/VersionedValue.java @@ -14,6 +14,38 @@ public class VersionedValue { private int defaultVersion = DEFAULT_VERSION; private final static Logger log = Logger.getLogger(VersionedValue.class.getCanonicalName()); + /** + * Notification of one change of the value, sent to the property change listeners. Besides + * the old and new value it carries the version the change was made at, and the listener + * that made it. + */ + public static final class Change extends java.beans.PropertyChangeEvent { + private final int version; + private final transient Object origin; + + Change(VersionedValue source, Object oldValue, Object newValue, int version, + Object origin) { + super(source, "updated", oldValue, newValue); + this.version = version; + this.origin = origin; + } + + /** + * @return the version this change was made at. + */ + public int getVersion() { + return version; + } + + /** + * @return the VersionedValueListener that made this change, or null if it was not made + * through a listener. + */ + public Object getOrigin() { + return origin; + } + } + public VersionedValue(T t) { version = DEFAULT_VERSION; nextVersion = DEFAULT_VERSION + 1; @@ -49,14 +81,18 @@ public synchronized boolean isVersionAtDefault() { /** * Sets the current value; possibly calling listeners. Handles versioning internally. + *

+ * The new version is taken and the value stored as one step, so a set made at the same time + * on another thread cannot take a later version, be stored first, and cause this set to be + * refused as out of date. * @param t new value of data stored. */ public void set(T t) { - int version; - synchronized(this) { - version = getNewVersion(); + Change change; + synchronized (this) { + change = store(getNewVersion(), t, false, null); } - set(version, t); + fire(change); } /** @@ -84,28 +120,52 @@ public boolean set(int atVersion, T t) { } private boolean setInternal(int atVersion, T t, boolean forceNotify) { - T old; - boolean updated = false; + Change change; synchronized (this) { if (atVersion <= version) return false; - int oldVersion = version; - version = atVersion; - if (nextVersion <= atVersion) { - nextVersion = atVersion + 1; - } - if (data.equals(t) && oldVersion != defaultVersion && !forceNotify) { - return true; - } - old = data; - if (oldVersion == defaultVersion || forceNotify) { - old = null; - } - data = t; + change = store(atVersion, t, forceNotify, null); } - firePropertyChange("updated", old, t); + fire(change); return true; } + /** + * Stores a value at a version newer than the current one. The caller must hold the lock of + * this object, and must pass the result to {@link #fire} once it has released the lock. + * @param atVersion new version number, greater than the current version + * @param t new value of data stored + * @param forceNotify notify the listeners even if the value has not changed + * @param origin listener making the change, or null + * @return the notification to send, or null if there is none + */ + Change store(int atVersion, T t, boolean forceNotify, Object origin) { + int oldVersion = version; + version = atVersion; + if (nextVersion <= atVersion) { + nextVersion = atVersion + 1; + } + if (data.equals(t) && oldVersion != defaultVersion && !forceNotify) { + return null; + } + T old = data; + if (oldVersion == defaultVersion || forceNotify) { + old = null; + } + data = t; + return new Change(this, old, t, atVersion, origin); + } + + /** + * Sends a notification returned by {@link #store} to the listeners. Must be called without + * holding the lock of this object. + * @param change notification to send, or null for none + */ + void fire(Change change) { + if (change != null) { + pcs.firePropertyChange(change); + } + } + /** * @return data stored at the current version. */ diff --git a/src/org/openlcb/implementations/VersionedValueListener.java b/src/org/openlcb/implementations/VersionedValueListener.java index d4868232..d8869d00 100644 --- a/src/org/openlcb/implementations/VersionedValueListener.java +++ b/src/org/openlcb/implementations/VersionedValueListener.java @@ -11,9 +11,34 @@ public abstract class VersionedValueListener implements PropertyChangeListener { int ownerVersion; protected VersionedValue parent; + private final boolean everyChange; + /** + * Creates a listener that is told the latest value whenever someone else has set a value + * newer than any this listener has set or been told of. + * @param parent value to listen to + */ public VersionedValueListener(VersionedValue parent) { + this(parent, false); + } + + /** + * Creates a listener. + * @param parent value to listen to + * @param everyChange if false, as for {@link #VersionedValueListener(VersionedValue)}, + * update() is called with the latest value whenever someone else has set + * a value newer than any this listener has set or been told of, so a + * change already overtaken by a newer one is not reported. If true, + * update() is called for every change made by someone else, with the + * value of that change, even if this listener has set a newer value + * since. This suits a listener that sends the owners' changes out to the + * network and also sets the values it receives from the network: a value + * received from the network must not stop an owner's earlier change from + * being sent. + */ + public VersionedValueListener(VersionedValue parent, boolean everyChange) { this.parent = parent; + this.everyChange = everyChange; parent.addPropertyChangeListener(this); ownerVersion = parent.getVersion(); } @@ -24,18 +49,26 @@ public void release() { } @Override + @SuppressWarnings("unchecked") public void propertyChange(PropertyChangeEvent propertyChangeEvent) { - int pVersion = parent.getVersion(); - boolean isUpdated = false; - synchronized (this) { - if (pVersion > ownerVersion) { - ownerVersion = pVersion; - isUpdated = true; + if (everyChange && propertyChangeEvent instanceof VersionedValue.Change) { + VersionedValue.Change change = (VersionedValue.Change) propertyChangeEvent; + if (change.getOrigin() != this) { + update((T) change.getNewValue()); } + return; } - if (isUpdated) { - update(parent.getLatestData()); + T data; + synchronized (this) { + synchronized (parent) { + if (parent.version <= ownerVersion) { + return; + } + ownerVersion = parent.version; + data = parent.data; + } } + update(data); } /** @@ -44,12 +77,7 @@ public void propertyChange(PropertyChangeEvent propertyChangeEvent) { * @param t new value */ public void setFromOwner(T t) { - int newVersion; - synchronized (this) { - newVersion = parent.getNewVersion(); - ownerVersion = newVersion; - } - parent.set(newVersion, t); + setFromOwner(t, false); } /** @@ -58,12 +86,25 @@ public void setFromOwner(T t) { * @param t new value */ public void setFromOwnerWithForceNotify(T t) { - int newVersion; + setFromOwner(t, true); + } + + /** + * Takes a new version and stores the value at it as one step, so that a set made at the same + * time on another thread cannot take a later version, be stored first, and cause this set + * to be refused as out of date. + * @param t new value + * @param forceNotify notify the other listeners even if the value has not changed + */ + private void setFromOwner(T t, boolean forceNotify) { + VersionedValue.Change change; synchronized (this) { - newVersion = parent.getNewVersion(); - ownerVersion = newVersion; + synchronized (parent) { + ownerVersion = parent.getNewVersion(); + change = parent.store(ownerVersion, t, forceNotify, this); + } } - parent.setWithForceNotify(newVersion, t); + parent.fire(change); } /** Calls the updater with the latest data. */ diff --git a/test/org/openlcb/implementations/BitProducerConsumerTest.java b/test/org/openlcb/implementations/BitProducerConsumerTest.java index 7b44cc44..e1a9cb83 100644 --- a/test/org/openlcb/implementations/BitProducerConsumerTest.java +++ b/test/org/openlcb/implementations/BitProducerConsumerTest.java @@ -8,7 +8,12 @@ import org.openlcb.EventID; import org.openlcb.NodeID; +import org.openlcb.OlcbInterface; +import org.mockito.InOrder; +import java.util.concurrent.CountDownLatch; + +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -620,6 +625,100 @@ public void testResetToDefault() { ":X194C5333N0504030201000708;"); } + /** + * Holds back the delivery of outgoing messages, both to the bus and to the local + * listeners, until release() is called. This lets a test act while messages are still in + * the send queue. + */ + private static class HeldLoopback implements OlcbInterface.SyncExecutor { + private final CountDownLatch released = new CountDownLatch(1); + + @Override + public void schedule(Runnable r) throws InterruptedException { + released.await(); + r.run(); + } + + void release() { + released.countDown(); + } + } + + /** + * The owner turns the bit on, and while that report is still in the send queue a report + * from another node turning it off is heard. Our report goes on the bus after the other + * node's, so every node watching the bus ends with the bit on, and so must we. + */ + @Test + public void testFollowsBusOrderWhenRemoteReportHeardBeforeOursIsSent() throws Exception { + createWithDefaults(); + MockVersionedValueListener owner = new MockVersionedValueListener<>(pc + .getValue()); + HeldLoopback loopback = new HeldLoopback(); + iface.setLoopbackThread(loopback); + + owner.setFromOwner(true); + sendFrame(":X195B4444N0504030201000709;"); + Assert.assertFalse(pc.getValue().getLatestData()); + + loopback.release(); + expectFrame(":X195B4333N0504030201000708;"); + expectNoFrames(); + + Assert.assertTrue(pc.getValue().getLatestData()); + InOrder order = inOrder(owner.stub); + order.verify(owner.stub).update(false); + order.verify(owner.stub).update(true); + verifyNoMoreInteractions(owner.stub); + } + + /** + * The owner turns the bit on and straight back off while the first report is still in the + * send queue. Both reports are sent, and as each comes back it sets the value it carried, + * so the bit follows the order of the messages on the bus and ends off. + */ + @Test + public void testOwnReportsSetValueInBusOrder() throws Exception { + createWithDefaults(); + MockVersionedValueListener owner = new MockVersionedValueListener<>(pc + .getValue()); + HeldLoopback loopback = new HeldLoopback(); + iface.setLoopbackThread(loopback); + + owner.setFromOwner(true); + owner.setFromOwner(false); + loopback.release(); + expectFrame(":X195B4333N0504030201000708;"); + expectFrame(":X195B4333N0504030201000709;"); + expectNoFrames(); + + Assert.assertFalse(pc.getValue().getLatestData()); + InOrder order = inOrder(owner.stub); + order.verify(owner.stub).update(true); + order.verify(owner.stub).update(false); + verifyNoMoreInteractions(owner.stub); + } + + @Test + public void testOtherLocalObjectSeesEventReports() throws Exception { + createWithDefaults(); + BitProducerConsumer other = new BitProducerConsumer(iface, onEvent, offEvent, + BitProducerConsumer.IS_CONSUMER); + expectFrame(":X194C7333N0504030201000708;"); + expectFrame(":X194C7333N0504030201000709;"); + expectNoFrames(); + + // Messages from one object still reach the other objects on the same interface. + pc.getValue().set(true); + expectFrame(":X195B4333N0504030201000708;"); + Assert.assertTrue(other.getValue().getLatestData()); + + pc.getValue().set(false); + expectFrame(":X195B4333N0504030201000709;"); + Assert.assertFalse(other.getValue().getLatestData()); + other.release(); + } + private void createWithDefaults() { pc = new BitProducerConsumer(iface, onEvent, offEvent, false); expectFrame(":X19547333N0504030201000708;", times(1)); diff --git a/test/org/openlcb/implementations/MockVersionedValueListener.java b/test/org/openlcb/implementations/MockVersionedValueListener.java index dd604410..6801caab 100644 --- a/test/org/openlcb/implementations/MockVersionedValueListener.java +++ b/test/org/openlcb/implementations/MockVersionedValueListener.java @@ -10,6 +10,10 @@ public MockVersionedValueListener(VersionedValue parent) { super(parent); } + public MockVersionedValueListener(VersionedValue parent, boolean everyChange) { + super(parent, everyChange); + } + public class UpdaterStub { public void update(T t) {} } diff --git a/test/org/openlcb/implementations/VersionedValueTest.java b/test/org/openlcb/implementations/VersionedValueTest.java index b5915042..5bb53482 100644 --- a/test/org/openlcb/implementations/VersionedValueTest.java +++ b/test/org/openlcb/implementations/VersionedValueTest.java @@ -2,6 +2,12 @@ import org.junit.*; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + import static org.mockito.Mockito.reset; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoMoreInteractions; @@ -144,4 +150,188 @@ public void testOutOfOrder() throws Exception { reset(l1.stub); } + /** + * A value that runs a hook straight after handing out a version, so that a test can make + * another thread act between the moment a setter takes its version and the moment it + * stores the value. + */ + private static class PausingValue extends VersionedValue { + Runnable afterNextNewVersion = null; + + PausingValue(boolean t) { + super(t); + } + + @Override + public int getNewVersion() { + int newVersion = super.getNewVersion(); + Runnable hook = afterNextNewVersion; + afterNextNewVersion = null; + if (hook != null) { + hook.run(); + } + return newVersion; + } + } + + /** + * Runs a set on another thread and returns once that set has either finished or is waiting + * for the lock of the value. Decides by the thread's state and the lock it waits for, not by + * time. + */ + private static Thread runSetOnOtherThread(VersionedValue value, Runnable set) { + Thread other = new Thread(set, "other setter"); + other.start(); + while (other.isAlive()) { + ThreadInfo info = ManagementFactory.getThreadMXBean().getThreadInfo(other.getId()); + if (info != null && info.getThreadState() == Thread.State.BLOCKED + && info.getLockInfo() != null + && info.getLockInfo().getIdentityHashCode() == System.identityHashCode(value)) { + break; + } + Thread.yield(); + } + return other; + } + + /** + * The owner takes a version for its change, and before storing it another thread sets the + * value from the network. The owner's change must still be stored and notified, and not be + * refused as out of date. The two notifications are sent after the lock is released, so + * their order is not fixed. + * @param ownerSets the owner's set of false; runs on the test thread + */ + private void helperOwnerChangeNotLost(PausingValue value, + java.util.function.Consumer ownerSets, + MockVersionedValueListener network) + throws InterruptedException { + List applied = Collections.synchronizedList(new ArrayList<>()); + value.addPropertyChangeListener(e -> applied.add((Boolean) e.getNewValue())); + Thread[] other = new Thread[1]; + value.afterNextNewVersion = () -> + other[0] = runSetOnOtherThread(value, () -> network.setFromOwner(true)); + ownerSets.accept(false); + other[0].join(); + + List sorted = new ArrayList<>(applied); + Collections.sort(sorted); + Assert.assertEquals("both sets stored and notified", java.util.Arrays.asList(false, true), + sorted); + // The network set took the later version. + Assert.assertTrue(value.getLatestData()); + } + + @Test + public void testOwnerChangeNotLostToConcurrentSet() throws Exception { + PausingValue value = new PausingValue(true); + value.set(true); // leaves the default version + MockVersionedValueListener owner = new MockVersionedValueListener<>(value); + MockVersionedValueListener network = new MockVersionedValueListener<>(value); + helperOwnerChangeNotLost(value, owner::setFromOwner, network); + // The owner is told of the newer value from the network. + verify(owner.stub).update(true); + verifyNoMoreInteractions(owner.stub); + } + + @Test + public void testOwnerForcedChangeNotLostToConcurrentSet() throws Exception { + PausingValue value = new PausingValue(true); + value.set(true); + MockVersionedValueListener owner = new MockVersionedValueListener<>(value); + MockVersionedValueListener network = new MockVersionedValueListener<>(value); + helperOwnerChangeNotLost(value, owner::setFromOwnerWithForceNotify, network); + verify(owner.stub).update(true); + verifyNoMoreInteractions(owner.stub); + } + + @Test + public void testUnownedSetNotLostToConcurrentSet() throws Exception { + PausingValue value = new PausingValue(true); + value.set(true); + MockVersionedValueListener network = new MockVersionedValueListener<>(value); + helperOwnerChangeNotLost(value, value::set, network); + } + + /** + * The owner changes the value, and before the change has reached a listener that applies + * values from the network, that listener sets a newer value from the network. A listener + * created with everyChange must still be told of the owner's change. + * @param everyChange how the network listener is created + * @return the network listener + */ + private MockVersionedValueListener helperOvertakenChange(boolean everyChange) + throws InterruptedException { + VersionedValue value = new VersionedValue<>(true); + value.set(true); // leaves the default version + @SuppressWarnings("unchecked") + MockVersionedValueListener[] owner = new MockVersionedValueListener[1]; + @SuppressWarnings("unchecked") + MockVersionedValueListener[] network = new MockVersionedValueListener[1]; + // Registered first, so it runs before the other listeners are told of the owner's + // change; it lets the network set finish first. + value.addPropertyChangeListener(e -> { + if (((VersionedValue.Change) e).getOrigin() == owner[0]) { + Thread other = new Thread(() -> network[0].setFromOwner(true)); + other.start(); + try { + other.join(); + } catch (InterruptedException ex) { + throw new RuntimeException(ex); + } + } + }); + network[0] = new MockVersionedValueListener<>(value, everyChange); + owner[0] = new MockVersionedValueListener<>(value); + + owner[0].setFromOwner(false); + + Assert.assertTrue(value.getLatestData()); + verify(owner[0].stub).update(true); + verifyNoMoreInteractions(owner[0].stub); + return network[0]; + } + + @Test + public void testEveryChangeListenerToldOfOvertakenChange() throws Exception { + MockVersionedValueListener l = helperOvertakenChange(true); + verify(l.stub).update(false); + verifyNoMoreInteractions(l.stub); + } + + @Test + public void testDefaultListenerSkipsOvertakenChange() throws Exception { + MockVersionedValueListener l = helperOvertakenChange(false); + verifyNoMoreInteractions(l.stub); + } + + @Test + public void testEveryChangeListenerSkipsOwnChanges() throws Exception { + v.set(33); + MockVersionedValueListener l1 = new MockVersionedValueListener<>(v, true); + MockVersionedValueListener l2 = new MockVersionedValueListener<>(v); + + l1.setFromOwner(13); + verify(l2.stub).update(13); + verifyNoMoreInteractions(l1.stub, l2.stub); + reset(l1.stub, l2.stub); + + l2.setFromOwner(21); + verify(l1.stub).update(21); + verifyNoMoreInteractions(l1.stub, l2.stub); + reset(l1.stub, l2.stub); + + // No notification when the value does not change, unless forced. + l2.setFromOwner(21); + verifyNoMoreInteractions(l1.stub, l2.stub); + l2.setFromOwnerWithForceNotify(21); + verify(l1.stub).update(21); + verifyNoMoreInteractions(l1.stub, l2.stub); + reset(l1.stub, l2.stub); + + v.set(5); + verify(l1.stub).update(5); + verify(l2.stub).update(5); + verifyNoMoreInteractions(l1.stub, l2.stub); + } + }