Make the number of task event loop configurable

presto-main/src/main/java/com/facebook/presto/execution/QueryManagerConfig.java

@@ -70,7 +70,7 @@ public class QueryManagerConfig
     private int queryManagerExecutorPoolSize = 5;
 
     private Duration remoteTaskMaxErrorDuration = new Duration(5, TimeUnit.MINUTES);
-    private int remoteTaskMaxCallbackThreads = 1000;
+    private int remoteTaskMaxCallbackThreads = Runtime.getRuntime().availableProcessors();
 
     private String queryExecutionPolicy = "all-at-once";
     private Duration queryMaxRunTime = new Duration(100, TimeUnit.DAYS);

presto-main/src/main/java/com/facebook/presto/server/remotetask/HttpRemoteTaskFactory.java

@@ -95,15 +95,7 @@ public class HttpRemoteTaskFactory
     private final QueryManager queryManager;
     private final DecayCounter taskUpdateRequestSize;
     private final boolean taskUpdateSizeTrackingEnabled;
-    private final EventLoopGroup eventLoopGroup = new DefaultEventLoopGroup(Runtime.getRuntime().availableProcessors(),
-            new ThreadFactoryBuilder().setNameFormat("task-event-loop-%s").setDaemon(true).build())
-    {
-        @Override
-        protected EventLoop newChild(Executor executor, Object... args)
-        {
-            return new SafeEventLoop(this, executor);
-        }
-    };
+    private final EventLoopGroup eventLoopGroup;
 
     @Inject
     public HttpRemoteTaskFactory(
@@ -182,6 +174,15 @@ else if (binaryTransportEnabled) {
 
         this.taskUpdateRequestSize = new DecayCounter(ExponentialDecay.oneMinute());
         this.taskUpdateSizeTrackingEnabled = taskConfig.isTaskUpdateSizeTrackingEnabled();
+        this.eventLoopGroup = new DefaultEventLoopGroup(config.getRemoteTaskMaxCallbackThreads(),
+                new ThreadFactoryBuilder().setNameFormat("task-event-loop-%s").setDaemon(true).build())
+        {
+            @Override
+            protected EventLoop newChild(Executor executor, Object... args)
+            {
+                return new SafeEventLoop(this, executor);
+            }
+        };
     }
 
     @Managed

presto-main/src/test/java/com/facebook/presto/execution/TestQueryManagerConfig.java

@@ -59,7 +59,7 @@ public void testDefaults()
                 .setQueryManagerExecutorPoolSize(5)
                 .setRemoteTaskMinErrorDuration(new Duration(5, TimeUnit.MINUTES))
                 .setRemoteTaskMaxErrorDuration(new Duration(5, TimeUnit.MINUTES))
-                .setRemoteTaskMaxCallbackThreads(1000)
+                .setRemoteTaskMaxCallbackThreads(Runtime.getRuntime().availableProcessors())
                 .setQueryExecutionPolicy("all-at-once")
                 .setQueryMaxRunTime(new Duration(100, TimeUnit.DAYS))
                 .setQueryMaxExecutionTime(new Duration(100, TimeUnit.DAYS))