Author: omalley
Date: Fri Mar 4 04:51:24 2011
New Revision: 1077750
URL: http://svn.apache.org/viewvc?rev=1077750&view=rev
Log:
commit d5ea1fcd178fc408b2aab50fc44ad17c69e44c90
Author: Arun C Murthy <[email protected]>
Date: Tue Nov 30 21:51:44 2010 -0800
. Reducing locking contention in TaskTracker.MapOutputServlet's
LocalDirAllocator. Contributed by Rajesh Balamohan.
+++ b/YAHOO-CHANGES.txt
+ . Reducing locking contention in TaskTracker.MapOutputServlet's
+ LocalDirAllocator. (Rajesh Balamohan via acmurthy)
+
Added:
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/mapred/TestLRUCache.java
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java?rev=1077750&r1=1077749&r2=1077750&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
(original)
+++
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
Fri Mar 4 04:51:24 2011
@@ -42,6 +42,7 @@ import java.util.Random;
import java.util.Set;
import java.util.TreeMap;
import java.util.Vector;
+import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.regex.Pattern;
@@ -177,6 +178,7 @@ public class TaskTracker implements MRCo
InterTrackerProtocol jobClient;
private TrackerDistributedCacheManager distributedCacheManager;
+ static int FILE_CACHE_SIZE = 2000;
// last heartbeat response recieved
short heartbeatResponseId = -1;
@@ -1233,6 +1235,7 @@ public class TaskTracker implements MRCo
*/
public TaskTracker(JobConf conf) throws IOException, InterruptedException {
originalConf = conf;
+ FILE_CACHE_SIZE = conf.getInt("mapred.tasktracker.file.cache.size", 2000);
maxMapSlots = conf.getInt(
"mapred.tasktracker.map.tasks.maximum", 2);
maxReduceSlots = conf.getInt(
@@ -1280,6 +1283,7 @@ public class TaskTracker implements MRCo
server.start();
this.httpPort = server.getPort();
checkJettyPort(httpPort);
+ LOG.info("FILE_CACHE_SIZE for mapOutputServlet set to : " +
FILE_CACHE_SIZE);
mapRetainSize = conf.getLong(TaskLogsTruncater.MAP_USERLOG_RETAIN_SIZE,
TaskLogsTruncater.DEFAULT_RETAIN_SIZE);
reduceRetainSize =
conf.getLong(TaskLogsTruncater.REDUCE_USERLOG_RETAIN_SIZE,
@@ -3360,6 +3364,40 @@ public class TaskTracker implements MRCo
System.exit(-1);
}
}
+
+ static class LRUCache<K, V> {
+ private int cacheSize;
+ private LinkedHashMap<K, V> map;
+
+ public LRUCache(int cacheSize) {
+ this.cacheSize = cacheSize;
+ this.map = new LinkedHashMap<K, V>(cacheSize, 0.75f, true) {
+ protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
+ return size() > LRUCache.this.cacheSize;
+ }
+ };
+ }
+
+ public synchronized V get(K key) {
+ return map.get(key);
+ }
+
+ public synchronized void put(K key, V value) {
+ map.put(key, value);
+ }
+
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public Iterator<Entry<K, V>> getIterator() {
+ return new LinkedList<Entry<K, V>>(map.entrySet()).iterator();
+ }
+
+ public synchronized void clear() {
+ map.clear();
+ }
+ }
/**
* This class is used in TaskTracker's Jetty to serve the map outputs
@@ -3368,6 +3406,10 @@ public class TaskTracker implements MRCo
public static class MapOutputServlet extends HttpServlet {
private static final long serialVersionUID = 1L;
private static final int MAX_BYTES_TO_READ = 64 * 1024;
+
+ private static LRUCache<String, Path> fileCache = new LRUCache<String,
Path>(FILE_CACHE_SIZE);
+ private static LRUCache<String, Path> fileIndexCache = new
LRUCache<String, Path>(FILE_CACHE_SIZE);
+
@Override
public void doGet(HttpServletRequest request,
HttpServletResponse response
@@ -3422,16 +3464,22 @@ public class TaskTracker implements MRCo
runAsUserName = tracker.getTaskController().getRunAsUser(rjob.jobConf);
}
// Index file
- Path indexFileName =
- lDirAlloc.getLocalPathToRead(TaskTracker.getIntermediateOutputDir(
- userName, jobId, mapId)
- + "/file.out.index", conf);
+ String intermediateOutputDir =
TaskTracker.getIntermediateOutputDir(userName, jobId, mapId);
+ String indexKey = intermediateOutputDir + "/file.out.index";
+ Path indexFileName = fileIndexCache.get(indexKey);
+ if (indexFileName == null) {
+ indexFileName = lDirAlloc.getLocalPathToRead(indexKey, conf);
+ fileIndexCache.put(indexKey, indexFileName);
+ }
// Map-output file
- Path mapOutputFileName =
- lDirAlloc.getLocalPathToRead(TaskTracker.getIntermediateOutputDir(
- userName, jobId, mapId)
- + "/file.out", conf);
+ String fileKey = intermediateOutputDir + "/file.out";
+ Path mapOutputFileName = fileCache.get(fileKey);
+ if (mapOutputFileName == null) {
+ mapOutputFileName = lDirAlloc.getLocalPathToRead(fileKey, conf);
+ fileCache.put(fileKey, mapOutputFileName);
+ }
+
/**
* Read the index file to get the information about where
@@ -3489,10 +3537,12 @@ public class TaskTracker implements MRCo
len =
mapOutputIn.read(buffer, 0, (int)Math.min(rem, MAX_BYTES_TO_READ));
}
-
- LOG.info("Sent out " + totalRead + " bytes for reduce: " + reduce +
+
+ if (LOG.isDebugEnabled()) {
+ LOG.info("Sent out " + totalRead + " bytes for reduce: " + reduce +
" from map: " + mapId + " given " + info.partLength + "/" +
info.rawLength);
+ }
} catch (IOException ie) {
Log log = (Log) context.getAttribute("log");
String errorMsg = ("getMapOutput(" + mapId + "," + reduceId +
Added:
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/mapred/TestLRUCache.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/mapred/TestLRUCache.java?rev=1077750&view=auto
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/mapred/TestLRUCache.java
(added)
+++
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/mapred/TestLRUCache.java
Fri Mar 4 04:51:24 2011
@@ -0,0 +1,103 @@
+/**
+ * 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.util.Iterator;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.Path;
+
+public class TestLRUCache extends TestCase {
+ private static final Log LOG =
+ LogFactory.getLog(TestLRUCache.class);
+
+ public void testPut() {
+ TaskTracker.LRUCache<String, Path> cache = new
TaskTracker.LRUCache<String, Path>(200);
+
+ for(int i=0;i<200;i++) {
+ cache.put(i+"", new Path("/foo"+i));
+ }
+
+ Iterator<Map.Entry<String, Path>> iterator = cache.getIterator();
+ int i=0;
+ while(iterator.hasNext()) {
+ Map.Entry<String, Path> entry = iterator.next();
+ String key = entry.getKey();
+ Path val = entry.getValue();
+ assertEquals(i+"", key);
+ i++;
+ }
+ LOG.info("Completed testPut");
+ }
+
+
+ public void testGet() {
+ TaskTracker.LRUCache<String, Path> cache = new
TaskTracker.LRUCache<String, Path>(200);
+
+ for(int i=0;i<200;i++) {
+ cache.put(i+"", new Path("/foo"+i));
+ }
+
+ for(int i=0;i<200;i++) {
+ Path path = cache.get(i+"");
+ assertEquals(path.toString(), (new Path("/foo"+i)).toString());
+ }
+ LOG.info("Completed testGet");
+ }
+
+
+ /**
+ * Test if cache can be cleared properly
+ */
+ public void testClear() {
+ TaskTracker.LRUCache<String, Path> cache = new
TaskTracker.LRUCache<String, Path>(200);
+
+ for(int i=0;i<200;i++) {
+ cache.put(i+"", new Path("/foo"+i));
+ }
+
+ cache.clear();
+ assertTrue(cache.size() == 0);
+ LOG.info("Completed testClear");
+ }
+
+ /**
+ * Test for cache overflow condition
+ */
+ public void testOverFlow() {
+ TaskTracker.LRUCache<String, Path> cache = new
TaskTracker.LRUCache<String, Path>(200);
+
+ int SIZE = 5000;
+
+ for(int i=0;i<SIZE;i++) {
+ cache.put(i+"", new Path("/foo"+i));
+ }
+
+ //Check if the items are removed properly when the cache size is exceeded
+ for(int i=SIZE-1;i>=SIZE-200;i--) {
+ Path path = cache.get(i+"");
+ assertEquals(path.toString(), (new Path("/foo"+i)).toString());
+ }
+
+ LOG.info("Completed testOverFlow");
+ }
+}
\ No newline at end of file
