hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From maha...@apache.org
Subject svn commit: r984533 - in /hadoop/mapreduce/trunk: ./ src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/ src/contrib/mumak/ src/contrib/mumak/bin/ src/contrib/mumak/conf/ src/contrib/mumak/src/java/org/apache/hadoop/mapred/ src/contrib/mu...
Date Wed, 11 Aug 2010 18:55:28 GMT
Author: mahadev
Date: Wed Aug 11 18:55:28 2010
New Revision: 984533

URL: http://svn.apache.org/viewvc?rev=984533&view=rev
Log:
MAPREDUCE-1523. Making Mumak work with Capacity-Scheduler (Anirban Das via mahadev)

Added:
    hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/JobInitializationPollerAspects.aj
    hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorCSJobInitializationThread.java
    hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorThreadWakeUpEvent.java
Modified:
    hadoop/mapreduce/trunk/CHANGES.txt
    hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java
    hadoop/mapreduce/trunk/src/contrib/mumak/bin/mumak.sh
    hadoop/mapreduce/trunk/src/contrib/mumak/build.xml
    hadoop/mapreduce/trunk/src/contrib/mumak/conf/mumak.xml
    hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorEngine.java
    hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorJobTracker.java
    hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/MockSimulatorEngine.java
    hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorEndToEnd.java
    hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorJobTracker.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java
    hadoop/mapreduce/trunk/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java

Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Wed Aug 11 18:55:28 2010
@@ -112,6 +112,9 @@ Trunk (unreleased changes)
     TestTrackerDistributedCacheManagerWithLinuxTaskController. (Devaraj Das
     via amareshwari)
 
+    MAPREDUCE-1523. Making Mumak work with Capacity-Scheduler (Anirban Das
+    via mahadev)
+
   OPTIMIZATIONS
 
     MAPREDUCE-1354. Enhancements to JobTracker for better performance and

Modified: hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java
(original)
+++ hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/JobInitializationPoller.java
Wed Aug 11 18:55:28 2010
@@ -634,4 +634,12 @@ public class JobInitializationPoller ext
   Set<JobID> getInitializedJobList() {
     return initializedJobs.keySet();
   }
+
+  public HashMap<String, JobInitializationThread> getThreadsToQueueMap() {
+    return threadsToQueueMap;
+  }
+  
+  public long getSleepInterval(){
+    return sleepInterval;
+  }
 }

Modified: hadoop/mapreduce/trunk/src/contrib/mumak/bin/mumak.sh
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/bin/mumak.sh?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/bin/mumak.sh (original)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/bin/mumak.sh Wed Aug 11 18:55:28 2010
@@ -172,7 +172,7 @@ function print_usage(){
   echo "Usage: $script [--config dir] trace.json topology.json"
 }
 
