You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ar...@apache.org on 2014/09/13 05:54:16 UTC
[10/50] [abbrv] git commit: MAPREDUCE-5997. native-task: Use
DirectBufferPool from Hadoop Common. Contributed by Todd Lipcon.
MAPREDUCE-5997. native-task: Use DirectBufferPool from Hadoop Common. Contributed by Todd Lipcon.
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/MR-2841@1613034 13f79535-47bb-0310-9956-ffa450edef68
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/77acc70d
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/77acc70d
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/77acc70d
Branch: refs/heads/HDFS-6581
Commit: 77acc70df5d64055ea809222f3d2f0d66c611196
Parents: 5149a8a
Author: Todd Lipcon <to...@apache.org>
Authored: Thu Jul 24 08:20:25 2014 +0000
Committer: Todd Lipcon <to...@apache.org>
Committed: Thu Jul 24 08:20:25 2014 +0000
----------------------------------------------------------------------
.../CHANGES.MAPREDUCE-2841.txt | 1 +
.../mapred/nativetask/NativeBatchProcessor.java | 8 +-
.../nativetask/buffer/DirectBufferPool.java | 93 ---------
.../mapred/nativetask/buffer/InputBuffer.java | 17 +-
.../nativetask/buffer/TestDirectBufferPool.java | 201 -------------------
5 files changed, 20 insertions(+), 300 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77acc70d/hadoop-mapreduce-project/CHANGES.MAPREDUCE-2841.txt
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/CHANGES.MAPREDUCE-2841.txt b/hadoop-mapreduce-project/CHANGES.MAPREDUCE-2841.txt
index cea5a76..e12f743 100644
--- a/hadoop-mapreduce-project/CHANGES.MAPREDUCE-2841.txt
+++ b/hadoop-mapreduce-project/CHANGES.MAPREDUCE-2841.txt
@@ -4,3 +4,4 @@ Changes for Hadoop Native Map Output Collector
MAPREDUCE-5985. native-task: Fix build on macosx. Contributed by Binglin Chang
MAPREDUCE-5994. Simplify ByteUtils and fix failing test. (todd)
MAPREDUCE-5996. native-task: Rename system tests into standard directory layout (todd)
+MAPREDUCE-5997. native-task: Use DirectBufferPool from Hadoop Common (todd)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77acc70d/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/NativeBatchProcessor.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/NativeBatchProcessor.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/NativeBatchProcessor.java
index fd68ea6..837da0e 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/NativeBatchProcessor.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/NativeBatchProcessor.java
@@ -24,12 +24,13 @@ import java.nio.ByteBuffer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.mapred.nativetask.buffer.BufferType;
-import org.apache.hadoop.mapred.nativetask.buffer.DirectBufferPool;
import org.apache.hadoop.mapred.nativetask.buffer.InputBuffer;
import org.apache.hadoop.mapred.nativetask.buffer.OutputBuffer;
import org.apache.hadoop.mapred.nativetask.util.ReadWriteBuffer;
import org.apache.hadoop.mapred.nativetask.util.ConfigUtil;
+import org.apache.hadoop.util.DirectBufferPool;
/**
* used to create channel, transfer data and command between Java and native
@@ -126,9 +127,8 @@ public class NativeBatchProcessor implements INativeHandler {
NativeRuntime.releaseNativeObject(nativeHandlerAddr);
nativeHandlerAddr = 0;
}
- if (null != in && null != in.getByteBuffer() && in.getByteBuffer().isDirect()) {
- DirectBufferPool.getInstance().returnBuffer(in.getByteBuffer());
- }
+ IOUtils.cleanup(LOG, in);
+ in = null;
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77acc70d/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/DirectBufferPool.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/DirectBufferPool.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/DirectBufferPool.java
deleted file mode 100644
index bd3c6bb..0000000
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/DirectBufferPool.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/**
- * 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.nativetask.buffer;
-
-import java.io.IOException;
-import java.lang.ref.WeakReference;
-import java.nio.ByteBuffer;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ConcurrentMap;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-
-/**
- * as direct buffer memory is not collected by GC, we keep a pool
- * to reuse direct buffers
- */
-public class DirectBufferPool {
-
- private static DirectBufferPool directBufferPool = null;
- private static Log LOG = LogFactory.getLog(DirectBufferPool.class);
- private ConcurrentMap<Integer, Queue<WeakReference<ByteBuffer>>> bufferMap = new ConcurrentHashMap<Integer, Queue<WeakReference<ByteBuffer>>>();
-
- private DirectBufferPool() {
- }
-
- public static synchronized DirectBufferPool getInstance() {
- if (null == directBufferPool) {
- directBufferPool = new DirectBufferPool();
- }
- return directBufferPool;
- }
-
- public static void destoryInstance(){
- directBufferPool = null;
- }
-
- public synchronized ByteBuffer borrowBuffer(int capacity) throws IOException {
- Queue<WeakReference<ByteBuffer>> list = bufferMap.get(capacity);
- if (null == list) {
- return ByteBuffer.allocateDirect(capacity);
- }
- WeakReference<ByteBuffer> ref;
- while ((ref = list.poll()) != null) {
- ByteBuffer buf = ref.get();
- if (buf != null) {
- return buf;
- }
- }
- return ByteBuffer.allocateDirect(capacity);
- }
-
- public void returnBuffer(ByteBuffer buffer) throws IOException {
- if (null == buffer || !buffer.isDirect()) {
- throw new IOException("the buffer is null or the buffer returned is not direct buffer");
- }
-
- buffer.clear();
- int capacity = buffer.capacity();
- Queue<WeakReference<ByteBuffer>> list = bufferMap.get(capacity);
- if (null == list) {
- list = new ConcurrentLinkedQueue<WeakReference<ByteBuffer>>();
- Queue<WeakReference<ByteBuffer>> prev = bufferMap.putIfAbsent(capacity, list);
- if (prev != null) {
- list = prev;
- }
- }
- list.add(new WeakReference<ByteBuffer>(buffer));
- }
-
- int getBufCountsForCapacity(int capacity) {
- return bufferMap.get(capacity).size();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77acc70d/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/InputBuffer.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/InputBuffer.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/InputBuffer.java
index eb15164..82193b8 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/InputBuffer.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/main/java/org/apache/hadoop/mapred/nativetask/buffer/InputBuffer.java
@@ -18,11 +18,16 @@
package org.apache.hadoop.mapred.nativetask.buffer;
+import org.apache.hadoop.util.DirectBufferPool;
+
+import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
-public class InputBuffer {
+public class InputBuffer implements Closeable {
+
+ static DirectBufferPool bufferPool = new DirectBufferPool();
private ByteBuffer byteBuffer;
private final BufferType type;
@@ -36,7 +41,7 @@ public class InputBuffer {
switch (type) {
case DIRECT_BUFFER:
- this.byteBuffer = DirectBufferPool.getInstance().borrowBuffer(capacity);
+ this.byteBuffer = bufferPool.getBuffer(capacity);
this.byteBuffer.order(ByteOrder.BIG_ENDIAN);
break;
case HEAP_BUFFER:
@@ -118,4 +123,12 @@ public class InputBuffer {
}
return byteBuffer.array();
}
+
+ @Override
+ public void close() {
+ if (byteBuffer != null && byteBuffer.isDirect()) {
+ bufferPool.returnBuffer(byteBuffer);
+ byteBuffer = null;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77acc70d/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/test/java/org/apache/hadoop/mapred/nativetask/buffer/TestDirectBufferPool.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/test/java/org/apache/hadoop/mapred/nativetask/buffer/TestDirectBufferPool.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/test/java/org/apache/hadoop/mapred/nativetask/buffer/TestDirectBufferPool.java
deleted file mode 100644
index 09c1ef5..0000000
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-nativetask/src/test/java/org/apache/hadoop/mapred/nativetask/buffer/TestDirectBufferPool.java
+++ /dev/null
@@ -1,201 +0,0 @@
-/**
- * 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.nativetask.buffer;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotSame;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.List;
-import java.util.ArrayList;
-
-import org.junit.Test;
-
-public class TestDirectBufferPool {
-
- @Test
- public void testGetInstance() throws Exception {
- final int num = 100;
- List<DirectBufferPool> pools = new ArrayList<DirectBufferPool>();
- Thread[] list = new Thread[num];
- for (int i = 0; i < num; i++) {
- Thread t = getPoolThread(pools);
- t.start();
- list[i] = t;
- }
- for (int i = 0; i < num; i++) {
- try {
- list[i].join(10000);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- DirectBufferPool p1 = pools.get(0);
- assertNotNull(p1);
- for (int i = 1; i < pools.size(); i++) {
- DirectBufferPool p2 = pools.get(i);
- assertNotNull(p2);
- assertSame(p1, p2);
- }
- }
-
- private Thread getPoolThread(final List<DirectBufferPool> pools) {
- Thread t = new Thread() {
- public void run() {
- pools.add(DirectBufferPool.getInstance());
- }
- };
- return t;
- }
-
-
- @Test
- public void testBufBorrow() throws IOException {
- final DirectBufferPool bufferPool = DirectBufferPool.getInstance();
- ByteBuffer b1 = bufferPool.borrowBuffer(100);
- assertTrue(b1.isDirect());
- assertEquals(0, b1.position());
- assertEquals(100, b1.capacity());
- bufferPool.returnBuffer(b1);
- ByteBuffer b2 = bufferPool.borrowBuffer(100);
- assertTrue(b2.isDirect());
- assertEquals(0, b2.position());
- assertEquals(100, b2.capacity());
- assertSame(b1, b2);
-
- ByteBuffer b3 = bufferPool.borrowBuffer(100);
- assertTrue(b3.isDirect());
- assertEquals(0, b3.position());
- assertEquals(100, b3.capacity());
- assertNotSame(b2, b3);
- bufferPool.returnBuffer(b2);
- bufferPool.returnBuffer(b3);
- }
-
- @Test
- public void testBufReset() throws IOException {
- final DirectBufferPool bufferPool = DirectBufferPool.getInstance();
- ByteBuffer b1 = bufferPool.borrowBuffer(100);
- assertTrue(b1.isDirect());
- assertEquals(0, b1.position());
- assertEquals(100, b1.capacity());
- b1.putInt(1);
- assertEquals(4, b1.position());
- bufferPool.returnBuffer(b1);
- ByteBuffer b2 = bufferPool.borrowBuffer(100);
- assertSame(b1, b2);
- assertTrue(b2.isDirect());
- assertEquals(0, b2.position());
- assertEquals(100, b2.capacity());
- }
-
- @Test
- public void testBufReturn() throws IOException {
- final DirectBufferPool bufferPool = DirectBufferPool.getInstance();
- int numOfBufs = 100;
- int capacity = 100;
- final ByteBuffer[] bufs = new ByteBuffer[numOfBufs];
- for (int i = 0; i < numOfBufs; i++) {
- bufs[i] = bufferPool.borrowBuffer(capacity);
- }
-
- assertEquals(0, bufferPool.getBufCountsForCapacity(capacity));
-
-
- int numOfThreads = numOfBufs;
- Thread[] list = new Thread[numOfThreads];
- for (int i = 0; i < numOfThreads; i++) {
- Thread t = retBufThread(bufferPool, bufs, i);
- t.start();
- list[i] = t;
- }
- for (int i = 0; i < numOfThreads; i++) {
- try {
- list[i].join(10000);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- assertEquals(numOfBufs, bufferPool.getBufCountsForCapacity(capacity));
- }
-
- private Thread retBufThread(final DirectBufferPool bufferPool, final ByteBuffer[] bufs, final int i) {
- Thread t = new Thread(new Runnable(){
- @Override
- public void run() {
- try {
- bufferPool.returnBuffer(bufs[i]);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- });
- return t;
- }
-
- @Test
- public void testBufException() {
- final DirectBufferPool bufferPool = DirectBufferPool.getInstance();
- boolean thrown = false;
- try {
- bufferPool.returnBuffer(null);
- } catch (IOException e) {
- thrown = true;
- }
- assertEquals(true, thrown);
-
- thrown = false;
- ByteBuffer buf = ByteBuffer.allocate(100);
- try {
- bufferPool.returnBuffer(buf);
- } catch (IOException e) {
- thrown = true;
- }
- assertEquals(true, thrown);
- }
-
- @Test
- public void testBufWeakRefClear() throws IOException {
- final DirectBufferPool bufferPool = DirectBufferPool.getInstance();
- int numOfBufs = 100;
- int capacity = 100;
- ByteBuffer[] list = new ByteBuffer[capacity];
- for (int i = 0; i < numOfBufs; i++) {
- list[i] = bufferPool.borrowBuffer(capacity);
- }
- for (int i = 0; i < numOfBufs; i++) {
- bufferPool.returnBuffer(list[i]);
- list[i] = null;
- }
-
- assertEquals(numOfBufs, bufferPool.getBufCountsForCapacity(capacity));
-
- for (int i = 0; i < 3; i++) {
- System.gc();
- }
-
- ByteBuffer b = bufferPool.borrowBuffer(capacity);
- assertEquals(0, bufferPool.getBufCountsForCapacity(capacity));
- }
-}