Support Synchronous Source in OnSubscribeRefCount #1753
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…eck to OperatorMulticast
|
Keep getting a failure here: Yet I can run that test in an infinite loop on my machine and never see it hang. |
|
I also don't see any use of |
|
Merging ... as this code passes in other environments, and will dig into this issue on main branch. |
benjchristensen
added a commit
that referenced
this pull request
Oct 14, 2014
Support Synchronous Source in OnSubscribeRefCount
|
Hmm, even in the main branch it's now breaking on Travis. I don't understand. Hung here: |
|
From Eclipse tests don't fail, from command-line I get these: |
|
I reverted this, something is very wrong. |
|
The refcount tests probably blocked all computation scheduler threads and this is why some tests fail. |
|
Those tests are pretty simple, so if they are causing threads to pileup then the operator implementation is a problem. |
|
A threaddump would be great, if you can get it hang in your command line long enough. |
|
I changed the unit tests and things seem to be working now: #1755 |
|
If I put these two unit tests back in it hangs: @Test
public void testRefCountUnsubscribeForSynchronousSource() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
Observable<Long> o = synchronousInterval().lift(detectUnsubscription(latch));
Subscriber<Long> sub = Subscribers.empty();
o.publish().refCount().subscribeOn(Schedulers.computation()).subscribe(sub);
Thread.sleep(100);
sub.unsubscribe();
assertTrue(latch.await(3, TimeUnit.SECONDS));
}
@Test
public void testSubscribeToPublishWithAlreadyUnsubscribedSubscriber() {
Subscriber<Object> sub = Subscribers.empty();
sub.unsubscribe();
ConnectableObservable<Object> o = Observable.empty().publish();
o.subscribe(sub);
o.connect();
}
private Operator<Long, Long> detectUnsubscription(final CountDownLatch latch) {
return new Operator<Long,Long>(){
@Override
public Subscriber<? super Long> call(Subscriber<? super Long> subscriber) {
latch.countDown();
return Subscribers.from(subscriber);
}};
}Here is where it hangs: It seems the The thread dumps of interest are: |
|
It's still very odd to me why the |
|
Since we do round-robin when assigning worker threads in computation scheduler, it is possible a blocked thread gets assigned to the test even if the others are free. |
|
Changing this: @Test
public void testRefCountUnsubscribeForSynchronousSource() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
Observable<Long> o = synchronousInterval().lift(detectUnsubscription(latch));
Subscriber<Long> sub = Subscribers.empty();
o.publish().refCount().subscribeOn(Schedulers.computation()).subscribe(sub);
Thread.sleep(100);
sub.unsubscribe();
assertTrue(latch.await(3, TimeUnit.SECONDS));
}
private Operator<Long, Long> detectUnsubscription(final CountDownLatch latch) {
return new Operator<Long,Long>(){
@Override
public Subscriber<? super Long> call(Subscriber<? super Long> subscriber) {
latch.countDown();
return Subscribers.from(subscriber);
}};
}to this @Test
public void testRefCountUnsubscribeForSynchronousSource() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
Observable<Long> o = synchronousInterval().doOnUnsubscribe(new Action0() {
@Override
public void call() {
latch.countDown();
}
});
Subscriber<Long> sub = Subscribers.empty();
o.publish().refCount().subscribeOn(Schedulers.computation()).subscribe(sub);
Thread.sleep(100);
sub.unsubscribe();
assertTrue(latch.await(3, TimeUnit.SECONDS));
}fixes the issue. So it seems that our Scheduler implementation taking a random thread can end up blocked on a hung thread. Obviously blocking an event loop thread is a "bad thing" ... and this is a nasty side effect. |
Yeah, what you said. |
|
K, got it I think ... The test passes though because the latch is immediately released as soon as a subscribe happens. So ... just a bad test. |
|
Good catch! |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Merge of #1695 into 1.x branch.