Non-determinism when using InProcessTransport with direct executors

[grandseiken]

What version of gRPC are you using?

1.0.1

What JVM are you using (java -version)?

1.8.0_101

What did you do?

I am attempting to write unit tests for RPC systems without mocking stubs, which would be lovely if it worked. Recent discussions (e.g. here, 1.0.1 release notes, this example) suggest this is the new done thing and is supposed to work. To this end I am running an InProcessServer and calling it over an InProcessChannel with both ends using a DirectExecutor.

What did you expect to see?

The 1.0.1 release notes imply that this is now deterministic (after the removal of the grpc-ready thread pool, which was causing issues for me before), so I expected to see determinism in my unit test, and for the server handler to be called on the same thread that made the stub method call. The RPC method is a client-side streaming RPC, but I don't know if that is relevant.

What did you see instead?

The unit test is not deterministic, since the service handler seems to end up being called on a grpc-timer thread provided by the ScheduledExecutorService created by GrpcUtil.TIMER_SERVICE. The task that causes the call is the NameResolverStartTask in ManagedChannelImpl, and there appears to be no way to provide an alternative timer service.

See ManagedChannelImpl.java line 307.

For reference, the stack trace that I see on the handler side is as follows:

        ...
	at foo.bar.FooServiceImpl$1.onNext(FooServiceImpl.java:33)
	at io.grpc.stub.ServerCalls$2$1.onMessage(ServerCalls.java:206)
	at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messageRead(ServerCallImpl.java:237)
	at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1.runInContext(ServerImpl.java:485)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:52)
	at io.grpc.internal.SerializeReentrantCallsDirectExecutor.execute(SerializeReentrantCallsDirectExecutor.java:65)
	at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener.messageRead(ServerImpl.java:481)
	at io.grpc.inprocess.InProcessTransport$InProcessStream$InProcessClientStream.writeMessage(InProcessTransport.java:475)
	at io.grpc.internal.DelayedStream$3.run(DelayedStream.java:201)
	at io.grpc.internal.DelayedStream.drainPendingCalls(DelayedStream.java:121)
	at io.grpc.internal.DelayedStream.setStream(DelayedStream.java:90)
	at io.grpc.internal.DelayedClientTransport$PendingStream.createRealStream(DelayedClientTransport.java:383)
	at io.grpc.internal.DelayedClientTransport$PendingStream.access$100(DelayedClientTransport.java:370)
	at io.grpc.internal.DelayedClientTransport$2.run(DelayedClientTransport.java:262)
	at com.google.common.util.concurrent.MoreExecutors$DirectExecutor.execute(MoreExecutors.java:456)
	at io.grpc.internal.DelayedClientTransport.setTransportSupplier(DelayedClientTransport.java:259)
	at io.grpc.internal.ManagedChannelImpl$InterimTransportImpl.closeWithRealTransports(ManagedChannelImpl.java:740)
	at io.grpc.DummyLoadBalancerFactory$DummyLoadBalancer.handleResolvedAddresses(DummyLoadBalancerFactory.java:133)
	at io.grpc.internal.ManagedChannelImpl$NameResolverListenerImpl.onUpdate(ManagedChannelImpl.java:681)
	at io.grpc.internal.AbstractManagedChannelImplBuilder$DirectAddressNameResolverFactory$1.start(AbstractManagedChannelImplBuilder.java:323)
	at io.grpc.internal.ManagedChannelImpl$1NameResolverStartTask.run(ManagedChannelImpl.java:254)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)

Am I doing something wrong? Is there a workaround? It seems like doing the name resolution work on the ManagedChannelImpl's user-provided executor rather than its scheduledExecutor might fix it, but I don't know if there's some reason it's not done that way.

[grandseiken]

I have a potential fix for the issue above, but would still be interested in workarounds that don't require changes to gRPC in case the fix is not suitable or takes a while to make it into a release. Thanks.

[carl-mastrangelo]

Can you describe what you are trying to do? The idle mode timer doesn't usually get used in unit tests, but you seem to be using it here.

[grandseiken]

Hi notcarl! Thanks for your reply, sorry for not being clear.

I did not intend to use any sort of idle mode, and setting idleTimeout on the channel builder does not appear to affect the issue I'm describing.

