public void doMonitor() {
    while (running) {
      String[] keys = null;
      // Make a copy of the names of the current jobs.
      synchronized(jobs) {
        keys = jobs.keySet().toArray(new String[0]);
      }

      // Check all the jobs. We do not want to block access to `jobs`
      // because that will prevent new jobs from being added.
      // This is safe because JobMonitor.run is the only code that can
      // remove a job from `jobs`. Thus all elements in `keys` will have
      // valid values.
      Map<String, List<DistRaid>> finishedJobs =
        new HashMap<String, List<DistRaid>>();

      for (String key: keys) {
        // For each policy being monitored, get the list of jobs running.
        DistRaid[] jobListCopy = null;
        synchronized(jobs) {
          List<DistRaid> jobList = jobs.get(key);
          synchronized(jobList) {
            jobListCopy = jobList.toArray(new DistRaid[jobList.size()]);
          }
        }
        // The code that actually contacts the JobTracker is not synchronized,
        // it uses copies of the list of jobs.
        for (DistRaid job: jobListCopy) {
          // Check each running job.
          try {
            boolean complete = job.checkComplete();
            if (complete) {
              addJob(finishedJobs, key, job);
              if (job.successful()) {
                jobsSucceeded++;
              }
            } else if (System.currentTimeMillis() -
                job.getStartTime() > maximumRunningTime){
              // If the job is running for more than one day
              throw new Exception("Job " + job.getJobID() + 
                  " is hanging more than " + maximumRunningTime/1000
                  + " seconds. Kill it");
            }
          } catch (Exception e) {
            // If there was an error, consider the job finished.
            addJob(finishedJobs, key, job);
            try {
              job.killJob();
            } catch (Exception ee) {
              LOG.error(ee);
            }
          }
        }
      }

      if (finishedJobs.size() > 0) {
        for (String key: finishedJobs.keySet()) {
          List<DistRaid> finishedJobList = finishedJobs.get(key);
          // Iterate through finished jobs and remove from jobs.
          // removeJob takes care of locking.
          for (DistRaid job: finishedJobList) {
            addCounter(raidProgress, job, INT_CTRS);
            removeJob(jobs, key, job);
            addJob(history, key, job);
            // delete the temp directory 
            job.cleanUp();
          }
        }
      }

      try {
        Thread.sleep(jobMonitorInterval);
      } catch (InterruptedException ie) {
      }
    }
  }
public void doMonitor ( ) { while ( running ) { String [ ] keys = null ; synchronized ( jobs ) { keys = jobs . keySet ( ) . toArray ( new String [ 0 ] ) ; } Map < String , List < DistRaid > > finishedJobs = new HashMap < String , List < DistRaid > > ( ) ; for ( String key : keys ) { DistRaid [ ] jobListCopy = null ; synchronized ( jobs ) { List < DistRaid > jobList = jobs . get ( key ) ; synchronized ( jobList ) { jobListCopy = jobList . toArray ( new DistRaid [ jobList . size ( ) ] ) ; } } for ( DistRaid job : jobListCopy ) { try { boolean complete = job . checkComplete ( ) ; if ( complete ) { addJob ( finishedJobs , key , job ) ; if ( job . successful ( ) ) { jobsSucceeded ++ ; } } else if ( System . currentTimeMillis ( ) - job . getStartTime ( ) > maximumRunningTime ) { throw new Exception ( "Job " + job . getJobID ( ) + " is hanging more than " + maximumRunningTime / 1000 + " seconds. Kill it" ) ; } } catch ( Exception e ) { addJob ( finishedJobs , key , job ) ; try { job . killJob ( ) ; } catch ( Exception ee ) { LOG . error ( ee ) ; } } } } if ( finishedJobs . size ( ) > 0 ) { for ( String key : finishedJobs . keySet ( ) ) { List < DistRaid > finishedJobList = finishedJobs . get ( key ) ; for ( DistRaid job : finishedJobList ) { addCounter ( raidProgress , job , INT_CTRS ) ; removeJob ( jobs , key , job ) ; addJob ( history , key , job ) ; job . cleanUp ( ) ; } } } try { Thread . sleep ( jobMonitorInterval ) ; } catch ( InterruptedException ie ) { } } }
Periodically checks status of running map-reduce jobs.