Revert "Use a safe eventLoop wrapper to avoid thread shutdown and fail the task correctly"

This reverts commit 48f2fb1.

Author: shangm2

presto-main/pom.xml

@@ -516,11 +516,6 @@
             <artifactId>netty-transport</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>io.netty</groupId>
-            <artifactId>netty-common</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>com.squareup.okhttp3</groupId>
             <artifactId>mockwebserver</artifactId>

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

@@ -76,6 +76,7 @@
 import com.sun.management.ThreadMXBean;
 import io.airlift.units.DataSize;
 import io.airlift.units.Duration;
+import io.netty.channel.EventLoop;
 import it.unimi.dsi.fastutil.longs.LongArrayList;
 import org.joda.time.DateTime;
 
@@ -142,7 +143,7 @@
  * This class now uses an event loop concurrency model to eliminate the need for explicit synchronization:
  * <ul>
  * <li>All mutable state access and modifications are performed on a single dedicated event loop thread</li>
- * <li>External threads submit operations to the event loop using {@code safeExecuteOnEventLoop()}</li>
+ * <li>External threads submit operations to the event loop using {@code taskEventLoop.execute()}</li>
  * <li>The event loop serializes all operations, eliminating race conditions without using locks</li>
  * </ul>
  * <p>
@@ -230,9 +231,9 @@ public final class HttpRemoteTask
     private final boolean taskUpdateSizeTrackingEnabled;
     private final SchedulerStatsTracker schedulerStatsTracker;
 
-    private final HttpRemoteTaskFactory.SafeEventLoop taskEventLoop;
+    private final EventLoop taskEventLoop;
 
