Send the operations to the async executor asynchronously.  The conversion from hbase
object to cloud bigtable proto and the async call both take time (microseconds worth) that
could be parallelized, or at least removed from the user's thread.
15603
private ApiFuture<?> offer(Mutation mutation) {
    ApiFuture<?> future = null;
    try {
      if (mutation == null) {
        future = ApiFutures.immediateFailedFuture(
          new IllegalArgumentException("Cannot perform a mutation on a null object."));
      } else if (mutation instanceof Put) {
        future = bulkMutation.add(adapter.adaptEntry((Put) mutation));
      } else if (mutation instanceof Delete) {
        future = bulkMutation.add(adapter.adaptEntry((Delete) mutation));
      } else if (mutation instanceof Increment) {
        future = dataClient.readModifyWriteRowAsync(adapter.adapt((Increment) mutation));
        operationAccountant.registerOperation(ApiFutureUtil.adapt(future));
      } else if (mutation instanceof Append) {
        future = dataClient.readModifyWriteRowAsync(adapter.adapt((Append) mutation));
        operationAccountant.registerOperation(ApiFutureUtil.adapt(future));
      } else {
        future = ApiFutures.immediateFailedFuture(new IllegalArgumentException(
            "Encountered unknown mutation type: " + mutation.getClass()));
      }
    } catch (Exception e) {
      // issueRequest(mutation) could throw an Exception for validation issues. Remove the heapsize
      // and inflight rpc count.
      future = ApiFutures.immediateFailedFuture(e);
    }
    return future;
  }
