import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public void markGlobalFailure() {
  if (! markedFailure.get() && markedFailure.compareAndSet(false, true)) {
    synchronized (failureListeners) {
      for (Runnable r :  failureListeners) {
        ExceptionUtils.runRunnable(r);
      }
      failureListeners.clear();
    }
    List<ListenableFuture<StepResult>> futures = this.futures.get();
    if (futures != null) {
      // try to short cut any steps we can
      // Sadly this is a duplicate from other cancels, but since we are not garunteed to be 
      // able to cancel here, we still need those points
      FutureUtils.cancelIncompleteFutures(futures, true);
    }
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
private void doClientWrite(boolean doLocal) {
  if(isClosed()) {
    return;
  }
  int wrote = 0;
  try {
    wrote = channel.write(getWriteBuffer());
    if(wrote > 0) {
      reduceWrite(wrote);
      se.addWriteAmount(wrote);
    }
    if(!doLocal) {
      se.setClientOperations(TCPClient.this);
    }
  } catch(Exception e) {
    ExceptionUtils.handleException(e);
    close();
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public int read() throws IOException {
  if(isClosed) {
    return -1;
  }
  synchronized(currentBB) {
    while(true) {
      if(currentBB.remaining() > 0) {
        return currentBB.get() & MergedByteBuffers.UNSIGNED_BYTE_MASK;
      } else {
        if(c.getReadBufferSize() > 0) {
          currentBB.add(c.getRead());
        } else if(isClosed) {
          return -1;
        } else {
          try {
            currentBB.wait(1000);
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            ExceptionUtils.handleException(e);
          }
        }
      }
    }
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
protected void shutdownService() {
  commonSelector.wakeup();
  for(final Client client: clients.values()) {
    IOUtils.closeQuietly(client);
  }
  for(final Server server: servers.values()) {
    IOUtils.closeQuietly(server);
  }
  if(commonSelector != null && commonSelector.isOpen()) {
    closeSelector(schedulerPool, commonSelector);
  }
  while(localNoThreadScheduler.hasTaskReadyToRun()) {
    try {
      localNoThreadScheduler.tick(null);
    } catch(Exception e) {
      ExceptionUtils.handleException(e);
    }
  }
  clients.clear();
  servers.clear();
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
private static BenchmarkResult runBenchmark(String classpath, 
                                            Class<? extends AbstractBenchmark> benchmarkClass, 
                                            String executionArgs) {
  String[] command = {SHELL, "-c", 
                      JAVA_EXECUTE_CMD + "-cp " + classpath + ' ' + 
                        benchmarkClass.getName() + ' ' + executionArgs};
  System.gc();
  try {
    ExecResult runResult = runCommand(command);
    if (StringUtils.isNullOrEmpty(runResult.stdErr)) {
      int delimIndex = runResult.stdOut.indexOf(AbstractBenchmark.OUTPUT_DELIM);
      if (delimIndex > 0) {
        delimIndex += AbstractBenchmark.OUTPUT_DELIM.length();
        long runVal = Long.parseLong(runResult.stdOut.substring(delimIndex));
        return new BenchmarkResult(runVal);
      } else {
        return new BenchmarkResult("Invalid benchmark output: " + runResult.stdOut);
      }
    } else {
      return new BenchmarkResult(runResult.stdErr);
    }
  } catch (Exception e) {
    return new BenchmarkResult(ExceptionUtils.stackToString(e));
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public void run() {
  while (isRunning()) {
    Runnable task = taskQueue.poll();
    // just reset status, we should only shutdown by having the service stopped
    Thread.interrupted();
    if (task != null) {
      if (parked) {
        parked = false;
      }
      ExceptionUtils.runRunnable(task);
    } else if (! parked) {
      // check neighbor worker to see if they need help
      task = checkNeighborWorker.taskQueue.poll();
      if (task != null) {
        ExceptionUtils.runRunnable(task);
      } else {
        parked = true;
      }
    } else {
      LockSupport.park();
    }
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public void run() {
  while (runningThread != null) {
    try {
      T next = getNext();
      
      acceptor.acceptConsumedItem(next);
    } catch (InterruptedException e) {
      stopIfRunning();
      // reset interrupted status
      Thread.currentThread().interrupt();
    } catch (Throwable t) {
      ExceptionUtils.handleException(t);
    }
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
/**
 * Invokes a single listener, if an executor is provided that listener is invoked on that 
 * executor, otherwise it runs in this thread.
 * 
 * @param listener Listener to run
 * @param executor Executor to run listener on, or null to run on calling thread
 * @param throwException {@code true} throws exceptions from runnable, {@code false} handles exceptions
 */
protected void runListener(Runnable listener, Executor executor, boolean throwException) {
  try {
    if (executor != null) {
      executor.execute(listener);
    } else {
      listener.run();
    }
  } catch (Throwable t) {
    if (throwException) {
      throw ExceptionUtils.makeRuntime(t);
    } else {
      ExceptionUtils.handleException(t);
    }
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Test
public void setExceptionHandlerTest() {
  TestExceptionHandler teh = new TestExceptionHandler();
  ConfigurableThreadFactory ctf = new ConfigurableThreadFactory(teh);
  
  assertEquals(teh, ctf.defaultThreadlyExceptionHandler);
  
  final AtomicReference<ExceptionHandler> ehi = new AtomicReference<>(null);
  TestRunnable tr = new TestRunnable() {
    @Override
    public void handleRunStart() {
      ehi.set(ExceptionUtils.getExceptionHandler());
    }
  };
  Thread t = ctf.newThread(tr);
  t.start();
  tr.blockTillFinished();
  
  assertTrue(ehi.get() == teh);
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Test
public void throwableFailureTest() throws InterruptedException {
  Throwable failure = new Throwable();
  ListenableFutureAdapterTask<Object> adapter = 
      new ListenableFutureAdapterTask<>(FutureUtils.immediateFailureFuture(failure));
  
  adapter.run();

  assertTrue(adapter.isDone());
  try {
    adapter.get();
    fail("Exception shuld have thrown");
  } catch (ExecutionException e) {
    assertTrue(failure == ExceptionUtils.getRootCause(e));
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Test
public void taskExceptionTest() {
  Integer key = 1;
  TestExceptionHandler teh = new TestExceptionHandler();
  final RuntimeException testException = new SuppressedStackRuntimeException();
  ExceptionUtils.setDefaultExceptionHandler(teh);
  TestRunnable exceptionRunnable = new TestRuntimeFailureRunnable(testException);
  TestRunnable followRunnable = new TestRunnable();
  distributor.execute(key, exceptionRunnable);
  distributor.execute(key, followRunnable);
  exceptionRunnable.blockTillFinished();
  followRunnable.blockTillStarted();  // verify that it ran despite the exception
  
  assertEquals(1, teh.getCallCount());
  assertEquals(testException, teh.getLastThrowable());
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Test
public void signalCompleteAnotherThreadTest() {
  TestRunnable waitRunnable = new TestRunnable() {
    @Override
    public void handleRunFinish() {
      try {
        verifier.waitForTest();
      } catch (Exception e) {
        throw ExceptionUtils.makeRuntime(e);
      }
    }
  };
  new Thread(waitRunnable).start();
  
  // should unblock thread
  verifier.signalComplete();
  
  waitRunnable.blockTillFinished(); // should return quickly
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
private Future<?> handleFiles(final List<File> fileList) {
  return scheduler.submit(new Runnable() {
    @Override
    public void run() {
      Iterator<File> it = fileList.iterator();
      while (it.hasNext()) {
        File f = it.next();
        Iterator<FileListenerInterface> lIt = listeners.iterator();
        while (lIt.hasNext()) {
          try {
            lIt.next().handleFile(f);
          } catch (Exception e) {
            ExceptionUtils.handleException(e);
          }
        }
      }
    }
  });
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public void handleFile(File file) {
  try {
    DigestResult dr = makeFileDigest(file);
    
    // duplicates should be rare, so we assume there is none
    List<File> dupFiles = new LinkedList<File>();
    List<File> existingList = digestToFile.putIfAbsent(dr, dupFiles);
    if (existingList != null) {
      dupFiles = existingList;
    }
    dupFiles.add(file);
    fileToDigest.put(file, dr);
  } catch (Exception e) {
    if (e instanceof FileNotFoundException) {
      // ignore
    } else {
      ExceptionUtils.handleException(e);
    }
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
public static boolean fileInArray(File[] fileArray, File searchFile) {
  for (int i = 0; i < fileArray.length; i++) {
    try {
      if (fileArray[i].getCanonicalPath().equals(searchFile.getCanonicalPath())) {
        return true;
      }
    } catch (IOException e) {
      ExceptionUtils.handleException(e);
      
      if (fileArray[i].getAbsolutePath().equals(searchFile.getAbsolutePath())) {
        return true;
      }
    }
  }
  
  return false;
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public void setStartHandler(StepStartHandler handler) {
  if (this.handler != null && handler != null) {
    // TODO - do we need to handle multiple start handlers?  The cost is memory
    ExceptionUtils.handleException(new IllegalStateException("Pre-run condition already set"));
  }
  this.handler = handler;
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
/**
 * Sets up a default {@link ExceptionHandlerInterface} so that if any uncaught exceptions occur, 
 * the script will display the exception and exit.  There should never be any uncaught 
 * exceptions, this likely would indicate a bug in Ambush. 
 */
protected static void setupExceptionHandler() {
  ExceptionUtils.setDefaultExceptionHandler(new ExceptionHandler() {
    @Override
    public void handleException(Throwable thrown) {
      synchronized (this) { // synchronized to prevent terminal corruption from multiple failures
        System.err.println("Unexpected uncaught exception: ");
        printFailureAndExit(thrown);
      }
    }
  });
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public void registerFailureNotification(Runnable listener) {
  synchronized (failureListeners) {
    if (markedFailure.get()) {
      ExceptionUtils.runRunnable(listener);
    } else {
      failureListeners.add(listener);
    }
  }
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public void hasError(Throwable t) {
  ExceptionUtils.handleException(t);
  bodyFuture.completed(hr, responseWriter);
  bodyFuture = new BodyFuture();
  responseWriter = new ResponseWriter(this.client);
  
}
 

import org.threadly.util.ExceptionUtils; //导入依赖的package包/类
@Override
public void run() {
  try {
    doParse();
  } catch (Exception e) {
    throw ExceptionUtils.makeRuntime(e);
  }
}