Java 类com.google.common.util.concurrent.ListeningExecutorService 实例源码

@Test
public void testSValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final byte[] hash = HashUtil.sha3(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECKey.ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.doSign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
    final ECKey.ECDSASignature duplicate = new ECKey.ECDSASignature(sigs.get(0).r, sigs.get(0).s);
    assertEquals(sigs.get(0), duplicate);
    assertEquals(sigs.get(0).hashCode(), duplicate.hashCode());
}

public ApacheThriftMethodInvoker(
        ListeningExecutorService executorService,
        ListeningScheduledExecutorService delayService,
        TTransportFactory transportFactory,
        TProtocolFactory protocolFactory,
        Duration connectTimeout,
        Duration requestTimeout,
        Optional<HostAndPort> socksProxy,
        Optional<SSLContext> sslContext)
{
    this.executorService = requireNonNull(executorService, "executorService is null");
    this.delayService = requireNonNull(delayService, "delayService is null");
    this.transportFactory = requireNonNull(transportFactory, "transportFactory is null");
    this.protocolFactory = requireNonNull(protocolFactory, "protocolFactory is null");
    this.connectTimeoutMillis = Ints.saturatedCast(requireNonNull(connectTimeout, "connectTimeout is null").toMillis());
    this.requestTimeoutMillis = Ints.saturatedCast(requireNonNull(requestTimeout, "requestTimeout is null").toMillis());
    this.socksProxy = requireNonNull(socksProxy, "socksProxy is null");
    this.sslContext = requireNonNull(sslContext, "sslContext is null");
}

@Test
public void testSValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final byte[] hash = HashUtil.sha3(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.doSign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
    final ECKey.ECDSASignature duplicate = new ECKey.ECDSASignature(sigs.get(0).r, sigs.get(0).s);
    assertEquals(sigs.get(0), duplicate);
    assertEquals(sigs.get(0).hashCode(), duplicate.hashCode());
}

@Provides
@Singleton
static DSLContext dbContext(
    DataSource dataSource, @ForDatabase ListeningExecutorService dbExecutor) {
  Configuration configuration =
      new DefaultConfiguration()
          .set(dbExecutor)
          .set(SQLDialect.MYSQL)
          .set(new Settings().withRenderSchema(false))
          .set(new DataSourceConnectionProvider(dataSource))
          .set(DatabaseUtil.sfmRecordMapperProvider());
  DSLContext ctx = DSL.using(configuration);
  // Eagerly trigger JOOQ classinit for better startup performance.
  ctx.select().from("curio_server_framework_init").getSQL();
  return ctx;
}

@Test
public void testSValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final byte[] hash = HashUtil.sha3(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.doSign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
    final ECKey.ECDSASignature duplicate = new ECKey.ECDSASignature(sigs.get(0).r, sigs.get(0).s);
    assertEquals(sigs.get(0), duplicate);
    assertEquals(sigs.get(0).hashCode(), duplicate.hashCode());
}

@Setup(Level.Trial)
@Override
public void setUp() throws Exception {
    ListeningExecutorService dsExec = MoreExecutors.newDirectExecutorService();
    executor = MoreExecutors.listeningDecorator(
            MoreExecutors.getExitingExecutorService((ThreadPoolExecutor) Executors.newFixedThreadPool(1), 1L,
                    TimeUnit.SECONDS));

    InMemoryDOMDataStore operStore = new InMemoryDOMDataStore("OPER", dsExec);
    InMemoryDOMDataStore configStore = new InMemoryDOMDataStore("CFG", dsExec);
    Map<LogicalDatastoreType, DOMStore> datastores = ImmutableMap.of(
        LogicalDatastoreType.OPERATIONAL, (DOMStore)operStore,
        LogicalDatastoreType.CONFIGURATION, configStore);

    domBroker = new SerializedDOMDataBroker(datastores, executor);
    schemaContext = BenchmarkModel.createTestContext();
    configStore.onGlobalContextUpdated(schemaContext);
    operStore.onGlobalContextUpdated(schemaContext);
    initTestNode();
}

