TestSubscriber, concurrent reading and writing of events

[tazle]

Some of our tests need to wait for certain events to arrive to a TestSubscriber to detect test completion. TestSubscriber does not currently allow this, because it's not possible to access the events that it has received in thread-safe manner. What we'd like to be able to do is:

Awaitility.waitAtMost(1, SECONDS).until(() -> assertEquals(expectedEvents, testSubscriber.getOnNextEvents()))

However, this sometimes fails with ConcurrentModificationException on the values list of TestSubscriber, because Awaitility is reading it while some onNext() event is being written to it.

We could do

Awaitility.waitAtMost(1, SECONDS).until(() -> assertEquals(expectedEvents.size(), testSubscriber.getValueCount());
assertEquals(expectedEvents, testSubscriber.,getOnNextEvents());

but that is not safe, since more events may arrive after the expected number of events have all been received (especially in error cases, if extra events are produced), so the comparison may still fail with ConcurrentModificationException.

We currently use version 1.2.0, but this seems to apply at least to the 1.x branch. 2.x has changed to a custom VolatileSizeArrayList, but it's not clear to me how that would in any way solve the remaining race in the second example.

I also wonder why VolatileSizeArrayList was introduced, since the existing separate volatile size variable would seem to provide enough happens-before edges to make the waiting safe (add() -> write to valueCount, read of valueCount -> read of the list). However, my understanding of the memory model is a bit hazy, so please correct me here if I'm wrong.

[akarnokd]

There is the awaitValueCount to make sure you access only as many elements as already committed. Then, use either assertReceivedOnNext or use getOnNextEvents().get(index) to access the backing ArrayList without triggering ConcurrentModificationException.

VolatileSizeArrayList was introduced to safeguard values() that returns a List so what you described should not happen in 2.x.

[akarnokd]

VolatileSizeArrayList was introduced to safeguard values() that returns a List so what you described should not happen in 2.x.

Correction: it only defends the get() method, the iterator() may still fail. Also its services are needed in both TestSubscriber and TestObserver.

[tazle]

Yes, unfortunately iterator() is used by e.g. toString(), which is how this originally popped up (albeit in the more racy form that compares list contents).

In the VolatileSizeArrayList variant values() could be made safe by returning a subList() that range-checks against the last index available at values() call time. I think that would also work for the 1.x variant, if getOnNextEvents() returned a values.subList(0, valueCount)

[akarnokd]

Or VolatileSizeArrayList could implement a custom Iterator.

Either way, looks like subList is a workaround you could apply externally to values()/getOnNextEvents() before doing assertions; no change to RxJava is needed in this case.

[tazle]

True. Or create a wrapper for TestSubscriber. No pressing need to touch TestSubscriber implementation in any case, although that would of course make it easier to use directly :)

[akarnokd]

Closing via #5713.