-if [ $# != 2 ]; then
+if [ $# <= 2 ]; then
   print_usage
   exit
 fi

Modified: hadoop/mapreduce/trunk/src/contrib/mumak/build.xml
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/build.xml?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/build.xml (original)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/build.xml Wed Aug 11 18:55:28 2010
@@ -32,6 +32,11 @@ to call at top-level: ant deploy-contrib
   <import file="../build-contrib.xml"/>
   <property name="mumak.stamp.file" value="${build.dir}/mumak.uptodate.stamp"/>
 
+  <path id="mumak-classpath">
+    <pathelement location="${hadoop.root}/build/contrib/capacity-scheduler/classes"/>
+    <path refid="contrib-classpath"/>
+  </path>
+
   <target name="compile-java-sources" depends="init, ivy-retrieve-common" unless="skip.contrib">
     <echo message="contrib: ${name}"/>
     <javac
@@ -41,7 +46,7 @@ to call at top-level: ant deploy-contrib
     destdir="${build.classes}"
     debug="${javac.debug}"
     deprecation="${javac.deprecation}">
-    <classpath refid="contrib-classpath"/>
+    <classpath refid="mumak-classpath"/>
     </javac>
   </target>
 
@@ -59,7 +64,7 @@ to call at top-level: ant deploy-contrib
     <echo message="Start weaving aspects in place"/>
     <iajc
       encoding="${build.encoding}" 
-      srcdir="${hadoop.root}/src/java/;${hadoop.root}/build/src/;${src.dir}"
+      srcdir="${hadoop.root}/src/java/;${hadoop.root}/src/contrib/capacity-scheduler/src/java/;${hadoop.root}/build/src/;${src.dir}"
       includes="org/apache/hadoop/**/*.java, org/apache/hadoop/**/*.aj"
       destDir="${build.classes}"
       debug="${javac.debug}"
@@ -67,7 +72,7 @@ to call at top-level: ant deploy-contrib
       source="${javac.version}"
       fork="yes"
       deprecation="${javac.deprecation}">
-      <classpath refid="contrib-classpath"/>
+      <classpath refid="mumak-classpath"/>
     </iajc>
     <touch file="${mumak.stamp.file}" mkdirs="true" />
     <echo message="Weaving of aspects is finished"/>

Modified: hadoop/mapreduce/trunk/src/contrib/mumak/conf/mumak.xml
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/conf/mumak.xml?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/conf/mumak.xml (original)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/conf/mumak.xml Wed Aug 11 18:55:28 2010
@@ -34,4 +34,11 @@
   node-local</description>
 </property>
 
+<property>
+  <name>mumak.job-submission.policy</name>
+  <value>REPLAY</value>
+  <description>Job submission policy
+  </description>
+</property>
+
 </configuration>

Added: hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/JobInitializationPollerAspects.aj
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/JobInitializationPollerAspects.aj?rev=984533&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/JobInitializationPollerAspects.aj
(added)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/JobInitializationPollerAspects.aj
Wed Aug 11 18:55:28 2010
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapred;
+
+/**
+ * This aspect is used for NOT starting the job initialization threads of the capacity scheduler.
+ * We schedule the body of these threads manually from the SimulatorJobTracker according
to 
+ * simulation time.
+ */
+ 
+public aspect JobInitializationPollerAspects {
+
+  pointcut overrideInitializationThreadStarts () : 
+    (target (JobInitializationPoller) || 
+     target (JobInitializationPoller.JobInitializationThread)) && 
+    call (public void start());
+  
+  void around() : overrideInitializationThreadStarts () {
+    // no-op
+  }
+  
+  
+}

Added: hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorCSJobInitializationThread.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorCSJobInitializationThread.java?rev=984533&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorCSJobInitializationThread.java
(added)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorCSJobInitializationThread.java
Wed Aug 11 18:55:28 2010
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapred.JobInitializationPoller.JobInitializationThread;
+
+public class SimulatorCSJobInitializationThread implements SimulatorEventListener {
+
+  long lastCalled;
+  CapacityTaskScheduler taskScheduler;
+  JobInitializationPoller jobPoller;
+  private final String queue;
+  final long sleepInterval;
+  /** The log object to send our messages to; only used for debugging. */
+  private static final Log LOG = LogFactory.getLog(SimulatorCSJobInitializationThread.class);
+
+  public SimulatorCSJobInitializationThread(TaskScheduler taskScheduler, 
+      String queue) {
+    this.taskScheduler = (CapacityTaskScheduler) taskScheduler;
+    jobPoller = this.taskScheduler.getInitializationPoller(); 
+    sleepInterval = jobPoller.getSleepInterval();
+    this.queue = queue;
+  }
+
+  @Override
+  public List<SimulatorEvent> accept(SimulatorEvent event) throws IOException {
+
+    SimulatorThreadWakeUpEvent e;
+    if(event instanceof SimulatorThreadWakeUpEvent) {
+      e = (SimulatorThreadWakeUpEvent) event;
+    }
+    else {
+      throw new IOException("Received an unexpected type of event in " + 
+      "SimThrdCapSchedJobInit");
+    }
+    jobPoller.cleanUpInitializedJobsList();
+    jobPoller.selectJobsToInitialize();
+    JobInitializationThread thread = 
+      jobPoller.getThreadsToQueueMap().get(this.queue);
+    thread.initializeJobs();   	
+    lastCalled = e.getTimeStamp();
+    List<SimulatorEvent> returnEvents = 
+      Collections.<SimulatorEvent>singletonList(
+          new SimulatorThreadWakeUpEvent(this,
+              lastCalled + sleepInterval));
+    return returnEvents;
+  }
+
+  @Override
+  public List<SimulatorEvent> init(long when) throws IOException {
+
+    return Collections.<SimulatorEvent>singletonList(
+        new SimulatorThreadWakeUpEvent(this,
+            when + sleepInterval));
+  }
+
+}

Modified: hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorEngine.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorEngine.java?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorEngine.java
(original)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorEngine.java
Wed Aug 11 18:55:28 2010
@@ -21,10 +21,15 @@ import java.io.IOException;
 import java.io.PrintStream;
 import java.util.List;
 import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.Map;
 import java.util.Iterator;
 import java.util.Random;
 import java.util.regex.Pattern;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
 import org.apache.hadoop.fs.FileSystem;
@@ -68,6 +73,13 @@ public class SimulatorEngine extends Con
   boolean shutdown = false;
   long terminateTime = Long.MAX_VALUE;
   long currentTime;
+  /** The HashSet for storing all the simulated threads useful for 
+   * job initialization for capacity scheduler.
+   */
+  HashSet<SimulatorCSJobInitializationThread> threadSet;
+  /** The log object to send our messages to; only used for debugging. */
+  private static final Log LOG = LogFactory.getLog(SimulatorEngine.class);
+  
   /** 
    * Master random seed read from the configuration file, if present.
    * It is (only) used for creating sub seeds for all the random number 
@@ -137,13 +149,13 @@ public class SimulatorEngine extends Con
   }
    
   /**
-   * Initiate components in the simulation.
-   * @throws InterruptedException
-   * @throws IOException if trace or topology files cannot be open
+   * Creates the configuration for mumak simulation. This is kept modular mostly for 
+   * testing purposes. so that the standard configuration can be modified before passing
+   * it to the init() function.
+   * @return JobConf: the configuration for the SimulatorJobTracker 
    */
-  @SuppressWarnings("deprecation")
-  void init() throws InterruptedException, IOException {
     
+  JobConf createMumakConf() {
     JobConf jobConf = new JobConf(getConf());
     jobConf.setClass("topology.node.switch.mapping.impl",
         StaticMapping.class, DNSToSwitchMapping.class);
@@ -157,7 +169,30 @@ public class SimulatorEngine extends Con
     jobConf.setUser("mumak");
     jobConf.set("mapred.system.dir", 
         jobConf.get("hadoop.log.dir", "/tmp/hadoop-"+jobConf.getUser()) + "/mapred/system");
-    jobConf.set("mapred.jobtracker.taskScheduler", JobQueueTaskScheduler.class.getName());
+    
+    return jobConf;
+  }
+
+  /**
+   * Initialize components in the simulation.
+   * @throws InterruptedException
+   * @throws IOException if trace or topology files cannot be opened.
+   */
+  void init() throws InterruptedException, IOException {
+    
+    JobConf jobConf = createMumakConf();
+    init(jobConf);
+  }
+    
+  /**
+   * Initiate components in the simulation. The JobConf is
+   * create separately and passed to the init().
+   * @param JobConf: The configuration for the jobtracker.
+   * @throws InterruptedException
+   * @throws IOException if trace or topology files cannot be opened.
+   */
+  @SuppressWarnings("deprecation")
+  void init(JobConf jobConf) throws InterruptedException, IOException {
     
     FileSystem lfs = FileSystem.getLocal(getConf());
     Path logPath =
@@ -212,11 +247,40 @@ public class SimulatorEngine extends Con
     jc = new SimulatorJobClient(jt, jobStoryProducer, submissionPolicy);
     queue.addAll(jc.init(firstJobStartTime));
 
+    //if the taskScheduler is CapacityTaskScheduler start off the JobInitialization
+    //threads too
+    if (jobConf.get("mapred.jobtracker.taskScheduler").equals
+       (CapacityTaskScheduler.class.getName())) {
+      LOG.info("CapacityScheduler used: starting simulatorThreads");
+      startSimulatorThreadsCapSched(now);
+    }
     terminateTime = getTimeProperty(jobConf, "mumak.terminate.time",
                                     Long.MAX_VALUE);
   }
   
   /**
+   * In this function, we collect the set of leaf queues from JobTracker, and 
+   * for each of them creates a simulated thread that performs the same
+   * check as JobInitializationPoller.JobInitializationThread in Capacity Scheduler.  
+   * @param now
+   * @throws IOException
+   */
+  private void startSimulatorThreadsCapSched(long now) throws IOException {
+    
+    Set<String> queueNames = jt.getQueueManager().getLeafQueueNames();
+    TaskScheduler taskScheduler = jt.getTaskScheduler();
+    threadSet = new HashSet<SimulatorCSJobInitializationThread>();
+    // We create a different thread for each queue and hold a 
+    //reference to  each of them 
+    for (String jobQueue: queueNames) {
+      SimulatorCSJobInitializationThread capThread = 
+        new SimulatorCSJobInitializationThread(taskScheduler,jobQueue);   
+      threadSet.add(capThread);
+      queue.addAll(capThread.init(now));
+    }
+  }
+  
+  /**
    * The main loop of the simulation. First call init() to get objects ready,
    * then go into the main loop, where {@link SimulatorEvent}s are handled removed from
    * the {@link SimulatorEventQueue}, and new {@link SimulatorEvent}s are created and inserted

Modified: hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorJobTracker.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorJobTracker.java?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorJobTracker.java
(original)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorJobTracker.java
Wed Aug 11 18:55:28 2010
@@ -197,6 +197,24 @@ public class SimulatorJobTracker extends
     SimulatorJobInProgress job = new SimulatorJobInProgress(jobId, jobSubmitDir, this,
                                                             this.conf, 
                                                             jobStory);
+    // Check whether the queue information provided is valid
+    try {
+      checkQueueValidity(job);
+    } catch(IOException ioe) {
+      LOG.warn("Queue given for job " + job.getJobID() + " is not valid:"
+        + ioe);
+      throw ioe;
+    }
+    
+    // Check the job if it cannot run in the cluster because of invalid memory
+    // requirements.
+    try {
+      checkMemoryRequirements(job);
+    } catch (IOException ioe) {
+      LOG.warn("Exception in checking Memory requirements of jobId: " + jobId
+               + ioe);
+      //throw ioe;
+    }
     return addJob(jobId, job);
   }
   
@@ -457,7 +475,10 @@ public class SimulatorJobTracker extends
           }
           SimulatorLaunchTaskAction newlaunchTask = 
         	  new SimulatorLaunchTaskAction(task, taskAttemptInfo);
-          
+          if (loggingEnabled) {
+            LOG.debug("Job " + jobID + " launched taskattempt " +
+                             taskAttemptID + " at " + getClock().getTime());
+          }
           actions.add(newlaunchTask);
         }
       }

Added: hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorThreadWakeUpEvent.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorThreadWakeUpEvent.java?rev=984533&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorThreadWakeUpEvent.java
(added)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/src/java/org/apache/hadoop/mapred/SimulatorThreadWakeUpEvent.java
Wed Aug 11 18:55:28 2010
@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred;
+
+public class SimulatorThreadWakeUpEvent extends SimulatorEvent {
+
+  public SimulatorThreadWakeUpEvent(SimulatorEventListener listener, 
+      long timestamp) {
+    super(listener, timestamp);
+  }
+
+}

Modified: hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/MockSimulatorEngine.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/MockSimulatorEngine.java?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/MockSimulatorEngine.java
(original)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/MockSimulatorEngine.java
Wed Aug 11 18:55:28 2010
@@ -40,8 +40,7 @@ public class MockSimulatorEngine extends
 
   public static final Log LOG = LogFactory.getLog(MockSimulatorEngine.class);
   
-  public MockSimulatorEngine(int nJobs,
-      @SuppressWarnings("unused") int nTrackers) {
+  public MockSimulatorEngine(int nJobs, int nTrackers) {
     super();
     fixedJobs = nJobs;
     jobs = new HashMap<JobID, JobStory>();
@@ -52,7 +51,11 @@ public class MockSimulatorEngine extends
   @Override
   public void run() throws IOException, InterruptedException {
     startTime = System.currentTimeMillis();
-    init();
+    JobConf jobConf = createMumakConf();
+    // Adding the default queue since the example trace is from queue-less hadoop
+    jobConf.set("mapred.queue.names",JobConf.DEFAULT_QUEUE_NAME);
+    
+    init(jobConf);
     validateInitialization();
     SimulatorEvent nextEvent;
     while ((nextEvent = queue.get()) != null && nextEvent.getTimeStamp() < terminateTime

Modified: hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorEndToEnd.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorEndToEnd.java?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorEndToEnd.java
(original)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorEndToEnd.java
Wed Aug 11 18:55:28 2010
@@ -56,8 +56,10 @@ public class TestSimulatorEndToEnd {
     
     MockSimulatorEngine mockMumak = new MockSimulatorEngine(numJobs, nTrackers);
 
+    Configuration mumakConf = new Configuration();
+    mumakConf.set("mapred.jobtracker.taskScheduler", JobQueueTaskScheduler.class.getName());
     String[] args = { traceFile.toString(), topologyFile.toString() };
-    int res = ToolRunner.run(conf, mockMumak, args);
+    int res = ToolRunner.run(mumakConf, mockMumak, args);
     Assert.assertEquals(res, 0);
   }
   

Modified: hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorJobTracker.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorJobTracker.java?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorJobTracker.java
(original)
+++ hadoop/mapreduce/trunk/src/contrib/mumak/src/test/org/apache/hadoop/mapred/TestSimulatorJobTracker.java
Wed Aug 11 18:55:28 2010
@@ -57,6 +57,7 @@ public class TestSimulatorJobTracker {
     jtConf.set("mapred.system.dir", jtConf.get("hadoop.tmp.dir", "/tmp/hadoop-"
         + jtConf.getUser())
         + "/mapred/system");
+    jtConf.set("mapred.queue.names",JobConf.DEFAULT_QUEUE_NAME);
     System.out.println("Created JobConf");
     return jtConf;
   }

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java Wed Aug 11 18:55:28
2010
@@ -3049,14 +3049,11 @@ public class JobTracker implements MRCon
         new JobInProgress(this, this.conf, restartCount, jobInfo, ts);
 
     synchronized (this) {
-      String queue = job.getProfile().getQueueName();
-      if (!(queueManager.getLeafQueueNames().contains(queue))) {
-        throw new IOException("Queue \"" + queue + "\" does not exist");
-      }
-
-      // check if queue is RUNNING
-      if (!queueManager.isRunning(queue)) {
-        throw new IOException("Queue \"" + queue + "\" is not running");
+      try {
+        checkQueueValidity(job);
+      } catch(IOException ioe) {
+        LOG.error("Queue given for job " + job.getJobID() + " is not valid: " + ioe);
+        throw ioe;
       }
       try {
         checkAccess(job, ugi, Queue.QueueOperation.SUBMIT_JOB, null);
@@ -3128,6 +3125,24 @@ public class JobTracker implements MRCon
   }
 
   /**
+   * For a JobInProgress that is being submitted, check whether 
+   * queue that the job has been submitted to exists and is RUNNING.
+   * @param job The JobInProgress object being submitted.
+   * @throws IOException
+   */
+  public void checkQueueValidity(JobInProgress job) throws IOException {
+    String queue = job.getProfile().getQueueName();
+    if (!(queueManager.getLeafQueueNames().contains(queue))) {
+        throw new IOException("Queue \"" + queue + "\" does not exist");
+    }
+
+    // check if queue is RUNNING
+    if (!queueManager.isRunning(queue)) {
+        throw new IOException("Queue \"" + queue + "\" is not running");
+    }
+  }
+
+  /**
    * Check the ACLs for a user doing the passed queue-operation and the passed
    * job operation.
    * <ul>
@@ -4480,7 +4495,7 @@ public class JobTracker implements MRCon
    * @param job
    * @throws IOException 
    */
-  private void checkMemoryRequirements(JobInProgress job)
+  void checkMemoryRequirements(JobInProgress job)
       throws IOException {
     if (!perTaskMemoryConfigurationSetOnJT()) {
       if (LOG.isDebugEnabled()) {

Modified: hadoop/mapreduce/trunk/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java?rev=984533&r1=984532&r2=984533&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java (original)
+++ hadoop/mapreduce/trunk/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java Wed Aug
11 18:55:28 2010
@@ -745,6 +745,10 @@ public class ZombieJob implements JobSto
       // any locality, so this group should count as "distance=2".
       // However, setup/cleanup tasks are also counted in the 4th group.
       // These tasks do not make sense.
+      if(cdfList==null) {
+    	  runtime = -1;
+    	  return runtime;
+      }
       try {
         runtime = makeUpRuntime(cdfList.get(locality));
       } catch (NoValueToMakeUpRuntime e) {
@@ -767,6 +771,9 @@ public class ZombieJob implements JobSto
    */
   private long makeUpRuntime(List<LoggedDiscreteCDF> mapAttemptCDFs) {
     int total = 0;
+    if(mapAttemptCDFs == null) {
+    	return -1;
+    }
     for (LoggedDiscreteCDF cdf : mapAttemptCDFs) {
       total += cdf.getNumberValues();
     }
@@ -862,6 +869,11 @@ public class ZombieJob implements JobSto
   }
 
   private State makeUpState(int taskAttemptNumber, double[] numAttempts) {
+	  
+  // if numAttempts == null we are returning FAILED.
+  if(numAttempts == null) {
+    return State.FAILED;
+  }
     if (taskAttemptNumber >= numAttempts.length - 1) {
       // always succeed
       return State.SUCCEEDED;



Mime
View raw message