@Before
public void setupStore() {
    InMemoryDOMDataStore operStore = new InMemoryDOMDataStore("OPER", MoreExecutors.newDirectExecutorService());
    InMemoryDOMDataStore configStore = new InMemoryDOMDataStore("CFG", MoreExecutors.newDirectExecutorService());
    schemaContext = TestModel.createTestContext();

    operStore.onGlobalContextUpdated(schemaContext);
    configStore.onGlobalContextUpdated(schemaContext);

    ImmutableMap<LogicalDatastoreType, DOMStore> stores = ImmutableMap.<LogicalDatastoreType, DOMStore> builder() //
            .put(CONFIGURATION, configStore) //
            .put(OPERATIONAL, operStore) //
            .build();
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
    domBroker = new SerializedDOMDataBroker(stores, executor);
}

@Before
public void setupStore() {
    InMemoryDOMDataStore operStore = new InMemoryDOMDataStore("OPER", MoreExecutors.newDirectExecutorService());
    InMemoryDOMDataStore configStore = new InMemoryDOMDataStore("CFG", MoreExecutors.newDirectExecutorService());
    schemaContext = TestModel.createTestContext();

    operStore.onGlobalContextUpdated(schemaContext);
    configStore.onGlobalContextUpdated(schemaContext);

    ImmutableMap<LogicalDatastoreType, DOMStore> stores = ImmutableMap.<LogicalDatastoreType, DOMStore> builder() //
            .put(CONFIGURATION, configStore) //
            .put(OPERATIONAL, operStore) //
            .build();

    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
    domBroker = new SerializedDOMDataBroker(stores, executor);
}

/**
 * @deprecated This method is only used from configuration modules and thus callers of it
 *             should use service injection to make the executor configurable.
 */
@Deprecated
public static synchronized ListeningExecutorService getDefaultCommitExecutor() {
    if (COMMIT_EXECUTOR == null) {
        final ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-commit-%d").build();
        /*
         * FIXME: this used to be newCacheThreadPool(), but MD-SAL does not have transaction
         *        ordering guarantees, which means that using a concurrent threadpool results
         *        in application data being committed in random order, potentially resulting
         *        in inconsistent data being present. Once proper primitives are introduced,
         *        concurrency can be reintroduced.
         */
        final ExecutorService executor = Executors.newSingleThreadExecutor(factory);
        COMMIT_EXECUTOR = MoreExecutors.listeningDecorator(executor);
    }

    return COMMIT_EXECUTOR;
}

private List<SingularityS3Log> getS3Logs(S3Configuration s3Configuration, Optional<String> group, Collection<String> prefixes) throws InterruptedException, ExecutionException, TimeoutException {
  if (prefixes.isEmpty()) {
    return Collections.emptyList();
  }

  ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(Math.min(prefixes.size(), s3Configuration.getMaxS3Threads()),
      new ThreadFactoryBuilder().setNameFormat("S3LogFetcher-%d").build()));

  try {
    List<SingularityS3Log> logs = Lists.newArrayList(getS3LogsWithExecutorService(s3Configuration, group, executorService, prefixes));
    Collections.sort(logs, LOG_COMPARATOR);
    return logs;
  } finally {
    executorService.shutdownNow();
  }
}

