swing.Publisher does not always notify all listeners

[scabug]

So here is something that had me scratch my head for quite some time. Sometimes, Reactors would not get notified by a publisher. This didn't happen very often, but often enough to break some functionality.

It turned out that it was caused by the mutable nature of the swing.Publisher#listeners. More precisely, it sometimes happens when a listener, in reaction to an event, decides to unsubscribe from the publisher, i.e., calls deafTo(). This immediately causes the listeners to change state, and therefore may lead to the for comprehension to terminate early, without notifying subsequent listeners.

I think that "if a specific event happens on a publisher, I'm not interested in that publisher anymore" is a valid and not terribly exotic use case for a Reactor, so I consider the current behavior a bug.

I'm attaching a scalatest class which triggers the bug (and also contains one possible solution).

[scabug]

Imported From: https://issues.scala-lang.org/browse/SI-8495?orig=1
Reporter: Christoph Langguth (clangguth)
Affected Versions: 2.10.3
Attachments:

• PublisherBugTest.scala (created on Apr 11, 2014 2:09:44 PM UTC, 2342 bytes)
• PublisherTest.scala (created on Apr 28, 2014 2:43:09 PM UTC, 1638 bytes)

[scabug]

@andy1138 said:
PR for review: #6

[scabug]

Christoph Langguth (clangguth) said:
Thanks Andy!

Looks good so far, just two small comments:

1. I'm not sure if the test description is entirely correct. Right now it's "listeners should not be called after they have been removed from the publisher". But that doesn't really reflect the original problem (and actually, is not what the test is testing). I'll attach a slightly modified version of the test in a minute, which (IMO) addresses both issues.

2. Is the containsItem() method really needed? It's essentially the same as the contains() one, just without the purging part. IOW: couldn't one just use the contains() method, or is there a performance hit? And if so, would it make sense to name it somewhat differently ("containsNoPurge()" or so - I know, not really a good name either...) and/or make it protected, to avoid confusion with the contains() method?

[scabug]

Christoph Langguth (clangguth) said (edited on Apr 28, 2014 2:45:38 PM UTC):
Ok, so here is the attachment I mentioned above.

It's a minimally modified version of andy1138@bc80053

(changed test description and made sure that duplicate notifications are also covered)

[scabug]

@andy1138 said:
Hi Christoph

I've updated the code with your test, much better, thanks

On containsItem, I think that if the item has been removed from 'listeners' then it should not be called, sounds like another bug waiting to happen. I then couldn't determine how long purgeReferences would take and guessed that in this case it wasn't needed so didn't want to call it. As for the name then I also could'nt come up with a good one :-). I did try making it protected but that prevented it being seen by publish(), the whole class is package[swing] protected so at least it's limiting who can use it. Happy to take suggestions of both name and visibility

[scabug]

Christoph Langguth (clangguth) said:
Hi Andy,

thanks again for the quick reaction!

I think the test now covers all cases that could possibly go wrong :-)

Concerning the logic (and therefore the code) of the publish() method, after diving into some of the hairy parts (like the ReferenceQueue and GC internals), here are my thoughts:

• an invocation of the publish() method at timestamp t0 has the semantics "notify every listener that is known at t0, exactly once". Let's call the set of these listeners L(t0).
• if during that invocation, the state of the listeners changes (say, because of a listenTo() or deafTo() call), that is fine. However, that state change can only happen at timestamp tX (where tX > t0). And it will only affect states L(tY) | (tY > tX > t0) at the publisher.
• since the publish() method is only interested in state L(t0) (which is already "captured" by the call to listeners.clone() ), there is actually no need to re-verify the state at the publisher's end of the communication. In fact, it is impossible for the publisher to know (or check) if the listener is still interested in the message. All that the publisher can reasonably do is to know is that the listener was interested at the time that the message had been sent (t0) - and therefore, it should deliver it. If the listener has changed its mind in the mean time, that's ok -- but it's something that only the listener can know, not the publisher. Therefore, I think that the call to contains()/containsItem() in the publish() method is actually not needed.

