Make NewThreadScheduler create Daemon threads

[benjchristensen]

This matches the behavior of Schedulers.COMPUTATION_EXECUTOR and Schedulers.IO_EXECUTOR.

See https://groups.google.com/forum/#!topic/rxjava/Qe1qi0aHtnE and #431 (comment)

[cloudbees-pull-request-builder]

RxJava-pull-requests #561 FAILURE
Looks like there's a problem with this pull request

[akarnokd]

This solves the non-termination problem, but should those terminations be so harsh? Perhaps allowing the core threads to timeout might be a better alternative to all of these schedulers, although the tuning of timeout value could be of concern.

[benjchristensen]

I don't understand what you mean. Can you elaborate? For example, what timeout are you referring to?

This code should not prevent the JVM from exiting:

    public static void main(String args[]) {
        Observable.from(1, 2, 3, 4).observeOn(Schedulers.newThread()).toBlockingObservable().forEach(new Action1<Integer>() {

            @Override
            public void call(Integer i) {
                System.out.println("i: " + i);
            }
        });
    }

It should be application logic that chooses to block, not the exit of a JVM.

[akarnokd]

I was talking about the thread pool and core thread timeouts (aka keepalive time):

private static class EventLoopScheduler extends Scheduler {
    private final ExecutorService executor;

    private EventLoopScheduler() {
        ThreadPoolExecutor e = (ThreadPoolExecutor)Executors.newFixedThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "RxNewThreadScheduler-" + count.incrementAndGet());
            }
        });
        e.setKeepAliveTime(1, TimeUnit.SECONDS); // <-- mabe lower, maybe greater time in general 
        e.allowCoreThreadTimeOut(true);
        executor = e;
    }
    ...
}

The problem I encounter with daemon-threaded pools is that when the main thread exists, these threads stop immediately and don't finish their given tasks. This is why there are sleep() calls in various tests. They require a similar subscribe (new pool)/unsubscribe(shutdown pool) pattern to be used. I guess there are pros and cons for both cases.

[benjchristensen]

If something needs to be waited on then toBlockingObservable() (or some other similar approach) should be used to block the main thread until the application completes. We don't want the main thread waiting on Scheduler threads. If an app let's the main thread complete then the app should finish.

Sleep and timeouts are non-deterministic and incorrect for using to keep an application running.

[akarnokd]

Fair enough. But there is still an issue with NewThreadScheduler. Even if quitting the application removes these threads, performing thousand scheduling operation will yield thousand threads or an exception about the inability to create more threads. So until the application quits, these threads stay. I suggest adding the

e.setKeepAliveTime(50, TimeUnit.MILLISECONDS);
e.allowCoreThreadTimeOut(true);

from my example to the NewThreadScheduler to be safe. 50 milliseconds might be to big since threads are no way to be reused, so decreasing it to a small number (zero is disallowed) is also possible.

Test program:

public static void main(String[] args) throws Exception {
        BlockingObservable<Integer> source = Observable.from(1).observeOn(Schedulers.newThread()).toBlockingObservable();
        for (int i = 0; i < 20000; i++) {
            Thread.sleep(1);
            source.single();
            if (i % 1000 == 0) {
                System.out.println(i);
            }
        }
    }
}

[cloudbees-pull-request-builder]

RxJava-pull-requests #562 SUCCESS
This pull request looks good

[akarnokd]

Should I open a new issue for the thread exhaustion problem?