Issues/3340 object array buffer implementation #3688
Conversation
|
Hi, @MikkelHJuul! Thank you for your hard effort putting all potential improvements in multiple PRs. We have reviewed all of them and estimated the impact on the existing behaviour. To understand the impact please consider the following sample: public static void main(String[] args) {
BaseSubscriber<String> subscriber = new BaseSubscriber<String>() {
@Override
protected void hookOnSubscribe(Subscription subscription) {
subscription.request(1);
}
};
Sinks.Many<String> sink = Sinks.many()
.replay()
.latest();
sink.asFlux().log("SUB1").subscribe(subscriber);
System.out.println("sending first");
sink.tryEmitNext("1");
System.out.println("sending more");
sink.tryEmitNext("2");
sink.tryEmitNext("3");
sink.tryEmitNext("4");
sink.tryEmitNext("5");
System.out.println("another subscriber");
sink.asFlux().log("SUB2").subscribe();
System.out.println("sub 1 more demand");
subscriber.request(4);once the sample is executed, the output is the following: as it could be noticed, all 4 elements that was not observed by subscriber 1 due to lack of demand are delivered (this is because of linked list data structure used right now) with your suggestion we have the following logs output: as it could be noticed only the last element is delivered while the others are dropped. For now, all of the PRs introduces behaviour change which is not acceptable for this operator (it stays in the codebase from the very first release of the 3.x line with this behaviour). To mitigate that breaking change we need to add an extra builder step in the or and Would this solution work for you? Thanks, |
This PR replace the internal buffer-implementation of the SizeBoundReplayBuffer to be an Atomic reference to an Object-array in stead of an atomic linked list.
I didn't manage to produce sane performance metrics (but my jmh also ran weird). I did however try implementing a dummy test where an atomic linked list was read vs. an atomic object array, that showed the array to be better performance (4 times better).
The implementation is obviously more heavy on the GC (but it's easily collectible garbage).
There is a thing to say on the Sinks.Many Normal/Unsafe interfaces, and this is obviously a lot more safe in concurrent usage.
SinksManyReplayLatest stress test looks like:
Results across all configurations:
while for the SizeBoundReplayBuffer looks like this:
This was after further fixes to the
addmethod, like so:This is very surprising since the first line has a racy non-atomic read and there are multiple racing atomic operations. I think having the head be final helped a bunch.