private void onFinish(SingularityExecutorTask task, Protos.TaskState taskState) {
  processKiller.cancelDestroyFuture(task.getTaskId());

  tasks.remove(task.getTaskId());
  processRunningTasks.remove(task.getTaskId());
  processBuildingTasks.remove(task.getTaskId());

  task.cleanup(taskState);

  ListeningExecutorService executorService = taskToShellCommandPool.remove(task.getTaskId());

  if (executorService != null) {
    executorService.shutdownNow();
    try {
      executorService.awaitTermination(5, TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
      LOG.warn("Awaiting shutdown of shell executor service", e);
    }
  }

  logging.stopTaskLogger(task.getTaskId(), task.getLogbackLog());

  checkIdleExecutorShutdown(task.getDriver());
}

@Test
public void testConcurrentFetchTasks() throws Exception {
  // Test for regression of AURORA-1625
  ListeningExecutorService executor =
      MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5));
  assertStoreContents();
  saveTasks(TASK_A, TASK_B, TASK_C, TASK_D);

  List<ListenableFuture<Integer>> futures = Lists.newArrayList();

  for (int i = 0; i < 100; i++) {
    futures.add(executor.submit(() -> Iterables.size(fetchTasks(Query.unscoped()))));
  }

  Future<List<Integer>> f = Futures.allAsList(futures);

  executor.shutdown();
  executor.awaitTermination(1, TimeUnit.MINUTES);

  assertEquals(Iterables.getOnlyElement(ImmutableSet.copyOf(f.get())), (Integer) 4);
}

private void shutdown(@Nullable final ListeningExecutorService executorService) {
    if (executorService != null) {
        executorService.shutdown();
        try {
            // Wait a while for existing tasks to terminate
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow(); // Cancel currently executing tasks
                // Wait a while for tasks to respond to being cancelled
                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                    log.warn("Thread pool for metacat refresh did not terminate");
                }
            }
        } catch (InterruptedException ie) {
            // (Re-)Cancel if current thread also interrupted
            executorService.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }
}

@Test
public void exceptionArentCached_deferredFuture() throws Exception {
  ListeningExecutorService exec = listeningDecorator(Executors.newSingleThreadExecutor());
  try {
    executor = TestDeduplicatingExecutor.create(s -> {
      if (s == first) {
        return exec.submit(() -> {
          Thread.sleep(50);
          throw new IllegalArgumentException();
        });
      }
      return Futures.immediateFuture(s);
    });
    exceptionsArentCached();
  } finally {
    exec.shutdownNow();
  }
}

private Optional<PostCommitActions> buildPostCommitter(CommitTable.Client commitTableClient ) {

            PostCommitActions postCommitter;
            PostCommitActions syncPostCommitter = new HBaseSyncPostCommitter(hbaseOmidClientConf.getMetrics(),
                                                                             commitTableClient);
            switch(hbaseOmidClientConf.getPostCommitMode()) {
                case ASYNC:
                    ListeningExecutorService postCommitExecutor =
                            MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor(
                                    new ThreadFactoryBuilder().setNameFormat("postCommit-%d").build()));
                    postCommitter = new HBaseAsyncPostCommitter(syncPostCommitter, postCommitExecutor);
                    break;
                case SYNC:
                default:
                    postCommitter = syncPostCommitter;
                    break;
            }

            return Optional.of(postCommitter);
        }

/**
 * Counts yield and q30 of fastqs in the fastqsPerSample multimap, using 1 thread per file.
 * The yield and q30 of the Undetermined sample will count towards the total yield and q30 of the flowcell.
 *
 * @param fastqsPerSample multimap of sampleName and fastqs to process
 * @param threadCount     number of maximum threads
 * @return FastqTracker with yield and q30 stats for the fastqs processed.
 */

@NotNull
static FastqTracker processFastqs(@NotNull final Multimap<String, File> fastqsPerSample, final int threadCount)
        throws InterruptedException {
    LOGGER.info("Using " + threadCount + " threads. Processing " + fastqsPerSample.size() + " fastQ files.");
    final FastqTrackerWrapper tracker = new FastqTrackerWrapper();
    final ListeningExecutorService threadPool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(threadCount));

    for (final String sampleName : fastqsPerSample.keySet()) {
        final Collection<File> fastqs = fastqsPerSample.get(sampleName);
        for (final File fastq : fastqs) {
            final String laneName = getLaneName(fastq);
            final ListenableFuture<FastqData> futureResult = threadPool.submit(() -> processFile(fastq));
            addCallback(futureResult, (data) -> tracker.addDataFromSampleFile(sampleName, laneName, data),
                    (error) -> LOGGER.error("Failed to process file: " + fastq.getName(), error));
        }
    }
    threadPool.shutdown();
    threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
    return tracker.tracker();
}

