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); + } + }