Author: omalley
Date: Tue Mar 8 05:55:58 2011
New Revision: 1079209
URL: http://svn.apache.org/viewvc?rev=1079209&view=rev
Log:
commit eca5731aff949a572ec8a35fab9ab7620249f16d
Author: Greg Roelofs <[email protected]>
Date: Wed Dec 8 18:09:21 2010 -0800
"Version 8b" fixes per Dick's review feedback in BZ 4150548 comment 34:
UberTask comment fixed; Task.commit() split into two pieces, only the
second of which is used by UberTask.
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java?rev=1079209&r1=1079208&r2=1079209&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/Task.java
Tue Mar 8 05:55:58 2011
@@ -885,7 +885,7 @@ abstract public class Task implements Wr
}
}
// wait for commit approval (via JT commitAction for this task) and
commit
- commit(umbilical, reporter, true);
+ commitAfterApproval(umbilical, reporter);
}
taskDone.set(true);
reporter.stopCommunicationThread();
@@ -1004,41 +1004,43 @@ abstract public class Task implements Wr
}
}
- // this is protected (rather than private) solely for UberTask map-only case
- protected void commit(TaskUmbilicalProtocol umbilical,
- TaskReporter reporter,
- boolean queryTTBeforeCommit) throws IOException {
+ private void commitAfterApproval(TaskUmbilicalProtocol umbilical,
+ TaskReporter reporter) throws IOException {
int retries = MAX_RETRIES;
- if (queryTTBeforeCommit) {
- while (true) {
- try {
- while (!umbilical.canCommit(taskIdForUmbilical)) {
- try {
-System.out.println("GRR DEBUG: Task commit(): TT canCommit() returned false;
sleeping 1 sec");
- // FIXME 1: shouldn't this count down retries, too, in case JT
glitched and no longer knows about us? (else infinite loop)
- Thread.sleep(1000); // FIXME 2: shouldn't hardcoded 1-second
sleep instead correspond to heartbeat interval for task?
- } catch(InterruptedException ie) {
- //ignore
- }
- reporter.setProgressFlag();
- }
-System.out.println("GRR DEBUG: Task commit(): TT canCommit() returned true");
- break;
- } catch (IOException ie) {
-System.out.println("GRR DEBUG: Task commit(): TT canCommit() threw
exception");
- LOG.warn("Failure asking whether task can commit: " +
- StringUtils.stringifyException(ie));
- if (--retries == 0) {
- //if it couldn't query successfully then delete the output
- discardOutput(taskContext);
- System.exit(68);
+ while (true) {
+ try {
+ while (!umbilical.canCommit(taskIdForUmbilical)) {
+ try {
+System.out.println("GRR DEBUG: Task commitAfterApproval(): TT canCommit()
returned false; sleeping 1 sec");
+ // FIXME 1: shouldn't this count down retries, too, in case JT
glitched and no longer knows about us? (else infinite loop)
+ Thread.sleep(1000); // FIXME 2: shouldn't hardcoded 1-second
sleep instead correspond to heartbeat interval for task?
+ } catch(InterruptedException ie) {
+ //ignore
}
+ reporter.setProgressFlag();
+ }
+System.out.println("GRR DEBUG: Task commitAfterApproval(): TT canCommit()
returned true");
+ break;
+ } catch (IOException ie) {
+System.out.println("GRR DEBUG: Task commitAfterApproval(): TT canCommit()
threw exception");
+ LOG.warn("Failure asking whether task can commit: " +
+ StringUtils.stringifyException(ie));
+ if (--retries == 0) {
+ // if it couldn't query successfully then delete the output
+ discardOutput(taskContext);
+ System.exit(68);
}
}
}
-System.out.println("GRR DEBUG: Task commit(): about to call commitTask()");
// task can Commit now
+ commit(umbilical, reporter);
+ }
+
+ // this is protected (rather than private) solely for UberTask map-only case
+ protected void commit(TaskUmbilicalProtocol umbilical,
+ TaskReporter reporter) throws IOException {
+System.out.println("GRR DEBUG: Task commit(): about to call commitTask()");
try {
LOG.info("Task " + taskId + " is allowed to commit now");
committer.commitTask(taskContext);
@@ -1046,7 +1048,7 @@ System.out.println("GRR DEBUG: Task com
} catch (IOException iee) {
LOG.warn("Failure committing: " +
StringUtils.stringifyException(iee));
- //if it couldn't commit a successfully then delete the output
+ // if it couldn't commit successfully then delete the output
discardOutput(taskContext);
throw iee;
}
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java?rev=1079209&r1=1079208&r2=1079209&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/UberTask.java
Tue Mar 8 05:55:58 2011
@@ -293,12 +293,17 @@ class UberTask extends Task {
if (numReduceTasks == 0) {
// For map-only jobs, we need to save (commit) each map's output, which
- // mainly entails asking the TT for permission (in case of speculation)
- // and then moving it up two subdirectory levels in HDFS (i.e., out
- // of _temporary/attempt_xxx). Use UberTask's reporter so we set the
- // progressFlag to which the communication thread is paying attention.
- // (It knows nothing about subReporter.)
- map.commit(umbilical, reporter, false);
+ // usually entails asking the TT for permission (in case of
speculation)
+ // and then moving it up two subdirectory levels in HDFS (i.e., out of
+ // _temporary/attempt_xxx). However, the TT gives permission only if
+ // the JT sent a commitAction for the task, which it hasn't yet done
+ // for UberTask and which it will never do for uber-subtasks of which
+ // it knows nothing. Therefore we just do the two-subdir thing (and
+ // make sure elsewhere that speculation is never on for UberTasks).
+ // Use UberTask's reporter so we set the progressFlag to which the
+ // communication thread is paying attention; it has no knowledge of
+ // subReporter.
+ map.commit(umbilical, reporter);
} else {
// For map+reduce or reduce-only jobs, we merely need to signal the
// communication thread to pass any progress on up to the TT. This