To summarize it in one sentence: the publish() method must ensure that every listener that was registered at the time of calling the method gets invoked exactly once. Therefore, I think that:

  def publish(e: Event) { for (l <- listeners.clone()) if (l.isDefinedAt(e)) l(e) }

is actually the correct implementation.

Thoughts?

Thanks! :-)
Chris

[scabug]

@andy1138 said:
Hi Chris

I've been distracted by work, and the 2nd part of this, but continuing on ...

Interesting, I think of it the other way around. as a client when I've call deafTo() then I would not expect any new messages from that publisher. eg if I'm writing events to a file then I should be able to do { deafTo(); logger.close() } and not have to worry that the logger would be written to afterwards.

While doing some 'lets find out what others do' I looked at the scala collection publisher (https://github.com/scala/scala/blob/v2.10.3/src/library/scala/collection/mutable/Publisher.scala#L50)

• Why is swing using its own publisher ? , sounds like an upgrade is required. (most of swing is pre scala collections)
• I'm trying to decide if this has the same problem, ie it also needs a .clone() added

Longer term, I think this should be changed to use the collections version of Publisher, what do you think?

Andy

[Sciss]

Perhaps related, I ran into this:

Exception in thread "AWT-EventQueue-0" java.util.ConcurrentModificationException: mutation occurred during iteration
	at scala.collection.mutable.MutationTracker$.checkMutations(MutationTracker.scala:43)
	at scala.collection.mutable.MutationTracker$CheckedIterator.hasNext(MutationTracker.scala:59)
	at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:563)
	at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:561)
	at scala.collection.AbstractIterable.foreach(Iterable.scala:919)
	at scala.swing.Reactions$Impl.apply(Reactions.scala:25)
	at scala.swing.Reactions$Impl.apply(Reactions.scala:19)
	at scala.swing.Publisher.$anonfun$publish$1(Publisher.scala:52)
	at scala.swing.Publisher.$anonfun$publish$1$adapted(Publisher.scala:52)
	at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:563)
	at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:561)
	at scala.swing.SetWrapper.foreach(SetWrapper.scala:20)
	at scala.swing.Publisher.publish(Publisher.scala:52)
	at scala.swing.Publisher.publish$(Publisher.scala:52)
	at scala.swing.Component.publish(Component.scala:48)
	at de.sciss.swingplus.ScrollBar$$anon$2.adjustmentValueChanged(ScrollBar.scala:36)
	at java.desktop/javax.swing.JScrollBar.fireAdjustmentValueChanged(JScrollBar.java:711)
	at java.desktop/javax.swing.JScrollBar$ModelListener.stateChanged(JScrollBar.java:733)
	at java.desktop/javax.swing.DefaultBoundedRangeModel.fireStateChanged(DefaultBoundedRangeModel.java:371)
	at java.desktop/javax.swing.DefaultBoundedRangeModel.setRangeProperties(DefaultBoundedRangeModel.java:309)
	at java.desktop/javax.swing.DefaultBoundedRangeModel.setValueIsAdjusting(DefaultBoundedRangeModel.java:238)
	at java.desktop/javax.swing.JScrollBar.setValueIsAdjusting(JScrollBar.java:591)
	at java.desktop/javax.swing.plaf.basic.BasicScrollBarUI$TrackListener.mouseReleased(BasicScrollBarUI.java:1219)
	at java.desktop/java.awt.AWTEventMulticaster.mouseReleased(AWTEventMulticaster.java:297)
	at java.desktop/java.awt.Component.processMouseEvent(Component.java:6635)
	at java.desktop/javax.swing.JComponent.processMouseEvent(JComponent.java:3342)

This my subclass of ScrollBar that fires events

class ScrollBar(orientation0: Orientation.Value, value0: Int, blockIncrement0: Int, minimum0: Int, maximum0: Int)
  extends swing.ScrollBar {
...
  peer.addAdjustmentListener(new java.awt.event.AdjustmentListener {
    def adjustmentValueChanged(e: java.awt.event.AdjustmentEvent): Unit =
      publish(new swing.event.ValueChanged(me))
  })

and I'm quite sure that the exception occurred because I was unregistering a reaction during a reaction callback.