-    public static HttpRemoteTask createHttpRemoteTask(
+    public HttpRemoteTask(
             Session session,
             TaskId taskId,
             String nodeId,
@@ -268,84 +269,7 @@ public static HttpRemoteTask createHttpRemoteTask(
             HandleResolver handleResolver,
             ConnectorTypeSerdeManager connectorTypeSerdeManager,
             SchedulerStatsTracker schedulerStatsTracker,
-            HttpRemoteTaskFactory.SafeEventLoop taskEventLoop)
-    {
-        HttpRemoteTask task = new HttpRemoteTask(session,
-                taskId,
-                nodeId,
-                location,
-                remoteLocation,
-                planFragment,
-                initialSplits,
-                outputBuffers,
-                httpClient,
-                maxErrorDuration,
-                taskStatusRefreshMaxWait,
-                taskInfoRefreshMaxWait,
-                taskInfoUpdateInterval,
-                summarizeTaskInfo,
-                taskStatusCodec,
-                taskInfoCodec,
-                taskInfoJsonCodec,
-                taskUpdateRequestCodec,
-                planFragmentCodec,
-                metadataUpdatesCodec,
-                nodeStatsTracker,
-                stats,
-                binaryTransportEnabled,
-                thriftTransportEnabled,
-                taskInfoThriftTransportEnabled,
-                thriftProtocol,
-                tableWriteInfo,
-                maxTaskUpdateSizeInBytes,
-                metadataManager,
-                queryManager,
-                taskUpdateRequestSize,
-                taskUpdateSizeTrackingEnabled,
-                handleResolver,
-                connectorTypeSerdeManager,
-                schedulerStatsTracker,
-                taskEventLoop);
-        task.initialize();
-        return task;
-    }
-
-    private HttpRemoteTask(Session session,
-            TaskId taskId,
-            String nodeId,
-            URI location,
-            URI remoteLocation,
-            PlanFragment planFragment,
-            Multimap<PlanNodeId, Split> initialSplits,
-            OutputBuffers outputBuffers,
-            HttpClient httpClient,
-            Duration maxErrorDuration,
-            Duration taskStatusRefreshMaxWait,
-            Duration taskInfoRefreshMaxWait,
-            Duration taskInfoUpdateInterval,
-            boolean summarizeTaskInfo,
-            Codec<TaskStatus> taskStatusCodec,
-            Codec<TaskInfo> taskInfoCodec,
-            Codec<TaskInfo> taskInfoJsonCodec,
-            Codec<TaskUpdateRequest> taskUpdateRequestCodec,
-            Codec<PlanFragment> planFragmentCodec,
-            Codec<MetadataUpdates> metadataUpdatesCodec,
-            NodeStatsTracker nodeStatsTracker,
-            RemoteTaskStats stats,
-            boolean binaryTransportEnabled,
-            boolean thriftTransportEnabled,
-            boolean taskInfoThriftTransportEnabled,
-            Protocol thriftProtocol,
-            TableWriteInfo tableWriteInfo,
-            int maxTaskUpdateSizeInBytes,
-            MetadataManager metadataManager,
-            QueryManager queryManager,
-            DecayCounter taskUpdateRequestSize,
-            boolean taskUpdateSizeTrackingEnabled,
-            HandleResolver handleResolver,
-            ConnectorTypeSerdeManager connectorTypeSerdeManager,
-            SchedulerStatsTracker schedulerStatsTracker,
-            HttpRemoteTaskFactory.SafeEventLoop taskEventLoop)
+            EventLoop taskEventLoop)
     {
         requireNonNull(session, "session is null");
         requireNonNull(taskId, "taskId is null");
@@ -468,11 +392,7 @@ private HttpRemoteTask(Session session,
                 handleResolver,
                 connectorTypeSerdeManager,
                 thriftProtocol);
-    }
 
-    // this is a separate method to ensure that the `this` reference is not leaked during construction
-    private void initialize()
-    {
         taskStatusFetcher.addStateChangeListener(newStatus -> {
             verify(taskEventLoop.inEventLoop());
 
@@ -487,7 +407,7 @@ private void initialize()
         });
 
         updateTaskStats();
-        safeExecuteOnEventLoop(this::updateSplitQueueSpace);
+        taskEventLoop.execute(this::updateSplitQueueSpace);
     }
 
     public PlanFragment getPlanFragment()
@@ -528,7 +448,7 @@ public URI getRemoteTaskLocation()
     @Override
     public void start()
     {
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             // to start we just need to trigger an update
             started = true;
             scheduleUpdate();
@@ -548,7 +468,7 @@ public void addSplits(Multimap<PlanNodeId, Split> splitsBySource)
             return;
         }
 
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             boolean updateNeeded = false;
             for (Entry<PlanNodeId, Collection<Split>> entry : splitsBySource.asMap().entrySet()) {
                 PlanNodeId sourceId = entry.getKey();
@@ -585,7 +505,7 @@ public void addSplits(Multimap<PlanNodeId, Split> splitsBySource)
     @Override
     public void noMoreSplits(PlanNodeId sourceId)
     {
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             if (noMoreSplits.containsKey(sourceId)) {
                 return;
             }
@@ -599,7 +519,7 @@ public void noMoreSplits(PlanNodeId sourceId)
     @Override
     public void noMoreSplits(PlanNodeId sourceId, Lifespan lifespan)
     {
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             if (pendingNoMoreSplitsForLifespan.put(sourceId, lifespan)) {
                 needsUpdate = true;
                 scheduleUpdate();
@@ -614,7 +534,7 @@ public void setOutputBuffers(OutputBuffers newOutputBuffers)
             return;
         }
 
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             if (newOutputBuffers.getVersion() > outputBuffers.getVersion()) {
                 outputBuffers = newOutputBuffers;
                 needsUpdate = true;
@@ -779,7 +699,7 @@ public ListenableFuture<?> whenSplitQueueHasSpace(long weightThreshold)
             return immediateFuture(null);
         }
         SettableFuture<?> future = SettableFuture.create();
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             if (whenSplitQueueHasSpaceThreshold.isPresent()) {
                 checkArgument(weightThreshold == whenSplitQueueHasSpaceThreshold.getAsLong(), "Multiple split queue space notification thresholds not supported");
             }
@@ -949,7 +869,7 @@ private void scheduleUpdate()
 
     private void sendUpdate()
     {
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             TaskStatus taskStatus = getTaskStatus();
             // don't update if the task hasn't been started yet or if it is already finished
             if (!started || !needsUpdate || taskStatus.getState().isDone()) {
@@ -1072,7 +992,7 @@ private TaskSource getSource(PlanNodeId planNodeId)
     @Override
     public void cancel()
     {
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             TaskStatus taskStatus = getTaskStatus();
             if (taskStatus.getState().isDone()) {
                 return;
@@ -1092,7 +1012,7 @@ public void cancel()
 
     private void cleanUpTask()
     {
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             checkState(getTaskStatus().getState().isDone(), "attempt to clean up a task that is not done yet");
 
             // clear pending splits to free memory
@@ -1140,7 +1060,7 @@ public void abort()
 
     private void abort(TaskStatus status)
     {
-        safeExecuteOnEventLoop(() -> {
+        taskEventLoop.execute(() -> {
             checkState(status.getState().isDone(), "cannot abort task with an incomplete status");
 
             taskStatusFetcher.updateTaskStatus(status);
@@ -1402,19 +1322,4 @@ public void onFailure(Throwable throwable)
             onFailureTaskInfo(throwable, this.action, this.request, this.cleanupBackoff);
         }
     }
-
-    /***
-     *  Wrap the task execution on event loop to fail the entire task on any failure.
-     */
-    private void safeExecuteOnEventLoop(Runnable r)
-    {
-        taskEventLoop.execute(() -> {
-            try {
-                r.run();
-            }
-            catch (Throwable t) {
-                failTask(t);
-            }
-        });
-    }
 }

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

@@ -17,7 +17,6 @@
 import com.facebook.airlift.json.Codec;
 import com.facebook.airlift.json.JsonCodec;
 import com.facebook.airlift.json.smile.SmileCodec;
-import com.facebook.airlift.log.Logger;
 import com.facebook.airlift.stats.DecayCounter;
 import com.facebook.airlift.stats.ExponentialDecay;
 import com.facebook.drift.codec.ThriftCodec;
@@ -50,25 +49,20 @@
 import com.google.common.collect.Multimap;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.airlift.units.Duration;
-import io.netty.channel.DefaultEventLoop;
 import io.netty.channel.DefaultEventLoopGroup;
-import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
 import org.weakref.jmx.Managed;
 
 import javax.annotation.PreDestroy;
 import javax.inject.Inject;
 
-import java.util.concurrent.Executor;
-
 import static com.facebook.presto.server.thrift.ThriftCodecWrapper.wrapThriftCodec;
 import static java.lang.Math.toIntExact;
 import static java.util.Objects.requireNonNull;
 
 public class HttpRemoteTaskFactory
         implements RemoteTaskFactory
 {
-    private static final Logger log = Logger.get(HttpRemoteTaskFactory.class);
     private final HttpClient httpClient;
     private final LocationFactory locationFactory;
     private final Codec<TaskStatus> taskStatusCodec;
@@ -95,7 +89,8 @@ public class HttpRemoteTaskFactory
     private final QueryManager queryManager;
     private final DecayCounter taskUpdateRequestSize;
     private final boolean taskUpdateSizeTrackingEnabled;
-    private final EventLoopGroup eventLoopGroup;
+    private final EventLoopGroup eventLoopGroup = new DefaultEventLoopGroup(Runtime.getRuntime().availableProcessors(),
+            new ThreadFactoryBuilder().setNameFormat("task-event-loop-%s").setDaemon(true).build());
 
     @Inject
     public HttpRemoteTaskFactory(
@@ -174,15 +169,6 @@ 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
@@ -210,7 +196,7 @@ public RemoteTask createRemoteTask(
             TableWriteInfo tableWriteInfo,
             SchedulerStatsTracker schedulerStatsTracker)
     {
-        return HttpRemoteTask.createHttpRemoteTask(
+        return new HttpRemoteTask(
                 session,
                 taskId,
                 node.getNodeIdentifier(),
@@ -246,37 +232,6 @@ public RemoteTask createRemoteTask(
                 handleResolver,
                 connectorTypeSerdeManager,
                 schedulerStatsTracker,
-                (SafeEventLoop) eventLoopGroup.next());
-    }
-
-    /***
-     *  One observation about event loop is if submitted task fails, it could kill the thread but the event loop group will not create a new one.
-     *  Here, we wrap it as safe event loop so that if any submitted job fails, we chose to log the error and fail the entire task.
-     */
-    static class SafeEventLoop
-            extends DefaultEventLoop
-    {
-        public SafeEventLoop(EventLoopGroup parent, Executor executor)
-        {
-            super(parent, executor);
-        }
-
-        @Override
-        protected void run()
-        {
-            do {
-                Runnable task = takeTask();
-                if (task != null) {
-                    try {
-                        runTask(task);
-                    }
-                    catch (Throwable t) {
-                        log.error("Error executing task on event loop", t);
-                    }
-                    updateLastExecutionTime();
-                }
-            }
-            while (!this.confirmShutdown());
-        }
+                eventLoopGroup.next());
     }
 }