Indeed, from looking at the source code it seems that this is because a newly-created channel queues up outgoing requests while some initial name-resolution work occurs; the name resolution work happens on the same scheduledExecutor as the idle timer (which we cannot replace with a direct executor); and when the name resolution is done the queued requests are dispatched; i.e. though we try to use a direct executor, the channel still uses a non-direct executor for important work.

I don't believe the problem is related to the idle timer, it just happens to use the same executor as the idle timer (which seems to be the mistake as far as I can see).

Here is a minimal example:

// foo_service.proto
syntax = "proto3";

package example.service;

message Empty {}

service FooService {
  rpc send (Empty) returns (Empty) {}
}

// FooTest.java
package example;

import com.google.common.truth.Truth;
import example.service.FooServiceGrpc;
import example.service.FooServiceGrpc.FooServiceImplBase;
import example.service.FooServiceOuterClass.Empty;
import io.grpc.Channel;
import io.grpc.Server;
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.stub.StreamObserver;
import org.junit.Test;

import java.util.concurrent.atomic.AtomicInteger;

public class FooTest {
  private AtomicInteger requestsReceived = new AtomicInteger();

  @Test
  public void test() throws Exception {
    // Prints "Running on: main"
    System.out.println("Running on: " + Thread.currentThread().getName());

    Server server = InProcessServerBuilder
        .forName("foo")
        .directExecutor()
        .addService(new FooService().bindService())
        .build();
    server.start();

    Channel channel = InProcessChannelBuilder
        .forName("foo")
        .directExecutor()
        // Could set idleTimeout here - has no effect
        .build();

    FooServiceGrpc.FooServiceStub stub = FooServiceGrpc.newStub(channel);
    stub.send(Empty.getDefaultInstance(), new StreamObserver<Empty>() {
      @Override
      public void onNext(Empty value) {}
      @Override
      public void onError(Throwable t) {}
      @Override
      public void onCompleted() {}
    });

    // Race condition! FooService onNext() is called on a different thread,
    // so the following line often fails.
    Truth.assertThat(requestsReceived.get()).isEqualTo(1);
    server.shutdown();
  }

  public class FooService extends FooServiceImplBase {
    @Override
    public void send(Empty empty, StreamObserver<Empty> observer) {
      // Prints "Handled on: grpc-timer-0", but we wanted "Handled on: main"
      System.out.println("Handled on: " + Thread.currentThread().getName());
      // How did we get here? Print stack trace
      new Exception().printStackTrace();
      requestsReceived.incrementAndGet();
      observer.onNext(Empty.getDefaultInstance());
      observer.onCompleted();
    }
  }
}

The reason that onNext is called on a different thread is due to the details I provided in the original comment, and it's fixed by the pull request I created.

Edit: updated to use non-streaming RPC in example, since it turns out it does not matter.

[ejona86]

@zhangkun83, what do you think?

We get to decide if this is important enough to "backport" to v1.0.x. Since I already have a commit, I am leaning toward fixing it in v1.0.x, especially since we took away mocking.

I was actually working to remove that abuse of an executor in ejona86@fa4e32a . It is held up now by the invasive load balancing changes. I think it will require a pretty different implementation after the LB changes, so if it is worthy of backporting, then we would probably just take my current commit and apply it to v1.0.x.

I question why we've not seen this before. When messing with my commit earlier, I wondered why using the executor wasn't causing problems. We just get lucky most of the time? Or maybe @dapengzhang0 was hitting more than one issue with the example tests?

We can't do the work on the user-provided executor, because it can cause a deadlock if it is a fixed-sized thread pool. We actually should do the work on the current thread; the only reason we don't is because we needed to avoid running the code with some locks held. My commit above passes a Runnable up the stack which is executed after the locks are freed.

As far as workarounds, you can "prime" the channel by doing a trash RPC (with blocking stub is probably easiest) before starting your verification. The method doesn't even need to be registered on the server.

[zhangkun83]

@ejona86, let's fix it on the 1.0.x branch first, and fix master once my LB refactor is done.

[grandseiken]

Thanks for the explanation @ejona86. Can confirm that the suggested workaround (do one throwaway blocking RPC on the channel before using it) avoids the problem. Looking forward to your fix!