Interrupted exception when using blocking call with stubs reuse

[Dinesh Adwani]

I have grpc implemnetaion in java where i am using a blocking stubs
i am using those stubs from pool and
interestingly when load increases i see interrupted exception


My requirement is to reuse channel and stubs

What is correct way to reuse channel and stubs ?





    public ComputeVariableServiceGrpc.ComputeVariableServiceBlockingStub getOrCreateStub() {
        ComputeVariableServiceGrpc.ComputeVariableServiceBlockingStub stub = null;
        try {
            stub = stubPool.poll(grpcProperties.getClient().getClientPoolTimeoutInMillis(), TimeUnit.MILLISECONDS);
            if (isNull(stub) || isChannelClosed((ManagedChannel) stub.getChannel())) {
                log.debug("Stub not available in queue or connection was closed,  creating one...");
                stub = ComputeVariableServiceGrpc.newBlockingStub(newManagedChannel());
            }

        }
        catch (InterruptedException e) {
            log.error("Error while waiting for stub from pool: [{}], [{}]", e.getCause(), e.getMessage());
        }
        return stub;
    }


    public ComputeVariablesResult computeVariable(String payload, String typeId) {
        ComputeVariableServiceGrpc.ComputeVariableServiceBlockingStub computeVariableServiceBlockingStub = getOrCreateStub();//poll from stub pool
        CdsComputeVariablesResult response = CdsComputeVariablesResult.newBuilder().build();

        try {
            Request request = CdsComputeVariablesRequest.newBuilder().setTypeId(typeId).setPayload(payload).build();

            return computeVariableServiceBlockingStub.computeVariables(request);
            }
        }
        catch (StatusRuntimeException statusRuntimeException) {
            if (statusRuntimeException.getCause() != null) {
                log.error("Error while computing variables, error cause and message is:  [{}] [{}]", statusRuntimeException.getCause(), statusRuntimeException.getStatus().getDescription());
            }
            else {
                log.error("Error while computing variables, error message is:  [{}]", statusRuntimeException.getStatus().getDescription());
            }
        }
        finally {
            log.debug("Stub returned to queue");
            releaseStub(computeVariableServiceBlockingStub);//release to pool
        }
        log.debug("ComputeVariableResult with typeId: [{}] is  [{}] ", typeId, response);
        return response;
    }

    public void releaseStub(ComputeVariableServiceGrpc.ComputeVariableServiceBlockingStub computeVariableServiceBlockingStub) {
        boolean added = stubPool.offer(computeVariableServiceBlockingStub);
        if (!added) {
            log.info("Failed to add stub to the pool, remaining capacity is:  [{}]", stubPool.remainingCapacity());
        }
    }

    public void destroy() {
        stubPool.forEach(stub -> {
            if (nonNull(stub.getChannel())) {
                ((ManagedChannel) stub.getChannel()).shutdown();
            }
        });
        stubPool.clear();
    }
} 

[Eric Anderson]

gRPC does not interrupt threads. So if you are seeing thread interruption, that is triggered by your code. Also, without the full exception, there's not much to go on (and interruption is generally hard to figure out even with the full exception).

Stubs and channels are thread-safe, so you don't need to use stubPool, unless you were purposefully reducing the overall concurrency. You should also generally share channels across RPCs. Stubs are cheap, but the channel is expensive.

--
You received this message because you are subscribed to the Google Groups "grpc.io" group.
To unsubscribe from this group and stop receiving emails from it, send an email to grpc-io+u...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/grpc-io/0c3ab2f1-72ae-4705-ba91-9ec11af10d20n%40googlegroups.com.