{"action":"created","issue":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2191","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2191/labels{/name}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2191/comments","events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2191/events","html_url":"https://github.com/ReactiveX/RxJava/issues/2191","id":52756194,"number":2191,"title":"Observable.cache() creates indefinite number of Threads with Schedulers.io()","user":{"login":"michelbetancourt","id":71665,"avatar_url":"https://avatars.githubusercontent.com/u/71665?v=3","gravatar_id":"","url":"https://api.github.com/users/michelbetancourt","html_url":"https://github.com/michelbetancourt","followers_url":"https://api.github.com/users/michelbetancourt/followers","following_url":"https://api.github.com/users/michelbetancourt/following{/other_user}","gists_url":"https://api.github.com/users/michelbetancourt/gists{/gist_id}","starred_url":"https://api.github.com/users/michelbetancourt/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/michelbetancourt/subscriptions","organizations_url":"https://api.github.com/users/michelbetancourt/orgs","repos_url":"https://api.github.com/users/michelbetancourt/repos","events_url":"https://api.github.com/users/michelbetancourt/events{/privacy}","received_events_url":"https://api.github.com/users/michelbetancourt/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":11,"created_at":"2014-12-23T16:52:56Z","updated_at":"2015-01-15T04:20:55Z","closed_at":null,"body":"Hi,\r\n\r\nIt would appear to me that there's an issue with using when using Observable.cache() and Schedulers.io().  The issue is that an indefinite amount of threads are created as opposed to having thread re-use.  \r\n\r\nThis can eventually lead the following fatal scenario: \"java.lang.OutOfMemoryError: Unable to create new native thread\".\r\n\r\nI did notice that when using Observable.cache() and Schedulers.from(Executors.newCachedThreadPool()) Rx is able to re-use threads.  \r\n\r\nOther than thread-caching configuration, it's not clear to me why these 2 implementations produce wildly different thread caching results and behavior.  I can only assume that this behavior is not intended.  It would be great to hear whether that's the case.\r\n\r\nHere's a sample program I've put together that demonstrates the 2 different results.  It's a very simple example that creates multiple observables and ensures that they are subscribed to and that result is read.  I'm using SettableFuture to mock a lag between calls.\r\n\r\nResult with Schedulers.from(Executors.newCachedThreadPool()) (last few items displayed, notice the re-use in threads):\r\nThread[pool-1-thread-18,5,main]\r\nThread[pool-1-thread-15,5,main]\r\nThread[pool-1-thread-20,5,main]\r\nThread[pool-1-thread-7,5,main]\r\nThread[pool-1-thread-6,5,main]\r\nThread[pool-1-thread-11,5,main]\r\nThread[pool-1-thread-17,5,main]\r\nThread[pool-1-thread-4,5,main]\r\nThread[pool-1-thread-21,5,main]\r\nThread[pool-1-thread-1,5,main]\r\nThread[pool-1-thread-18,5,main]\r\nThread[pool-1-thread-10,5,main]\r\nThread[pool-1-thread-7,5,main]\r\nThread[pool-1-thread-11,5,main]\r\nThread[pool-1-thread-20,5,main]\r\n\r\nResult with Schedulers.io() (last few items displayed, notice that the number is incremental, no re-use):\r\nThread[RxCachedThreadScheduler-187,5,main]\r\nThread[RxCachedThreadScheduler-196,5,main]\r\nThread[RxCachedThreadScheduler-189,5,main]\r\nThread[RxCachedThreadScheduler-198,5,main]\r\nThread[RxCachedThreadScheduler-191,5,main]\r\nThread[RxCachedThreadScheduler-200,5,main]\r\nThread[RxCachedThreadScheduler-193,5,main]\r\nThread[RxCachedThreadScheduler-195,5,main]\r\nThread[RxCachedThreadScheduler-197,5,main]\r\nThread[RxCachedThreadScheduler-199,5,main]\r\n\r\npublic class RxIndefiniateThreads {\r\n\r\n    static final Scheduler scheduler = Schedulers.from(Executors.newCachedThreadPool());\r\n\r\n    // static final Scheduler scheduler = Schedulers.io();\r\n    public static void main(String[] args) throws Exception {\r\n        \r\n        int i = 0;\r\n        final AtomicInteger counter = new AtomicInteger(0);\r\n        while (true) {\r\n            final SettableFuture<Object> first = SettableFuture.create();\r\n            final SettableFuture<Object> second = SettableFuture.create();\r\n            Observable<Object> observe = Async.fromCallable(new Callable<Object>() {\r\n\r\n                @Override\r\n                public Object call() throws Exception {\r\n                    System.out.println(Thread.currentThread());\r\n                    return first.get();\r\n                }\r\n\r\n            }, scheduler).observeOn(scheduler).flatMap(new Func1<Object, Observable<Object>>() {\r\n\r\n                @Override\r\n                public Observable<Object> call(Object t1) {\r\n                    System.out.println(Thread.currentThread());\r\n                    try {\r\n                        return Observable.just((Object) (t1.toString() + \" :: \" + second.get()));\r\n                    } catch (Exception e) {\r\n                        throw new IllegalStateException(\"Mock exception for second\");\r\n                    }\r\n                }\r\n            }).cache();\r\n\r\n            final Future<Object> future = observe.toBlocking().toFuture();\r\n            // Thread.sleep(2);\r\n            i++;\r\n            new Thread(new Runnable(){\r\n\r\n                @Override\r\n                public void run() {\r\n                    int count = counter.incrementAndGet();\r\n                    try {\r\n                        Thread.sleep(1);\r\n                    } catch (InterruptedException e) {\r\n                        e.printStackTrace();\r\n                    }\r\n                    first.set(\"done\");\r\n                    try {\r\n                        Thread.sleep(1);\r\n                    } catch (InterruptedException e) {\r\n                        e.printStackTrace();\r\n                    }\r\n                    second.set(\"done\");\r\n                    try {\r\n                        future.get();\r\n                    } catch (Exception e) {\r\n                        e.printStackTrace();\r\n                    }\r\n                    // System.out.println(\"doneWith=\" + count);\r\n                }\r\n                \r\n            }).start();\r\n            if (i == 100) {\r\n                break;\r\n            }\r\n        }\r\n        System.out.println(\"done\");\r\n    }\r\n    \r\n}\r\n\r\n\r\ndependencies:\r\n<dependency>\r\n  \t\t<groupId>com.google.guava</groupId>\r\n  \t\t<artifactId>guava</artifactId>\r\n  \t\t<version>18.0</version>\r\n  \t</dependency>\r\n  \t<dependency>\r\n            <groupId>io.reactivex</groupId>\r\n            <artifactId>rxjava</artifactId>\r\n            <version>1.0.3</version>\r\n        </dependency>\r\n        <dependency>\r\n            <groupId>io.reactivex</groupId>\r\n            <artifactId>rxjava-async-util</artifactId>\r\n            <version>0.21.0</version>\r\n            <exclusions>\r\n                <exclusion>\r\n                    <groupId>io.reactivex</groupId>\r\n                    <artifactId>rxjava</artifactId>\r\n                </exclusion>\r\n            </exclusions>\r\n        </dependency>\r\n  \t"},"comment":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/comments/70038048","html_url":"https://github.com/ReactiveX/RxJava/issues/2191#issuecomment-70038048","issue_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2191","id":70038048,"user":{"login":"duncani","id":688725,"avatar_url":"https://avatars.githubusercontent.com/u/688725?v=3","gravatar_id":"","url":"https://api.github.com/users/duncani","html_url":"https://github.com/duncani","followers_url":"https://api.github.com/users/duncani/followers","following_url":"https://api.github.com/users/duncani/following{/other_user}","gists_url":"https://api.github.com/users/duncani/gists{/gist_id}","starred_url":"https://api.github.com/users/duncani/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/duncani/subscriptions","organizations_url":"https://api.github.com/users/duncani/orgs","repos_url":"https://api.github.com/users/duncani/repos","events_url":"https://api.github.com/users/duncani/events{/privacy}","received_events_url":"https://api.github.com/users/duncani/received_events","type":"User","site_admin":false},"created_at":"2015-01-15T04:20:55Z","updated_at":"2015-01-15T04:20:55Z","body":"@zsxwing, fair enough, I've submitted PR: https://github.com/ReactiveX/RxJava/pull/2455"}}