@Inject
protected DimensionDataCloudControllerComputeService(ComputeServiceContext context, Map<String, Credentials> credentialStore,
                                                     @Memoized Supplier<Set<? extends Image>> images, @Memoized Supplier<Set<? extends Hardware>> sizes,
                                                     @Memoized Supplier<Set<? extends Location>> locations, ListNodesStrategy listNodesStrategy,
                                                     GetImageStrategy getImageStrategy, GetNodeMetadataStrategy getNodeMetadataStrategy,
                                                     CreateNodesInGroupThenAddToSet runNodesAndAddToSetStrategy, RebootNodeStrategy rebootNodeStrategy,
                                                     DestroyNodeStrategy destroyNodeStrategy, ResumeNodeStrategy startNodeStrategy,
                                                     SuspendNodeStrategy stopNodeStrategy, Provider<TemplateBuilder> templateBuilderProvider,
                                                     @Named("DEFAULT") Provider<TemplateOptions> templateOptionsProvider,
                                                     @Named(TIMEOUT_NODE_RUNNING) Predicate<AtomicReference<NodeMetadata>> nodeRunning,
                                                     @Named(TIMEOUT_NODE_TERMINATED) Predicate<AtomicReference<NodeMetadata>> nodeTerminated,
                                                     @Named(TIMEOUT_NODE_SUSPENDED) Predicate<AtomicReference<NodeMetadata>> nodeSuspended,
                                                     InitializeRunScriptOnNodeOrPlaceInBadMap.Factory initScriptRunnerFactory,
                                                     RunScriptOnNode.Factory runScriptOnNodeFactory, InitAdminAccess initAdminAccess,
                                                     PersistNodeCredentials persistNodeCredentials, Timeouts timeouts,
                                                     @Named(Constants.PROPERTY_USER_THREADS) ListeningExecutorService userExecutor,
                                                     CleanupServer cleanupServer,
                                                     Optional<ImageExtension> imageExtension,
                                                     Optional<SecurityGroupExtension> securityGroupExtension) {
   super(context, credentialStore, images, sizes, locations, listNodesStrategy, getImageStrategy,
            getNodeMetadataStrategy, runNodesAndAddToSetStrategy, rebootNodeStrategy, destroyNodeStrategy,
            startNodeStrategy, stopNodeStrategy, templateBuilderProvider, templateOptionsProvider, nodeRunning,
            nodeTerminated, nodeSuspended, initScriptRunnerFactory, initAdminAccess, runScriptOnNodeFactory,
            persistNodeCredentials, timeouts, userExecutor, imageExtension, securityGroupExtension);
   this.cleanupServer = checkNotNull(cleanupServer, "cleanupServer");
}

@Override
protected void configure() {
    bind(TaskContext.class).annotatedWith(Names.named("rootContext"))
            .toProvider(SeededKeyProvider.<TaskContext>seededKeyProvider())
            .in(ExecuteScoped.class);
    bind(ListeningExecutorService.class).annotatedWith(Names.named("programExecutor"))
            .toProvider(SeededKeyProvider.<ListeningExecutorService>seededKeyProvider())
            .in(ExecuteScoped.class);
    bind(ListeningExecutorService.class).annotatedWith(Names.named("programTimeout"))
            .toProvider(SeededKeyProvider.<ListeningExecutorService>seededKeyProvider())
            .in(ExecuteScoped.class);
    bind(TaskMetricEmitter.class).annotatedWith(Names.named("task"))
            .toProvider(SeededKeyProvider.<TaskMetricEmitter>seededKeyProvider())
            .in(ExecuteScoped.class);
    bind(Tracer.class)
            .toProvider(SeededKeyProvider.<Tracer>seededKeyProvider())
            .in(ExecuteScoped.class);
    bind(String.class).annotatedWith(Names.named("programName"))
            .toProvider(SeededKeyProvider.<String>seededKeyProvider())
            .in(ExecuteScoped.class);
}

