NewThreadScheduler thread exhaustion

[akarnokd]

The current implementation of NewThreadScheduler creates a new ThreadPoolExecutor for each non-recursive schedule invocation. The pool is never terminated (except when the application terminates), therefore, pools no longer in use still keep one thread active (blocked on waiting for the next task which can't arrive). If too many pools are created, the operating system might run out of fresh threads, causing "native thread creation failed" throughout the application. To demonstrate the problem, here is a small test case:

public class NewThreadSchedulerTest {
    @Test
    public void testExhaustion() throws InterruptedException {
        Scheduler s = Schedulers.newThread();
        BlockingObservable<Integer> source = Observable.from(1)
                .subscribeOn(s).toBlockingObservable();
        for (int i = 0; i < 10000; i++) {
            Assert.assertEquals((Integer)1, source.single());
        }
    }
}

(This test fails on my Windows 7 x86 after about 4000 iterations.)

Its equivalent Rx.NET code, however, works fine:

class Program {
        static void Main(string[] args) {
            IScheduler s = NewThreadScheduler.Default;
            IObservable<int> source = Observable.ToObservable(new int[] { 1 }, s)
                    .SubscribeOn(s);
            int sum = 0;
            for (int i = 0; i < 10000; i++) {
                sum += source.Single();
                Thread.Sleep(1);
                if (i % 256 == 0) {
                    Console.Write(i);
                    Console.Write(": ");
                    Console.WriteLine(sum);
                }
            }
        }
    }

The program's thread count remains about 15. (It seems that RxNET uses vanilla threads in combination with some kind of CurrentThreadScheduler in it and has an explicit option to quit if no more tasks are queued or pending).

I've tried to fix NewThreadScheduler by setting a keepalive on the thread pool, but unfortunately it breaks the OperationParallelMergeTest because a restarted thread in the thread pool will get a different identifier. I've tried adding wip counter to capture when to terminate the pool, but didn't work (probably due escaped scheduler reference is invoked outside the callback function/action).

Any ideas?

[benjchristensen]

I think the issue is two-fold:

1. NewThreadScheduler does not shutdown the executor in the subscription it returns

2. BlockingObservable does not obey guideline 4.3 and unsubscribe when it completes. Generally we rely upon a user provided Observer and the SafeObserver doing this, but BlockingObservable never receives a user provided Observer.

4.3. Assume resources are cleaned up after an OnError or OnCompleted
message
Paragraph 4.1 states that no more messages should arrive after an OnError or OnCompleted message. This makes it possible to cleanup any resource used by the subscription the moment an OnError or OnCompleted arrives. Cleaning up resources immediately will make sure that any side-effect occurs in a predictable fashion. It also makes sure that the runtime can reclaim these resources.
Sample
Observable.Using(
() => new FileStream(@"d:\temp\test.txt", FileMode.Create), fs => Observable.Range(0, 10000)
.Select(v => Encoding.ASCII.GetBytes(v.ToString()))
            .WriteToStream(fs))
        .Subscribe();
In this sample the Using operator creates a resource that will be disposed upon unsubscription. The Rx contract for cleanup ensures that unsubscription will be called automatically once an OnError or OnCompleted message is sent.

[benjchristensen]

I'm working on this.

[benjchristensen]

I've confirmed it's what I thought ... the BlockingObservable doesn't call unsubscribe when onComplete/onError happen and thus the resources are not freed. The example code works fine if BlockingObservable is not involved and a normal async subscribe is used.

[akarnokd]

Before fixing BO, would it be possible to merge our PRs changing the same files so we don't have to rebase everything again?

Unfortunately, there is no good way to detect if an iteration has stopped in Java. We can have a forEach overload that takes a predicate.

[benjchristensen]

Yeah I can merge it. I intend on working more on this tonight, just haven't had time to finish it during my day hours.

For an iteration stopping, isn't it sufficient to know that we have exited that block of code and just have the 'completion marker' style thing invoked right after we exit the iteration?

[akarnokd]

When Java 7 introduced the DirectoryStream, they faced a similar issue: one would need to call close on the Iterator but then you can't use for-each loop for enumeration. The solution was to have DirectoryStream Closeable and return only a single Iterator, therefore, this pattern is established:

try (DirectoryStream ds = list("*")) {
   for (Path p : ds) {
      if (p.toString().endsWith(".xml")) {
         break;
      }
   }
}

We can have a BlockingObservable.onetimeIterable() returning an IterableUnsubscribable instance and require a similar pattern.

Alternatively, the following forEach overload could do the trick:

BlockingObservable.forEachUntil(Func1<T, Boolean> breakoutConsumer);

Where the function breakoutConsumer should return true if it wishes to continue the loop.

[benjchristensen]

That seems to solve a different problem, breaking out of a forEach loop on an infinite stream as opposed to knowing when it completes.

The takeUntil operator already enables conditionally breaking an infinite stream, forEachUntil is just merging those together.

What exactly is the use case you were referring to for this statement => "there is no good way to detect if an iteration has stopped in Java"?

[akarnokd]

The TakeUntil seems to be appropriate, but we still need a BO.await() to run to run through the results and block.

A case:

public Integer one() {
return Observable.from(1).subscibeOn(Schedulers.threadPoolForComputation())
.repeat().toBlockingObservable().toIterable().iterator().next();
}

We don't have the IDisposable in Java, which seems to handle exactly these issues.

[benjchristensen]

What is the issue in that code example other than the one we already know that causes BlockingObservable to not correctly unsubscribe?

[akarnokd]

Up to the toBlockingObservable(), it is an infinite stream of 1s. We take the first value of this stream and leave the whole thing. There is no way the BO can detect that nothing is iterating it further and can't stop because the upstream won't complete. One would need a way to tell the Iterator to stop.

[benjchristensen]

I'd argue that .next() on an infinite iterator is bad code. Same way that calling toList() on an infinite stream is bad code.

[akarnokd]

Yes, if you know that it is an infinite iterator. The example could feature a

range(1, 10000000).scheduleOn(Schedulers.threadPoolForComputation())
.toBlockingObservable().toIterable().iterator().next()

which it is finite yet wasteful. We could return a subclass of Iterable and Iterator that allow unsubscribing. When someone uses the StoppableIterable and StoppableIterator without calling stop, that is definitely a bad code. When you have just the Iterable, there is no contract or way to do the stopping. Similar problem appeared with Streams in the Lambda-dev recently.

[akarnokd]

Closing this as the original test case now passes as expected.