@Override public void run(Context context) throws IOException, InterruptedException {
  outer=context;
  int numberOfThreads=getThreadCount(context);
  mapClass=getMapperClass(context);
  if (LOG.isDebugEnabled()) {
    LOG.debug("Running with " + numberOfThreads + " threads");
  }
  numberOfThreads--;
  InputSplit split=context.getInputSplit();
  try {
    List<Future<?>> taskList=null;
    if (threadPool != null) {
      taskList=new ArrayList<Future<?>>();
synchronized (threadPool) {
        for (int i=0; i < numberOfThreads; ++i) {
          MapRunner runner=new MapRunner();
          BaseMapper<K1,V1,K2,V2> mapper=runner.getMapper();
          if (!threadPool.isShutdown()) {
            Collection<Future<Object>> tasks=mapper.submitTasks(threadPool,split);
            taskList.addAll(tasks);
            numberOfThreads-=tasks.size();
            Future<?> future=threadPool.submit(runner);
            taskList.add(future);
          }
 else {
            throw new InterruptedException("Thread Pool has been shut down");
          }
        }
        threadPool.notify();
      }
      MapRunner r=new MapRunner();
      r.run();
      for (      Future<?> f : taskList) {
        f.get();
      }
    }
 else {
      runners=new ArrayList<MapRunner>(numberOfThreads);
      for (int i=0; i < numberOfThreads; ++i) {
        MapRunner thread;
        thread=new MapRunner();
        thread.start();
        runners.add(i,thread);
      }
      MapRunner r=new MapRunner();
      r.run();
      for (int i=0; i < numberOfThreads; ++i) {
        MapRunner thread=runners.get(i);
        thread.join();
        Throwable th=thread.throwable;
        if (th != null) {
          if (th instanceof IOException) {
            throw (IOException)th;
          }
 else           if (th instanceof InterruptedException) {
            throw (InterruptedException)th;
          }
 else {
            throw new RuntimeException(th);
          }
        }
      }
    }
  }
 catch (  ClassNotFoundException e) {
    LOG.error("MapRunner class not found",e);
  }
catch (  ExecutionException e) {
    LOG.error("Error waiting for MapRunner threads to complete",e);
  }
}

@ Override public void run ( Context context ) throws IOException , InterruptedException { outer = context ; int numberOfThreads = getThreadCount ( context ) ; mapClass = getMapperClass ( context ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Running with " + numberOfThreads + " threads" ) ; } numberOfThreads -- ; InputSplit split = context . getInputSplit ( ) ; try { List < Future < ? > > taskList = null ; if ( threadPool != null ) { taskList = new ArrayList < Future < ? > > ( ) ; synchronized ( threadPool ) { for ( int i = 0 ; i < numberOfThreads ; ++ i ) { MapRunner runner = new MapRunner ( ) ; BaseMapper < K1 , V1 , K2 , V2 > mapper = runner . getMapper ( ) ; if ( ! threadPool . isShutdown ( ) ) { Collection < Future < Object > > tasks = mapper . submitTasks ( threadPool , split ) ; taskList . addAll ( tasks ) ; numberOfThreads -= tasks . size ( ) ; Future < ? > future = threadPool . submit ( runner ) ; taskList . add ( future ) ; } else { throw new InterruptedException ( "Thread Pool has been shut down" ) ; } } threadPool . notify ( ) ; } MapRunner r = new MapRunner ( ) ; r . run ( ) ; for ( Future < ? > f : taskList ) { f . get ( ) ; } } else { runners = new ArrayList < MapRunner > ( numberOfThreads ) ; for ( int i = 0 ; i < numberOfThreads ; ++ i ) { MapRunner thread ; thread = new MapRunner ( ) ; thread . start ( ) ; runners . add ( i , thread ) ; } MapRunner r = new MapRunner ( ) ; r . run ( ) ; for ( int i = 0 ; i < numberOfThreads ; ++ i ) { MapRunner thread = runners . get ( i ) ; thread . join ( ) ; Throwable th = thread . throwable ; if ( th != null ) { if ( th instanceof IOException ) { throw ( IOException ) th ; } else if ( th instanceof InterruptedException ) { throw ( InterruptedException ) th ; } else { throw new RuntimeException ( th ) ; } } } } } catch ( ClassNotFoundException e ) { LOG . error ( "MapRunner class not found" , e ) ; } catch ( ExecutionException e ) { LOG . error ( "Error waiting for MapRunner threads to complete" , e ) ; } }
run the application ' s maps using a thread pool .