@BeforeMethod
public void setUp() {
    _now = new DateTime(2016, 1, 1, 0, 0, DateTimeZone.UTC);

    _clock = mock(Clock.class);
    when(_clock.millis()).thenAnswer(new Answer<Long>() {
        @Override
        public Long answer(InvocationOnMock invocationOnMock) throws Throwable {
            return _now.getMillis();
        }
    });

    _service = mock(ListeningExecutorService.class);
    _delegate = new InMemorySubscriptionDAO(_clock);

    // Insert some test data into the delegate
    for (int i=0; i < 3; i++) {
        _delegate.insertSubscription("owner", "sub" + i, Conditions.alwaysTrue(), Duration.standardDays(1), Duration.standardMinutes(5));
    }
}

@Test
public void testFindAsync() throws ExecutionException, InterruptedException {
    UUID tenantId = UUIDs.timeBased();
    UUID customerId = UUIDs.timeBased();
    Device device = getDevice(tenantId, customerId);
    deviceDao.save(device);

    UUID uuid = device.getId().getId();
    Device entity = deviceDao.findById(uuid);
    assertNotNull(entity);
    assertEquals(uuid, entity.getId().getId());

    ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
    ListenableFuture<Device> future = service.submit(() -> deviceDao.findById(uuid));
    Device asyncDevice = future.get();
    assertNotNull("Async device expected to be not null", asyncDevice);
}

@VisibleForTesting
Map<ArtifactLocation, String> parsePackageStrings(List<ArtifactLocation> sources)
    throws Exception {

  ListeningExecutorService executorService =
      MoreExecutors.listeningDecorator(
          Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));

  Map<ArtifactLocation, ListenableFuture<String>> futures = Maps.newHashMap();
  for (final ArtifactLocation source : sources) {
    futures.put(source, executorService.submit(() -> getDeclaredPackageOfJavaFile(source)));
  }
  Map<ArtifactLocation, String> map = Maps.newHashMap();
  for (Entry<ArtifactLocation, ListenableFuture<String>> entry : futures.entrySet()) {
    String value = entry.getValue().get();
    if (value != null) {
      map.put(entry.getKey(), value);
    }
  }
  return map;
}

/**
 * Tunnelling method so that OperationController with access to the delegate can use the
 * execution thread pool established in the HBaseControl.
 * @param operationExecutable
 * @return
 * @throws UnsupportedOperationException
 */
public ListenableFuture<OpResultSet> execAsync(Callable<OpResultSet> operationExecutable) throws UnsupportedOperationException {
    String logMsg;
    final ListeningExecutorService asyncPool;
    final ListenableFuture<OpResultSet> execTask;
    asyncPool = HBaseControl.this.execPool;
    if (asyncPool == null) {
        logMsg = HBaseControl.class.getSimpleName()
                 + " (context.id='"
                 + HBaseControl.this.context.getId()
                 + "') does not support asynchronous operations";
        throw new UnsupportedOperationException(logMsg);
    }
    execTask = asyncPool.submit(operationExecutable);
    return execTask;
}

@Test
public void sValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final Sha256Hash hash = Sha256Hash.create(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECKey.ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.sign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
    final ECKey.ECDSASignature duplicate = new ECKey.ECDSASignature(sigs.get(0).r, sigs.get(0).s);
    assertEquals(sigs.get(0), duplicate);
    assertEquals(sigs.get(0).hashCode(), duplicate.hashCode());
}

@Test
public void sValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final Sha256Hash hash = Sha256Hash.create(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECKey.ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.sign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
    final ECKey.ECDSASignature duplicate = new ECKey.ECDSASignature(sigs.get(0).r, sigs.get(0).s);
    assertEquals(sigs.get(0), duplicate);
    assertEquals(sigs.get(0).hashCode(), duplicate.hashCode());
}

