runTransaction runs callback on very limited gRPC thread

[rawler]

Steps to reproduce

1. Use provided example
2. Transactions-part of example is expected to complete in less than 7s, but in practice completes in ~40s.

Code example

    Firestore firestore = FirestoreOptions.getDefaultInstance().getService();

    final List<DocumentReference> documents = new ArrayList<>();
    final List<ApiFuture<WriteResult>> writeResults = new ArrayList<>();
    for (int i=0; i < 64; i++) {
      final DocumentReference docRef = firestore.collection("counters").document(""+i);
      writeResults.add(docRef.set(Map.of("value", 0)));
      documents.add(docRef);
    }
    for (var writeResult : writeResults) {
      writeResult.get();
    }

    System.err.println("Starting update-transactions");
    var transactionsStart = Instant.now();

//  Using these specific options, it seems to work
//     var runner = Executors.newCachedThreadPool();
//     var options = TransactionOptions.createReadWriteOptionsBuilder().setExecutor(runner).build();

    final List<ApiFuture<Void>> transactions = new ArrayList<>();
    for (var docRef : documents) {
      transactions.add(firestore.runTransaction(transaction -> {
          // Just simulate something blocking
          Thread.sleep(5000);
          transaction.update(docRef, "value", 1);
          return null;
        }));
    }

    for (var transactionFuture : transactions) {
      transactionFuture.get();
    }

    var elapsed = Duration.between(transactionsStart, Instant.now());
    System.err.println("Done updating in "+elapsed);

External references such as API reference guides

https://cloud.google.com/firestore/docs/samples/firestore-transaction-document-update#firestore_transaction_document_update-java

Any additional information below

Breaking inside the transaction-handler, it seems to be called on the a "grpc-transport-X" thread. I would assume the intention, judging by TransactionRunner is to use the default executor from the service-instance. This seems to be a java.util.concurrent.ScheduledThreadPoolExecutor, configured with a maximumPoolSize of 2147483647, but in practice somehow limited to 8, probably being the cause of the problem.

[dconeybe]

Thank you for the report. I will take a look.

[rawler]

Dug a bit deeper. Turns out that since ScheduledThreadPoolExecutor is using an unbounded queue implementation, its "corePoolSize" is also its practical maximumPoolSize. So the pool running here really is bound to a maximum of 8 threads.

Looking at where this is used, in GrpcFirestoreRpc, it's both passed on as the actual Grpc-transport Executor, but also reused to drive all snapshotListeners and all transaction-callbacks (that don't explicitly use their own pool). This means that any callback-function to snapshot-listeners or transactions, are in practice also bound to a maximum of 8-concurrent. (Unless the GrpcTransportOptions are overriden during setup).

Unless this is by design, perhaps GrpcFirestoreOptions should have some "callbackExecutor", that does not need to be Scheduled, and perhaps default to Executors.newCachedThreadPool?