Repository: tez Updated Branches: refs/heads/TEZ-3334 72b3e19c4 -> c3a7c2127
TEZ-3628. Give Tez shuffle handler threads custom names (jeagles) Project: http://git-wip-us.apache.org/repos/asf/tez/repo Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/c3a7c212 Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/c3a7c212 Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/c3a7c212 Branch: refs/heads/TEZ-3334 Commit: c3a7c21271d2038ae2a7b4006b6452ac05dd5eee Parents: 72b3e19 Author: Jonathan Eagles <[email protected]> Authored: Mon Mar 13 20:18:45 2017 -0500 Committer: Jonathan Eagles <[email protected]> Committed: Mon Mar 13 20:18:45 2017 -0500 ---------------------------------------------------------------------- TEZ-3334-CHANGES.txt | 1 + .../apache/tez/auxservices/ShuffleHandler.java | 33 ++++++++++++-------- 2 files changed, 21 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tez/blob/c3a7c212/TEZ-3334-CHANGES.txt ---------------------------------------------------------------------- diff --git a/TEZ-3334-CHANGES.txt b/TEZ-3334-CHANGES.txt index 5d5ee71..0fb021e 100644 --- a/TEZ-3334-CHANGES.txt +++ b/TEZ-3334-CHANGES.txt @@ -4,6 +4,7 @@ Apache Tez Change Log INCOMPATIBLE CHANGES: ALL CHANGES: + TEZ-3628. Give Tez shuffle handler threads custom names TEZ-3621. Optimize the Shuffle Handler content length calculation for keep alive TEZ-3620. UnorderedPartitionedKVOutput is missing the shuffle service config in the confKeys set TEZ-3618. Shuffle Handler Loading cache equality tests always results is false http://git-wip-us.apache.org/repos/asf/tez/blob/c3a7c212/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java b/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java index fa6d888..5da144f 100644 --- a/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java +++ b/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java @@ -48,7 +48,6 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Pattern; @@ -118,7 +117,9 @@ import org.jboss.netty.channel.MessageEvent; import org.jboss.netty.channel.SimpleChannelUpstreamHandler; import org.jboss.netty.channel.group.ChannelGroup; import org.jboss.netty.channel.group.DefaultChannelGroup; +import org.jboss.netty.channel.socket.nio.NioServerBossPool; import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; +import org.jboss.netty.channel.socket.nio.NioWorkerPool; import org.jboss.netty.handler.codec.frame.TooLongFrameException; import org.jboss.netty.handler.codec.http.DefaultHttpResponse; import org.jboss.netty.handler.codec.http.HttpChunkAggregator; @@ -132,6 +133,7 @@ import org.jboss.netty.handler.codec.http.QueryStringDecoder; import org.jboss.netty.handler.ssl.SslHandler; import org.jboss.netty.handler.stream.ChunkedWriteHandler; import org.jboss.netty.util.CharsetUtil; +import org.jboss.netty.util.ThreadNameDeterminer; import org.slf4j.LoggerFactory; import com.google.common.annotations.VisibleForTesting; @@ -142,7 +144,6 @@ import com.google.common.cache.LoadingCache; import com.google.common.cache.RemovalListener; import com.google.common.cache.RemovalNotification; import com.google.common.cache.Weigher; -import com.google.common.util.concurrent.ThreadFactoryBuilder; import com.google.protobuf.ByteString; public class ShuffleHandler extends AuxiliaryService { @@ -487,17 +488,23 @@ public class ShuffleHandler extends AuxiliaryService { maxSessionOpenFiles = conf.getInt(SHUFFLE_MAX_SESSION_OPEN_FILES, DEFAULT_SHUFFLE_MAX_SESSION_OPEN_FILES); - ThreadFactory bossFactory = new ThreadFactoryBuilder() - .setNameFormat("ShuffleHandler Netty Boss #%d") - .build(); - ThreadFactory workerFactory = new ThreadFactoryBuilder() - .setNameFormat("ShuffleHandler Netty Worker #%d") - .build(); - - selector = new NioServerSocketChannelFactory( - Executors.newCachedThreadPool(bossFactory), - Executors.newCachedThreadPool(workerFactory), - maxShuffleThreads); + final String BOSS_THREAD_NAME_PREFIX = "Tez Shuffle Handler Boss #"; + NioServerBossPool bossPool = new NioServerBossPool(Executors.newCachedThreadPool(), 1, new ThreadNameDeterminer() { + @Override + public String determineThreadName(String currentThreadName, String proposedThreadName) throws Exception { + return BOSS_THREAD_NAME_PREFIX + currentThreadName.substring(currentThreadName.lastIndexOf('-') + 1); + } + }); + + final String WORKER_THREAD_NAME_PREFIX = "Tez Shuffle Handler Worker #"; + NioWorkerPool workerPool = new NioWorkerPool(Executors.newCachedThreadPool(), maxShuffleThreads, new ThreadNameDeterminer() { + @Override + public String determineThreadName(String currentThreadName, String proposedThreadName) throws Exception { + return WORKER_THREAD_NAME_PREFIX + currentThreadName.substring(currentThreadName.lastIndexOf('-') + 1); + } + }); + + selector = new NioServerSocketChannelFactory(bossPool, workerPool); super.serviceInit(new YarnConfiguration(conf)); }