public BatchExecutor(
    BigtableClient client,
    BigtableOptions options,
    TableMetadataSetter tableMetadataSetter,
    ListeningExecutorService service,
    OperationAdapter<Get, ReadRowsRequest.Builder> getAdapter,
    OperationAdapter<Put, MutateRowRequest.Builder> putAdapter,
    OperationAdapter<Delete, MutateRowRequest.Builder> deleteAdapter,
    RowMutationsAdapter rowMutationsAdapter,
    AppendAdapter appendAdapter,
    IncrementAdapter incrementAdapter,
    ResponseAdapter<com.google.bigtable.v1.Row, Result> rowToResultAdapter) {
  this.client = client;
  this.options = options;
  this.tableMetadataSetter = tableMetadataSetter;
  this.service = service;
  this.getAdapter = getAdapter;
  this.putAdapter = putAdapter;
  this.deleteAdapter = deleteAdapter;
  this.rowMutationsAdapter = rowMutationsAdapter;
  this.appendAdapter = appendAdapter;
  this.incrementAdapter = incrementAdapter;
  this.rowToResultAdapter = rowToResultAdapter;
  rowResultConverter = new RowResultConverter(rowToResultAdapter);
}

@Provides
@Singleton
@ChangeUpdateExecutor
public ListeningExecutorService createChangeUpdateExecutor(@GerritServerConfig Config config) {
  int poolSize = config.getInt("receive", null, "changeUpdateThreads", 1);
  if (poolSize <= 1) {
    return MoreExecutors.newDirectExecutorService();
  }
  return MoreExecutors.listeningDecorator(
      MoreExecutors.getExitingExecutorService(
          new ThreadPoolExecutor(
              1,
              poolSize,
              10,
              TimeUnit.MINUTES,
              new ArrayBlockingQueue<Runnable>(poolSize),
              new ThreadFactoryBuilder().setNameFormat("ChangeUpdate-%d").setDaemon(true).build(),
              new ThreadPoolExecutor.CallerRunsPolicy())));
}

@Provides
@Singleton
@IndexExecutor(INTERACTIVE)
ListeningExecutorService getInteractiveIndexExecutor(
    @GerritServerConfig Config config, WorkQueue workQueue) {
  if (interactiveExecutor != null) {
    return interactiveExecutor;
  }
  int threads = this.threads;
  if (threads <= 0) {
    threads = config.getInt("index", null, "threads", 0);
  }
  if (threads <= 0) {
    threads = Runtime.getRuntime().availableProcessors() / 2 + 1;
  }
  return MoreExecutors.listeningDecorator(workQueue.createQueue(threads, "Index-Interactive"));
}

@AssistedInject
ChangeIndexer(
    @GerritServerConfig Config cfg,
    SchemaFactory<ReviewDb> schemaFactory,
    NotesMigration notesMigration,
    ChangeNotes.Factory changeNotesFactory,
    ChangeData.Factory changeDataFactory,
    ThreadLocalRequestContext context,
    DynamicSet<ChangeIndexedListener> indexedListeners,
    StalenessChecker stalenessChecker,
    @IndexExecutor(BATCH) ListeningExecutorService batchExecutor,
    @Assisted ListeningExecutorService executor,
    @Assisted ChangeIndex index) {
  this.executor = executor;
  this.schemaFactory = schemaFactory;
  this.notesMigration = notesMigration;
  this.changeNotesFactory = changeNotesFactory;
  this.changeDataFactory = changeDataFactory;
  this.context = context;
  this.indexedListeners = indexedListeners;
  this.stalenessChecker = stalenessChecker;
  this.batchExecutor = batchExecutor;
  this.autoReindexIfStale = autoReindexIfStale(cfg);
  this.index = index;
  this.indexes = null;
}