445723
static <T extends RadixTable> void radixSort(
		final int size,
		final T table,
		final Source<? super T> source,
		final ExecutorService executor
	) {
		try {
			final int numTasks = table.numTasks;

			// Dynamically choose pass size
			/*
			// Must be power of two and less than or equal to 32
			final int BITS_PER_PASS;
			if(size < 0x80000) {
				BITS_PER_PASS = 8;
			} else {
				BITS_PER_PASS = 16;
			}
			final int PASS_SIZE = 1 << BITS_PER_PASS;
			final int PASS_MASK = PASS_SIZE - 1;
			 */

			// The same futures list is used by multiple stages below
			final List<Future<?>> runnableFutures;

			// The size of division of work (number of elements per task).
			final int sizePerTask;

			if(executor==null) {
				assert numTasks==1 : "Must have an executor when numTasks != 1";
				runnableFutures = null;
				sizePerTask = size;
			} else {
				assert numTasks>=2 : "Must not have an executor when numTasks < 2";
				runnableFutures = new ArrayList<>(numTasks);
				int spt = size / numTasks;
				if((spt*numTasks)<size) spt++; // Round-up instead of down
				sizePerTask = spt;
			}

			// Set of all bits seen to skip bit ranges that won't sort.
			int bitsSeen = 0;

			// Set of all bits not seen to skip bit ranges that won't sort.
			int bitsNotSeen = 0;

			// May only use concurrent import for random access sources
			if(executor!=null && source.useRandomAccess()) {
				// Perform concurrent import
				final List<Future<Source.ImportDataResult>> importStepFutures = new ArrayList<>(numTasks);
				for(
					int taskStart=0, toTaskNum=0;
					taskStart<size;
					taskStart+=sizePerTask, toTaskNum++
				) {
					int taskEnd = taskStart + sizePerTask;
					if(taskEnd > size) taskEnd = size;

					final int finalTaskStart = taskStart;
					final int finalTaskEnd   = taskEnd;
					final int finalToTaskNum = toTaskNum;
					importStepFutures.add(
						executor.submit(
							new Callable<Source.ImportDataResult>() {
								@Override
								public Source.ImportDataResult call() {
									return source.importData(table, finalTaskStart, finalTaskEnd, finalToTaskNum);
								}
							}
						)
					);
				}

				// Combine results
				for(Future<Source.ImportDataResult> importStepFuture : importStepFutures) {
					Source.ImportDataResult result = importStepFuture.get();
					bitsSeen |= result.bitsSeen;
					bitsNotSeen |= result.bitsNotSeen;
				}
			} else {
				// Single-threaded import
				Source.ImportDataResult result = source.importData(table, 0, size, 0);
				bitsSeen |= result.bitsSeen;
				bitsNotSeen |= result.bitsNotSeen;
			}
			bitsNotSeen ^= 0xffffffff;

			// Swap toQueues and fromQueues
			table.swapQueues();

			// Perform gather/scatter iterations
			int lastShiftUsed = 0;
			for(int shift=BITS_PER_PASS; shift<32; shift += BITS_PER_PASS) {
				// Skip this bit range when all values have equal bits.  For example
				// when going through the upper bits of lists of all smaller positive
				// or negative numbers.
				if(((bitsSeen>>>shift)&PASS_MASK) != ((bitsNotSeen>>>shift)&PASS_MASK) ) {
					lastShiftUsed = shift;
					if(executor!=null) {
						// Get some final values for anonymous inner class
						final int finalShift = shift;
						// Perform each concurrently with balanced concurrency
						int toTaskNum = 0;
						int taskFromQueueStart = 0;
						int taskTotalLength = 0;
						for(int fromQueueNum=0; fromQueueNum<PASS_SIZE; fromQueueNum++) {
							for(int fromTaskNum=0; fromTaskNum<numTasks; fromTaskNum++) {
								taskTotalLength += table.getFromQueueLength(fromTaskNum, fromQueueNum);
							}
							if(
								taskTotalLength>0 // Skip no output, such as all handle in previous tasks
								&& (
									taskTotalLength >= sizePerTask // Found fair share (or more)
									|| fromQueueNum==PASS_MASK // or is last task
								)
							) {
								final int finalToTaskNum = toTaskNum;
								final int finalTaskFromQueueStart = taskFromQueueStart;
								final int taskFromQueueEnd = fromQueueNum + 1;
								// Gather/scatter concurrent
								assert runnableFutures != null;
								runnableFutures.add(
									executor.submit(
										new Runnable() {
											@Override
											public void run() {
												for(int fromQueueNum=finalTaskFromQueueStart; fromQueueNum<taskFromQueueEnd; fromQueueNum++) {
													table.gatherScatter(
														finalShift,
														fromQueueNum,
														finalToTaskNum
													);
												}
											}
										}
									)
								);

								// Reset to next task
								toTaskNum++;
								taskFromQueueStart = taskFromQueueEnd;
								taskTotalLength = 0;
							}
						}
						// Wait for each gather/scatter task to complete
						assert runnableFutures != null;
						waitForAll(runnableFutures);
						runnableFutures.clear();
					} else {
						for(int fromQueueNum=0; fromQueueNum<PASS_SIZE; fromQueueNum++) {
							table.gatherScatter(shift, fromQueueNum, 0);
						}
					}

					// Swap from and to
					table.swapQueues();
				}
			}

			// Negative before positive to perform as signed integers
			int fromQueueStart = (lastShiftUsed+BITS_PER_PASS)==32 ? (PASS_SIZE>>>1) : 0;

			// May only use concurrent export for random access sources
			if(executor!=null && source.useRandomAccess()) {
				// Use indexed strategy with balanced concurrency
				final int fromQueueLast = (fromQueueStart-1) & PASS_MASK;
				int taskFromQueueStart = fromQueueStart;
				int taskOutIndex = 0;
				int taskTotalLength = 0;
				int fromQueueNum = fromQueueStart;
				do {
					for(int fromTaskNum=0; fromTaskNum<numTasks; fromTaskNum++) {
						taskTotalLength += table.getFromQueueLength(fromTaskNum, fromQueueNum);
					}
					if(
						taskTotalLength>0 // Skip no output, such as all handle in previous tasks
						&& (
							taskTotalLength >= sizePerTask // Found fair share (or more)
							|| fromQueueNum==fromQueueLast // or is last task
						)
					) {
						final int finalTaskFromQueueStart = taskFromQueueStart;
						final int finalTaskOutIndex = taskOutIndex;
						final int finalTaskFromQueueEnd = (fromQueueNum + 1) & PASS_MASK;
						// Queue concurrent
						assert runnableFutures != null;
						runnableFutures.add(
							executor.submit(
								new Runnable() {
									@Override
									public void run() {
										source.exportData(
											table,
											finalTaskFromQueueStart,
											finalTaskFromQueueEnd,
											finalTaskOutIndex
										);
									}
								}
							)
						);

						// Reset to next task
						taskFromQueueStart = finalTaskFromQueueEnd;
						taskOutIndex      += taskTotalLength;
						taskTotalLength    = 0;
					}
				} while(
					(fromQueueNum = (fromQueueNum + 1) & PASS_MASK)
					!= fromQueueStart
				);
				// Wait for each export task to complete
				waitForAll(runnableFutures);
				// This is the last stage, not needed: runnableFutures.clear()
			} else {
				// Use indexed strategy, single-threaded
				source.exportData(
					table,
					fromQueueStart,
					fromQueueStart,
					0
				);
			}
		} catch(InterruptedException | ExecutionException e) {
			throw new WrappedException(e);
		}
	}
