Weird Issue with SubscribeOn and Schedulers

[pequnio3]

Hio,

I'm using Rx ina function to perform batched batch-writes to a database we are using. I have a class which performs this functionality and uses subscribeOn with an executor pool to perform these operations.

I'm running into a situation where the below function ends up hanging and I believe it hangs due to my subscribeOn usage because when I take out subscribeOn, I'm able to reach the flat map following it, but unable to reach that code when subscribeOn is left in.

protected Observable<SherpaBatchWriteResponse> performBatchWriteOnBatchedRecords(final Observable<IDataRecord> encryptedRecordsObs) {
ListeningExecutorService executorService = MoreExecutors.listeningDecorator(new ThreadPoolExecutor(10, 10, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()));
return encryptedRecordsObs
                .buffer(MAX_BATCH_SET_RECORDS)
                .subscribeOn(Schedulers.executor(executorService))
                .flatMap(new Func1<List<IDataRecord>, Observable<SherpaBatchWriteResponse>>() {
                    @Override
                    public Observable<SherpaBatchWriteResponse> call(List<IDataRecord> writeRecords) {
                        try {
                            SherpaBWRequestItems[] items = createSherpaBatchWriteRequestItemsForDataRecords(writeRecords);
                            return Observable.just(sherpaService.batchWriteRecords(items, (long) switchboardConfig.sherpaBatchGetTimeout()));
                        } catch (SwitchboardException e) {
                            return Observable.error(e);
                        } catch (SherpaException e) {
                            return Observable.error(new SwitchboardException(SwitchboardErrorCode.SHERPA_BATCH_WRITE_FAILED_TO_WRITE_SOME_RECORDS, e));
                        }
                    }
                })

What also makes this situation weird is that I have two code paths that use this function, and one code path succeeds when using it while the other hangs. Granted I believe both code paths get to the point where it should be called

I have one code path that hangs that looks like

performBatchWriteOnBatchedRecords(Observable.from(Lists.newArrayList(record1,record2)))

where as a different code path succeeds that looks like

// observable that emits idataRecords and the iDataRecords are fetched in with 
// observables via a network call
Observable<IDataRecord> async = getRecordsAsync();
performBatchWriteOnBatchedRecords(async)

Do you have any idea why this may be happening or any pitfalls I may be falling into?

I've created my own operator to run a specific map/flatmap function on a pool, so I've been able to overcome this, but I'm still interested on why subscribeOn was being goofy for me.

Thanks,
Blake

[pequnio3]

Also the operator I wrote is below. I didn't use schedulers and this seems to do a good job a putting that one function on an external thread, but I'd be interested in what subscribe on does differently.

public class OperatorFlatMapInPool<R, T1> implements Observable.Operator<R, T1> {
    final ListeningExecutorService pool;
    final Func1<T1, Observable<R>> func;

    public OperatorFlatMapInPool(Func1<T1, Observable<R>> pFunc, ListeningExecutorService pPool) {
        this.pool = pPool;
        this.func = pFunc;
    }

    @Override
    public Subscriber<? super T1> call(final Subscriber<? super R> subscriber) {

        return new Subscriber<T1>(subscriber) {
            @Override
            public void onCompleted() {
                subscriber.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                subscriber.onError(e);
            }

            @Override
            public void onNext(final T1 t1) {
                // do function on another thread
                ListenableFuture<Observable<R>> result = pool.submit(new Callable<Observable<R>>() {
                    @Override
                    public Observable<R> call() throws Exception {
                        return func.call(t1);
                    }
                });
                // do rest of operations on main thread
                Observable.from(result)
                          .flatMap(new Func1<Observable<R>, Observable<R>>() {
                              @Override
                              public Observable<R> call(Observable<R> rObservable) {
                                  return rObservable;
                              }
                          })
                          .subscribe(new Action1<R>() {
                              @Override
                              public void call(R r) {
                                  subscriber.onNext(r);
                              }
                          },
                          new Action1<Throwable>() {
                              @Override
                              public void call(Throwable throwable) {
                                  subscriber.onError(throwable);
                              }
                          });
            }
        };
    }
}

[benjchristensen]

The subscribeOn operator simply invokes the subscribe to the given Observable on the given Scheduler. See the code here: https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/operators/OperatorSubscribeOn.java#L56

// do rest of operations on main thread

That is not moving to a "main thread", the work will happen on whatever thread is invoking onNext which in this case is from pool.

If you want the events to be received on a different thread you need to call observeOn to move the notifications.

For example:

Observable.from(result).observeOn(Schedulers.computation())

Using subscribeOn is effectively the same as you using pool.submit ... but not the way this example does it.

The Observable.from(result) code is on a Future and that will just cause Observable to block, as java.util.concurrent.Future does not have callbacks. You are using ListenableFuture, but RxJava does not know about that, so you'd need to wire that up yourself.

Quickly hacked code that shows manually running the code on a thread:

        Observable.create(new OnSubscribe<R>() {
            public void call(Subscriber<? super R> s) {
                s.add(Subscriptions.from(pool.submit(new Runnable() {
                    @Override
                    public void call() {
                        try {
                            s.onNext(data);
                            s.onCompleted();
                        } catch (Exception e) {
                            s.onError(e);
                        }
                    }
                })));
            }
        });

vs using subscribeOn to subscribe and run on a new thread:

        Observable.create(new OnSubscribe<R>() {
            public void call(Subscriber<? super R> s) {
                try {
                    s.onNext(data);
                    s.onCompleted();
                } catch (Exception e) {
                    s.onError(e);
                }
            }
        }).subscribeOn(Schedulers.newThread());

[akarnokd]

Hi. Have you tried debugging your first code? Does it reach the flatmap's function? Do you actually subscribe to the result of the method in the hanging case? (Otherwise this construct wastes a threadpool with 10 threads. ObserveOn won't paralellize it, you need parallelMerge for that). I don't see any reason for the hang if not the service call blocks.