@AssistedInject
ChangeIndexer(
    SchemaFactory<ReviewDb> schemaFactory,
    @GerritServerConfig Config cfg,
    NotesMigration notesMigration,
    ChangeNotes.Factory changeNotesFactory,
    ChangeData.Factory changeDataFactory,
    ThreadLocalRequestContext context,
    DynamicSet<ChangeIndexedListener> indexedListeners,
    StalenessChecker stalenessChecker,
    @IndexExecutor(BATCH) ListeningExecutorService batchExecutor,
    @Assisted ListeningExecutorService executor,
    @Assisted ChangeIndexCollection indexes) {
  this.executor = executor;
  this.schemaFactory = schemaFactory;
  this.notesMigration = notesMigration;
  this.changeNotesFactory = changeNotesFactory;
  this.changeDataFactory = changeDataFactory;
  this.context = context;
  this.indexedListeners = indexedListeners;
  this.stalenessChecker = stalenessChecker;
  this.batchExecutor = batchExecutor;
  this.autoReindexIfStale = autoReindexIfStale(cfg);
  this.index = null;
  this.indexes = indexes;
}

@Inject
ReindexAfterRefUpdate(
    @GerritServerConfig Config cfg,
    OneOffRequestContext requestContext,
    Provider<InternalChangeQuery> queryProvider,
    ChangeIndexer.Factory indexerFactory,
    ChangeIndexCollection indexes,
    ChangeNotes.Factory notesFactory,
    AllUsersName allUsersName,
    AccountCache accountCache,
    @IndexExecutor(QueueType.BATCH) ListeningExecutorService executor) {
  this.requestContext = requestContext;
  this.queryProvider = queryProvider;
  this.indexerFactory = indexerFactory;
  this.indexes = indexes;
  this.notesFactory = notesFactory;
  this.allUsersName = allUsersName;
  this.accountCache = accountCache;
  this.executor = executor;
  this.enabled = cfg.getBoolean("index", null, "reindexAfterRefUpdate", true);
}

@Inject
AllChangesIndexer(
    SchemaFactory<ReviewDb> schemaFactory,
    ChangeData.Factory changeDataFactory,
    GitRepositoryManager repoManager,
    @IndexExecutor(BATCH) ListeningExecutorService executor,
    ChangeIndexer.Factory indexerFactory,
    ChangeNotes.Factory notesFactory,
    ProjectCache projectCache) {
  this.schemaFactory = schemaFactory;
  this.changeDataFactory = changeDataFactory;
  this.repoManager = repoManager;
  this.executor = executor;
  this.indexerFactory = indexerFactory;
  this.notesFactory = notesFactory;
  this.projectCache = projectCache;
}

@Test
public void sValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final Sha256Hash hash = Sha256Hash.create(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECKey.ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.sign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
    final ECKey.ECDSASignature duplicate = new ECKey.ECDSASignature(sigs.get(0).r, sigs.get(0).s);
    assertEquals(sigs.get(0), duplicate);
    assertEquals(sigs.get(0).hashCode(), duplicate.hashCode());
}

public PreviewTableCache(final QueryRunner.QueryRunnerFactory queryRunnerFactory,
                         final Duration previewCacheLifetime,
                         final ExecutorService executor,
                         final int previewLimit)
{
    this.queryRunnerFactory = checkNotNull(queryRunnerFactory, "queryRunnerFactory session was null!");

    ListeningExecutorService listeningExecutor = MoreExecutors.listeningDecorator(executor);

    BackgroundCacheLoader<PartitionedTableWithValue, List<List<Object>>> tableLoader =
            new BackgroundCacheLoader<PartitionedTableWithValue, List<List<Object>>>(listeningExecutor)
    {
                @Override
                public List<List<Object>> load(PartitionedTableWithValue key)
                        throws Exception
                {
                    return queryRows(buildQueryWithLimit(key, previewLimit));
                }
    };

    this.previewTableCache = CacheBuilder.newBuilder()
                                        .expireAfterWrite(Math.round(previewCacheLifetime.getValue()),
                                                          previewCacheLifetime.getUnit())
                                        .maximumSize(previewLimit)
                                        .build(tableLoader);
}