310877
static AsyncResponse executeAsync(final @Nonnull String processName,
									  final @Nonnull Runnable operation) {
		return executeAsync(newSingleThreadExecutor(), processName, operation, true);
	}
1291835
@GwtIncompatible static <T> T invokeAnyImpl(
      ListeningExecutorService executorService,
      Collection<? extends Callable<T>> tasks,
      boolean timed,
      long timeout,
      TimeUnit unit)
      throws InterruptedException, ExecutionException, TimeoutException {
    checkNotNull(executorService);
    checkNotNull(unit);
    int ntasks = tasks.size();
    checkArgument(ntasks > 0);
    List<Future<T>> futures = Lists.newArrayListWithCapacity(ntasks);
    BlockingQueue<Future<T>> futureQueue = Queues.newLinkedBlockingQueue();
    long timeoutNanos = unit.toNanos(timeout);

    // For efficiency, especially in executors with limited
    // parallelism, check to see if previously submitted tasks are
    // done before submitting more of them. This interleaving
    // plus the exception mechanics account for messiness of main
    // loop.

    try {
      // Record exceptions so that if we fail to obtain any
      // result, we can throw the last exception we got.
      ExecutionException ee = null;
      long lastTime = timed ? System.nanoTime() : 0;
      Iterator<? extends Callable<T>> it = tasks.iterator();

      futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
      --ntasks;
      int active = 1;

      while (true) {
        Future<T> f = futureQueue.poll();
        if (f == null) {
          if (ntasks > 0) {
            --ntasks;
            futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
            ++active;
          } else if (active == 0) {
            break;
          } else if (timed) {
            f = futureQueue.poll(timeoutNanos, TimeUnit.NANOSECONDS);
            if (f == null) {
              throw new TimeoutException();
            }
            long now = System.nanoTime();
            timeoutNanos -= now - lastTime;
            lastTime = now;
          } else {
            f = futureQueue.take();
          }
        }
        if (f != null) {
          --active;
          try {
            return f.get();
          } catch (ExecutionException eex) {
            ee = eex;
          } catch (RuntimeException rex) {
            ee = new ExecutionException(rex);
          }
        }
      }

      if (ee == null) {
        ee = new ExecutionException(null);
      }
      throw ee;
    } finally {
      for (Future<T> f : futures) {
        f.cancel(true);
      }
    }
  }
