Skip to content

TEZ-4269: Re-Work Threadpool in DAGAppMaster #92

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jteagles merged 2 commits into from
Jan 28, 2021
Merged

TEZ-4269: Re-Work Threadpool in DAGAppMaster #92

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7fd8d16
TEZ-4269: Re-Work Threadpool in DAGAppMaster
belugabehr Jan 25, 2021
42835aa
Remove reference to rawExecutor
belugabehr Jan 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 1 addition & 4 deletions tez-dag/src/main/java/org/apache/tez/dag/app/AppContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ExecutorService;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
Expand Down Expand Up @@ -74,9 +74,6 @@ public interface AppContext {

DAG getCurrentDAG();

// For testing only!
ThreadPoolExecutor getThreadPool();

ListeningExecutorService getExecService();

void setDAG(DAG dag);
Expand Down
17 changes: 4 additions & 13 deletions tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,10 +47,10 @@
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -302,7 +302,6 @@ public class DAGAppMaster extends AbstractService {
private Path tezSystemStagingDir;
private FileSystem recoveryFS;

private ThreadPoolExecutor rawExecutor;
private ListeningExecutorService execService;

// TODO May not need to be a bidi map
Expand Down Expand Up @@ -623,9 +622,9 @@ public synchronized void serviceInit(final Configuration conf) throws Exception
TezConfiguration.TEZ_AM_DAG_APPCONTEXT_THREAD_COUNT_LIMIT_DEFAULT);
// NOTE: LinkedBlockingQueue does not have a capacity Limit and can thus
// occupy large memory chunks when numerous Runables are pending for execution
rawExecutor = new ThreadPoolExecutor(threadCount, threadCount,
60L, TimeUnit.SECONDS, new LinkedBlockingQueue(),
new ThreadFactoryBuilder().setDaemon(true).setNameFormat("App Shared Pool - " + "#%d").build());
ExecutorService rawExecutor =
Executors.newFixedThreadPool(threadCount, new ThreadFactoryBuilder()
.setDaemon(true).setNameFormat("App Shared Pool - #%d").build());
execService = MoreExecutors.listeningDecorator(rawExecutor);

initServices(conf);
Expand Down Expand Up @@ -1505,14 +1504,6 @@ public DAG getCurrentDAG() {
return dag;
}

@Override
// For Testing only!
public ThreadPoolExecutor getThreadPool() {
synchronized (DAGAppMaster.this) {
return rawExecutor;
}
}

@Override
public ListeningExecutorService getExecService() {
return execService;
Expand Down