Reduce shuffle and merge should be done a child JVM

• Key: HADOOP-968
• Type: Improvement
• Status: Closed
• Resolution: Fixed
• Priority: Major
• Assignee: Devaraj Das
• Reporter: Owen O'Malley
• Affects Version/s: 0.10.1
• Fix Version/s: 0.13.0

Description

The Reduce's shuffle and initial merge is done in the TaskTracker's JVM. It would be better to have it run in the Task's child JVM. The advantages are:
1. The class path and environment would be set up correctly.
2. User code doesn't need to be loaded into the TaskTracker.
3. Lower memory usage and contention in the TaskTracker.

Comments

[Devaraj Das]

The salient points of the design:
On the TaskTracker
1) The TaskTracker maintains the list of TaskCompletionEvents for a job. Whenever a ReduceTask is assigned to a TaskTracker it extracts the JobId out of that.
2) For that jobid it starts fetching MapTask completion events as long as any ReduceTask for that job is in the SHUFFLE phase (this ensures that the TaskTracker sees all MapTask lost events and keeps an updated cache of all events). When all the ReduceTasks for a given job have gone past the SHUFFLE phase, the TaskTracker does not fetch any more MapTask completion events until another ReduceTask gets assigned to it. If no other ReduceTask from the same job gets assigned to it, and the job completes, it clears the cache of TaskCompletionEvents.
3) The event-fetcher thread blocks on runningJobs object. Whenever the method addTaskToJob in TaskTracker adds a new Task to a job, it invokes runningJobs.notify(), so that the event-fetcher thread can unblock and continue.
4) The event-fetcher thread also goes through the runningJobs and immediately stops fetching events for those jobs that have been killed/failed.

On the TaskUmbilicalProtocol, ReduceTaskRunner & ReduceTask:
1) A new method - TaskCompletionEvent[] getSuccessMapCompleteEvents(String taskId, int fromIndex, int maxLocs) throws IOException; - has been added for enabling the ReduceTask to fetch TaskCompletionEvents cached at the TaskTracker. The semantics of this method are mirrored to the one in InterTrackerProtocol - getTaskCompletionEvents, except that in the umbilical protocol, we are interested in just the successful map events. Fetch failures are handled in the same way as is done today. Thus, most of the fetcher code in ReduceTaskRunner remains the same (the code now is part of ReduceTask in a new class called ReduceCopier, and the ReduceTaskRunner very closely matches to MapTaskRunner in terms of functionality/code).

Comments?

[Devaraj Das]

Attached patch for review. The unit tests run fine with the patch and I am in the process of running the sort benchmark.

[Devaraj Das]

A well-tested patch.

[Devaraj Das]

This is w.r.t the current (HADOOP-1218 made it go out of sync with the trunk), plus, it has some changes in TaskTracker.java.

[Devaraj Das]

This has the potential to quickly go stale since the patch touches many files in major ways. So would appreciate a quick review/commit on this one. Thanks.

[Hadoop QA]

+1

http://issues.apache.org/jira/secure/attachment/12355215/968.apr10.patch applied and successfully tested against trunk revision r528230.

Test results: http://lucene.zones.apache.org:8080/hudson/job/Hadoop-Patch/42/testReport/
Console output: http://lucene.zones.apache.org:8080/hudson/job/Hadoop-Patch/42/console