450394
@SuppressWarnings("WeakerAccess")
  public ApiFuture<Void> deleteInstanceAsync(String instanceId) {
    String instanceName = NameUtil.formatInstanceName(projectId, instanceId);

    com.google.bigtable.admin.v2.DeleteInstanceRequest request =
        com.google.bigtable.admin.v2.DeleteInstanceRequest.newBuilder()
            .setName(instanceName)
            .build();

    return ApiFutures.transform(
        stub.deleteInstanceCallable().futureCall(request),
        new ApiFunction<Empty, Void>() {
          @Override
          public Void apply(Empty input) {
            return null;
          }
        },
        MoreExecutors.directExecutor());
  }
969733
@Override
    public java.util.concurrent.Future<DescribeReservedInstancesOfferingsResult> describeReservedInstancesOfferingsAsync(
            com.amazonaws.handlers.AsyncHandler<DescribeReservedInstancesOfferingsRequest, DescribeReservedInstancesOfferingsResult> asyncHandler) {

        return describeReservedInstancesOfferingsAsync(new DescribeReservedInstancesOfferingsRequest(), asyncHandler);
    }
412439
@Override public void dinvoke( H2ONode sender ) {
    _h2o = sender;
    Key k = _key;
    _key = null;          // Not part of the return result
    assert k.home();      // Gets are always from home (less we do replication)
    // Shipping a result?  Track replicas so we can invalidate.  There's a
    // narrow race on a moving K/V mapping tracking this Value just as it gets
    // deleted - in which case, simply retry for another Value.
    do  _val = H2O.get(k);      // The return result
    while( _val != null && !_val.setReplica(sender) );
    tryComplete();
  }
1076705
@Override public void dinvoke( H2ONode sender ) {
    _h2o = sender;
    Key k = _key;
    _key = null;          // Not part of the return result
    assert k.home();      // Gets are always from home (less we do replication)
    // Shipping a result?  Track replicas so we can invalidate.  There's a
    // narrow race on a moving K/V mapping tracking this Value just as it gets
    // deleted - in which case, simply retry for another Value.
    do  _val = Value.STORE_get(k); // The return result
    while( _val != null && !_val.setReplica(sender) );
    tryComplete();
  }
868326
@SuppressWarnings("WeakerAccess")
  public ApiFuture<Instance> getInstanceAsync(String instanceId) {
    String name = NameUtil.formatInstanceName(projectId, instanceId);

    com.google.bigtable.admin.v2.GetInstanceRequest request =
        com.google.bigtable.admin.v2.GetInstanceRequest.newBuilder().setName(name).build();

    return ApiFutures.transform(
        stub.getInstanceCallable().futureCall(request),
        new ApiFunction<com.google.bigtable.admin.v2.Instance, Instance>() {
          @Override
          public Instance apply(com.google.bigtable.admin.v2.Instance proto) {
            return Instance.fromProto(proto);
          }
        },
        MoreExecutors.directExecutor());
  }
207320
async function (events) {
    const oThis = this
      , promiseArray = []
    ;

    // nothing to do
    if (!events || events.length === 0) {
      oThis.updateIntercomDataFile();
      return Promise.resolve();
    }

    //TODO: last processed transaction index.
    for (var i = 0; i < events.length; i++) {
      const eventObj = events[i]
        , j = i
      ;

      // await oThis.processEventObj(eventObj);

      if (oThis.parallelProcessingAllowed()) {
        promiseArray.push(new Promise(function (onResolve, onReject) {
          setTimeout(function () {
            oThis.processEventObj(eventObj)
              .then(onResolve)
              .catch(function (error) {
                logger.error('##### inside catch block #####: ', error);
                return onResolve();
              });
          }, (j * 1000 + 100));
        }));
      } else {
        await oThis.processEventObj(eventObj)
          .catch(function (error) {
            logger.error('inside catch block: ', error);
            return Promise.resolve();
          });
      }
    }

    if (oThis.parallelProcessingAllowed()) {
      await Promise.all(promiseArray);
    }

    oThis.updateIntercomDataFile();
    return Promise.resolve();
  }