[pequnio3]

Hey Ben, thanks for the help! I have a few more questions.

So to clarify my understanding of this, the subscribeOn functionality puts the entire chain on a different thread (Determined by scheduler)? I.e. if I have

Observable.just(1)
.flatMap(flatMapFn)
.map(mapFn1)
.subscribeOn(Schedulers.newThread())
.map(mapFn2)

This will result in the Observable.just(1)'s subscribe being on the new thread and then everything subsequent to that happening on that thread. So regardless of where I put subscribeOnIn the chain of observables, the observable that starts it off will be on the new thread as will everything after that (mapFn1,mapFn2 and FlatMap).

Is there anyway for mapFn2 to be run back again on the thread that originally hooked the observables together?

[pequnio3]

Hey akarnokd! Thanks for the reply!

I've put breakpoints in my code and in the successful circumstance, it does reach the flatmap, but does not reach the breakpoint in my unsuccessful circumstance. The other weird thing, is that I can successfully get there in my unit tests of the broken code path (wat!) I'm fine with the 10 threads being wasted here because in actual implementation, multiple requests will be using each of those threads (I dont want 1 big batch write to hog the thread pool from other requests).

In what situations would something like this hang?

[akarnokd]

Does the following construct work for you (please excuse the lambda syntax, I'm on a tablet right now)?

Observable.from(Arrays.asList(1, 2)).buffer(2).
subscribeOn(Schedulers.executor(executorService)).flatMap(b -> Observable.just(1)).subscribe(System.out::println)

Does it work if you buffer by 3?

[pequnio3]

i did:

 @Test
    public void aTest() {
        final ListeningExecutorService executorService = MoreExecutors.listeningDecorator(new ThreadPoolExecutor(10, 10, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()));
        System.out.println("A: " +Thread.currentThread().getId());
        Observable.from(Arrays.asList(1, 2))
                  .buffer(3)
                  .subscribeOn(Schedulers.executor(executorService))
                  .flatMap(new Func1<List<Integer>, Observable<?>>() {
                      @Override
                      public Observable<Integer> call(List<Integer> integers) {
                          return Observable.just(1);
                      }
                  })
                  .map(new Func1<Object, Object>() {
                      @Override
                      public Object call(Object o) {
                          System.out.println("B: " +Thread.currentThread().getId());
                          System.out.println("done");
                          return null;
                      }
                  })
                  .toBlockingObservable()
                  .last();
        System.out.println("C: " +Thread.currentThread().getId());
    }

worked for both 2 and 3buffer, output was:

A: 1
B: 10
done
C: 1

[akarnokd]

The last thing I can think of is to create a thread dump when the first case hangs.

[benjchristensen]

Is this still outstanding and needing resolution?

[pequnio3]

not quite. I didnt get to figure out how to fix it, however,

.subscribeOn(Schedulers.executor(executorService))

appears to be remove from the newester version of rxjava, and we have figured out a solution for ourselves.

Thanks for your help!

[benjchristensen]

Closing out as a solution was found.

Since the last comments Scheduler.from(Executor e) was added that creates a valid Scheduler from an Executor that complies with the Rx concurrency requirements.

[bravekingzhang]

terrible