[Owen O'Malley]

1. I notice that a lot of your iterators are not typed causing you to do casts of itr.next().
2. In many cases, the loop "for(Item item: itemSet){..}" is easier to read and more concise.
3. Maps should not be iterated through using:
for(Map.Entry<Key,Value> item: myMap) {...}
rather than:
Iterator itr = myMap.keySet().iterator();
while (itr.hasNext()) { Value value = myMap.get(itr.next()); ... }
4. It looks like each reduce from a job will cause its job's FetchState to be added to the list a multiple time, so it will fetch multiple times per a loop.
5. I'd remove the sleep from queryJobTracker and move it to the MapEventsFetcherThread's run loop.
6. The doFetch is badly named, since it doesn't actually do the fetch. It should be called findReduces or something.
7. The name of the parameter of the first parameter in TaskUmbilicalProtocol.getMapCompletionEvents is "taskid", but if fact it is a job id.
8. The MapEventsFetcherThread's name doesn't need to include the task in the normal case, but I guess for unit tests it might be useful.
9. I assume that the shuffle code in ReduceTask matches the old code in ReduceTaskRunner. smile

[Owen O'Malley]

Point 3 should be "Maps SHOULD BE iterated through using". Sorry for any confusion.

[Devaraj Das]

Thanks for the review, Owen. Some comments below.

> 1. I notice that a lot of your iterators are not typed causing you to do casts of itr.next
> 2. In many cases, the loop "for(Item item: itemSet){..}" is easier to read and more
> concise.
> 3. Maps should not be iterated through using:
> for(Map.Entry<Key,Value> item: myMap) {...}

Done (old habits die hard smile).

> 4. It looks like each reduce from a job will cause its job's FetchState to be added to
> the list a multiple time, so it will fetch multiple times per a loop.
No change. There is already a "break" statement in the loop as soon as one FetchState gets added.

> 5. I'd remove the sleep from queryJobTracker and move it to the
> MapEventsFetcherThread's run loop.
Done

> 6. The doFetch is badly named, since it doesn't actually do the fetch. It should be
> called findReduces or something.
Changed that to reducesInShuffle

> 7. The name of the parameter of the first parameter in
> TaskUmbilicalProtocol.getMapCompletionEvents is "taskid", but if fact it is a job id.
Made the name change in TaskUmbilicalProtocol.java

> 8. The MapEventsFetcherThread's name doesn't need to include the task in the
> normal case, but I guess for unit tests it might be useful.
No change

> 9. I assume that the shuffle code in ReduceTask matches the old code in
> ReduceTaskRunner. smile
Smile yes the only change that has been introduced to take care of variable initializations (for example, the variable reduceTask's initialization is different).

[Devaraj Das]

This removes a redundant call to System.currentTimeMillis in the TaskTracker.

[Hadoop QA]

-1, build or testing failed

2 attempts failed to build and test the latest attachment http://issues.apache.org/jira/secure/attachment/12355549/968.apr14.patch against trunk revision r528230.

Test results: http://lucene.zones.apache.org:8080/hudson/job/Hadoop-Patch/50/testReport/
Console output: http://lucene.zones.apache.org:8080/hudson/job/Hadoop-Patch/50/console

Please note that this message is automatically generated and may represent a problem with the automation system and not the patch.

[Doug Cutting]

Here's a re-indented version of this patch.

[Devaraj Das]

This patch addresses an issue to do with metrics reporting wherein stopMonitoring was not called for the ReduceTask. This potentially could lead to hung ReduceTasks after they are finished since the task JVM might not be able to exit until the (non-daemon) monitoring thread goes away.

[Nigel Daley]

+1. The latest patch passes my nightly suit of benchmarks. I have resubmitted it to the patch process.

[David Bowen]

Devaraj, please can you clarify what you fixed with regard to the lack of calls to stopMonitoring? I think you're right that it is a problem, but I would expect the fix to be to change the Timer constructor in o.a.h.metrics.spi.AbstractMetricsContext. Instead of the zero-arg constructor we should use the two arg constructor that takes a thread name and a boolean isDaemon. I didn't see this change in the patch.

[Devaraj Das]

I am pasting the relevant block of code from the latest patch (methodname :TaskTracker.java::Child::main()). Basically, I invoke the close() method of metricsContext for the context "mapred" just before the point where the child task closes the log manager (LogManager.shutdown()) and it is about to die. I thought this was the best way to stop the monitoring and let the task exit nicely without having to touch other parts of the metrics code.

} finally {
+ MetricsContext metricsContext = MetricsUtil.getContext("mapred");
+ metricsContext.close();
// Shutting down log4j of the child-vm...
// This assumes that on return from Task.run()
// there is no more logging done.

Makes sense?

[Doug Cutting]

I think David's point was that it would be best to fix the metrics code so that its thread is a daemon thread, rather than to expect client code to explicitly stop that thread.

[Devaraj Das]

Ok, changed the Timer constructor to the two-argument one. Also, retained the metricsContext.close() call that I had in the last patch. In general, I think that it is a good idea to call close() on an object if the close() method is clearly documented. Does this seem all right?

[Doug Cutting]

I just committed this. Thanks, Devaraj!

[Hadoop QA]

-1, could not apply patch.

The patch command could not apply the latest attachment http://issues.apache.org/jira/secure/attachment/12355713/968-with-metrics-fix.new.patch as a patch to trunk revision r529763.

Console output: http://lucene.zones.apache.org:8080/hudson/job/Hadoop-Patch/58/console

Please note that this message is automatically generated and may represent a problem with the automation system and not the patch.

[Hadoop QA]

Integrated in Hadoop-Nightly #61 (See http://lucene.zones.apache.org:8080/hudson/job/Hadoop-Nightly/61/)