353147
public static Function<IHTTPSession, Response> asyncHandler(@javax.annotation.Nonnull final ExecutorService pool, final String mimeType, @javax.annotation.Nonnull final Consumer<OutputStream> logic, final boolean async) {
    return session -> {
      @javax.annotation.Nonnull final PipedInputStream snk = new PipedInputStream();
      @javax.annotation.Nonnull final Semaphore onComplete = new Semaphore(0);
      pool.submit(() -> {
        try (@javax.annotation.Nonnull OutputStream out = new BufferedOutputStream(new AsyncOutputStream(new PipedOutputStream(snk)))) {
          try {
            logic.accept(out);
          } finally {
            onComplete.release();
          }
        } catch (@javax.annotation.Nonnull final IOException e) {
          e.printStackTrace();
          throw new RuntimeException(e);
        }
      });
      if (!async) {
        try {
          onComplete.acquire();
        } catch (@javax.annotation.Nonnull final InterruptedException e) {
          throw new RuntimeException(e);
        }
      }
      return NanoHTTPD.newChunkedResponse(Response.Status.OK, mimeType, new BufferedInputStream(snk));
    };
  }
1038584
private void asyncMoreRequest(int startRow) {
    try {
      DataRequest moreRequest = new DataRequest();
      moreRequest.queryId = queryId;
      moreRequest.startRow = startRow;
      moreRequest.maxSize = maxBatchSize;
      logger.debug("Client requesting {} .. {}", startRow, (startRow + maxBatchSize - 1));

      DataResponse response = server.data(moreRequest);
      logger.debug("Client got response {} .. {}, more={}",
          new Object[] { response.startRow, (response.startRow + response.data.size() - 1), response.more });
      nextData.add(new Window(response.data, response.more));
    } catch (AvroRemoteException e) {
      this.nextData.addError(toSparqlException(e));
    } catch (Throwable t) {
      this.nextData.addError(t);
    }
  }
166414
@Override
  public void adapt(T operation, com.google.cloud.bigtable.data.v2.models.MutationApi<?> mutation) {
    throw new UnsupportedOperationException(
      String.format("The %s operation is unsupported.", operationDescription));
}
1124422
function (callback) {
        debug('');
        async.parallel([

          // prepare eventStore...
          function (callback) {
            debug('prepare eventStore...');

            self.eventStore.on('connect', function () {
              self.emit('connect');
            });

            self.eventStore.on('disconnect', function () {
              self.emit('disconnect');
            });

            self.eventStore.init(callback);
          },

          // prepare aggregateLock...
          function (callback) {
            debug('prepare aggregateLock...');

            self.aggregateLock.on('connect', function () {
              self.emit('connect');
            });

            self.aggregateLock.on('disconnect', function () {
              self.emit('disconnect');
            });

            self.aggregateLock.connect(callback);
          },

          // prepare commandBumper...
          function (callback) {
            if (!self.commandBumper) {
              return callback(null);
            }
            debug('prepare commandBumper...');

            self.commandBumper.on('connect', function () {
              self.emit('connect');
            });

            self.commandBumper.on('disconnect', function () {
              self.emit('disconnect');
            });

            self.commandBumper.connect(callback);
          }
        ], callback);
      }
232195
CompletableFuture<Void> connect(LocalConnection connection) {
    LocalConnection localConnection = new LocalConnection(listener.context, connections);
    connections.add(localConnection);
    connection.connect(localConnection);
    localConnection.connect(connection);
    return CompletableFuture.runAsync(() -> listener.listener.accept(localConnection), listener.context.executor());
  }
818670
public static CompletableFuture<Void> runAsync(Runnable runnable,
            Executor executor) {
        return runAsync(runnable,
                handleExceptionally("runAsync", runnable, executor), executor);
    }
178760
public void sendAsync() {
        if (!configuration.mailSendFilter().filter(this)) return;
        MailRequestCallbackFactory factory = configuration.mailRequestCallbackFactory();
        if (factory == null) {
            prepareSend();
            request().async().post(entity());
        } else
            sendAsync(factory.create(this));
    }