Zip with Backpressure Support #1446
Conversation
This supports both upstream and downstream backpressure.
| // Concurrent* since we need to read it from across threads | ||
| final ConcurrentLinkedQueue items = new ConcurrentLinkedQueue(); | ||
| final RxRingBuffer items = RxRingBuffer.getSpmcInstance(); |
There was a problem hiding this comment.
There is at most 1 consumer and 1 producer thread, why the SPMC queue?
There was a problem hiding this comment.
Oversight on my part, this should work as SPSC. I'll make the change.
|
RxJava-pull-requests #1397 ABORTED |
|
I get random hangs and test failures in OperatorGroupByTest > testFirstGroupsCompleteAndParentSlowToThenEmitFinalGroupsAndThenComplete hangs OperatorPivotTest > testPivotEvenAndOdd Exception OnSubscribeCacheTest > testWithPublishSubjectAndRepeat OnSubscribeCacheTest > testWithBehaviorSubjectAndRepeat RxRingBufferWithoutUnsafeTest > testConcurrency |
| } | ||
| if (emitted > THRESHOLD) { | ||
| for (Object obj : observers) { | ||
| ((InnerSubscriber) obj).request(emitted); |
There was a problem hiding this comment.
Why are we requesting more data from sources if the downstream didn't ask for values? It looks like as if request(n) is used as a repeated 'batching' operation. I guess this was required for backpressure unaware downstream which would "hang" unless new data is requested even now and then.
There was a problem hiding this comment.
So far I have treated upstream and downstream as decoupled on async operators like this with a buffer in them. Even if no further is requested from downstream it tries to keep the buffers full. This is similar to merge and observeOn. This is done because it can't know when a request will come in but assumes it will and keeps filling the buffers so the data is available when the downstream requests it.
I'm open to exploring alternatives if performance testing shows a different approach is better for throughput.
You'll also notice that it does not perform the request until after it passes the THRESHOLD value. This is done so it requests in batches rather than one at a time, which can be more efficient in certain use cases.
What are you recommending doing differently?
There was a problem hiding this comment.
Instead of introducing buffers, I'd keep the current unbounded operator behavior on the onNext stream for most operators and only decorate the request(n) stream going backwards to the source; basically what I did in my zip operator. Otherwise, I'd use a batch(n) operator I suggested.
But currently, I only considered from and range as the sources where this simplified request mangling is straightforward. Are there any other RxJava sources?
There was a problem hiding this comment.
Having unbounded buffers defeats the point of backpressure and makes it non-deterministic as to whether buffer float will occur from perspective of a user. The idea is to eliminate unbounded buffers unless the user asked for it such as in replay, toList or onBackpressureBuffer.
I think we should probably remove the The others all need to be researched. |
Oddly this hung in the |
|
From my command line it also hangs: It fails here: |
|
Here is the code ... and I don't understand this syntax at all: https://github.com/Netflix/RxJava/blob/master/rxjava-contrib/rxjava-scalaz/src/test/scala/rx/lang/scala/scalaz/ObservableZipSpec.scala#L26 |
|
Scalaz turns you code into messy tokensoup. I can't read it either. |
Similar to filter, it needs to request(1) each time it filters an onNext.
|
RxJava-pull-requests #1406 FAILURE |
|
RxJava-pull-requests #1407 SUCCESS |
|
Non-deterministic tests can be tracked here: #1455 |
Zip with Backpressure Support
This supports both upstream and downstream backpressure.