@SuppressWarnings("unchecked")
public void handle(List<SequencePair> pairs)
{
    NodeManagerFactory.getManagerReference();

    ListeningExecutorService executor = DynamicExecutors.newListeningDynamicScalingThreadPool("score-pairs-%d", pairs.size());

    List<Callable<double[]>> tasks = Lists.transform(pairs, new Function<SequencePair, Callable<double[]>>()
    {
        @Override
        @Nullable
        public Callable<double[]> apply(@Nullable SequencePair input)
        {
            return new LocalPairsScoreCallable(input);
        }
    });

    List<double[]> scores = Futures2.invokeAllAndShutdownWhenFinish(tasks, executor, new FutureCallback[0]);
    System.out.println(scores.size() == pairs.size());
}

@Test
public void sValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final Sha256Hash hash = Sha256Hash.create(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECKey.ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.sign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
}

@Test
public void sValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final Sha256Hash hash = Sha256Hash.create(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECKey.ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.sign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
}

@Test
public void sValue() throws Exception {
    // Check that we never generate an S value that is larger than half the curve order. This avoids a malleability
    // issue that can allow someone to change a transaction [hash] without invalidating the signature.
    final int ITERATIONS = 10;
    ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(ITERATIONS));
    List<ListenableFuture<ECKey.ECDSASignature>> sigFutures = Lists.newArrayList();
    final ECKey key = new ECKey();
    for (byte i = 0; i < ITERATIONS; i++) {
        final Sha256Hash hash = Sha256Hash.create(new byte[]{i});
        sigFutures.add(executor.submit(new Callable<ECKey.ECDSASignature>() {
            @Override
            public ECKey.ECDSASignature call() throws Exception {
                return key.sign(hash);
            }
        }));
    }
    List<ECKey.ECDSASignature> sigs = Futures.allAsList(sigFutures).get();
    for (ECKey.ECDSASignature signature : sigs) {
        assertTrue(signature.s.compareTo(ECKey.HALF_CURVE_ORDER) <= 0);
    }
}

@Test
public void testIter() {
    List<ListenableFuture<String>> results = new ArrayList<ListenableFuture<String>>();

    ListeningExecutorService es = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));
    results.add(es.submit(new TestCall("cat")));
    results.add(es.submit(new TestCall("dog")));
    results.add(es.submit(new TestCall("blah")));
    results.add(es.submit(new TestCall("hello")));
    results.add(es.submit(new TestCall("goodbye")));
    results.add(es.submit(new TestCall("woohoo")));

    List<String> fList = new FutureList<String>(results);
    for (String s : fList) {
        System.out.println("Item : "+s);
    }
}

@Test
public void testIter2() {
    FutureList<Integer> futures = new FutureList<Integer>();

    Random rand = new Random(System.currentTimeMillis());
    int number = 20;
    ListeningExecutorService es = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(number));
    for (int i=0;i<number;i++) {
        int sleepTime = rand.nextInt(100)*100;
        System.out.println("Submitting: "+sleepTime);

        futures.addFuture(es.submit(new SleeperCall(sleepTime)));
    }

    // Add some raw values that I do not need to compute
    futures.add(-3);
    futures.add(-1);
    futures.add(-2);

    for (Integer s : futures) {
        System.out.println("Item : "+s);
    }
}

public AlignModel train(List<InputRecord> records, ProbTable labelledExamples) {
  ListeningExecutorService service = listeningDecorator(newCachedThreadPool());
  try {
    this.labelledProbs = labelledExamples.makeNormalizedCopy();
    initCounts(records);
    maximization(); // this just initializes the probabilities for the first time

    int iteration = 0;
    boolean keepTraining = true;
    log.info("Starting EM rounds...");
    while (keepTraining) {
      iteration += 1;

      expectation(records, service);
      double thisChange = maximization();

      keepTraining = !hasConverged(thisChange, iteration);
      log.info("Completed EM round " + iteration + " mass delta " + String.format("%.15f", thisChange));
    }
    log.info("Training complete in " + iteration + " rounds!");
    return new AlignModel(gramOpts, probs);
  } finally {
    MoreExecutors.shutdownAndAwaitTermination(service, 60, TimeUnit.SECONDS);
  }
}