[GitHub] [flink] pnowojski commented on a change in pull request #8052: [FLINK-11458][checkpointing, rest] Add TERMINATE/SUSPEND Job with Savepoint (FLIP-34)

[GitBox <gi...@apache.org>]

pnowojski commented on a change in pull request #8052: [FLINK-11458][checkpointing, rest] Add TERMINATE/SUSPEND Job with Savepoint (FLIP-34)
URL: https://github.com/apache/flink/pull/8052#discussion_r275413044
 
 

 ##########
 File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/BlockingCallMonitoringThreadPool.java
 ##########
 @@ -0,0 +1,111 @@
+/*
+ * 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.flink.runtime.taskmanager;
+
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A Thread Pool used by the {@link Task} to launch checkpointing tasks, namely
+ * the {@link Task#triggerCheckpointBarrier(long, long, CheckpointOptions, boolean)}
+ * and the {@link Task#notifyCheckpointComplete(long)} callback.
+ *
+ * <p>This thread pool monitors the number of in-flight calls executing a {@link CheckpointType#SYNC_SAVEPOINT},
+ * and adjusts its pool size accordingly, so that the minimum number of threads are allocated. This is to
+ * avoid any redundant idle threads consuming resources. In addition,this thread pool logs (in debug level)
+ * when more than 1 such calls are in-flight.
+ */
+public class BlockingCallMonitoringThreadPool {
+
+	private static final Logger LOG = LoggerFactory.getLogger(BlockingCallMonitoringThreadPool.class);
+
+	private final AtomicInteger inFlightBlockingCallCounter = new AtomicInteger(0);
+
+	private final ThreadPoolExecutor executor;
+
+	public BlockingCallMonitoringThreadPool(final ThreadFactory dispatcherThreadFactory) {
+		this.executor = new ThreadPoolExecutor(
+				1,
+				1,
+				10L,
+				TimeUnit.SECONDS,
+				new LinkedBlockingQueue<>(),
+				checkNotNull(dispatcherThreadFactory));
+	}
+
+	public void submit(final Runnable runnable, final boolean blocking) {
+		if (blocking) {
+			submitBlocking(runnable);
+		} else {
+			submit(runnable);
+		}
+	}
+
+	private void submit(final Runnable task) {
+		adjustThreadPoolSize(inFlightBlockingCallCounter.get());
+		executor.execute(task);
+	}
+
+	private void submitBlocking(final Runnable task) {
+		adjustThreadPoolSize(inFlightBlockingCallCounter.incrementAndGet());
+		CompletableFuture.runAsync(task, executor).whenComplete(
+				(ignored, e) -> inFlightBlockingCallCounter.decrementAndGet());
+	}
+
+	private void adjustThreadPoolSize(final int activeBlockingCalls) {
+		if (activeBlockingCalls > 1) {
+			LOG.debug("There are {} active threads with blocking calls", activeBlockingCalls);
+		}
+
+		final int newPoolSize = executor.getCorePoolSize() + activeBlockingCalls;
+
+		// We have to reset the core pool size because (quoted from javadoc):
+		// ``
+		// If there are more than corePoolSize but less than maximumPoolSize threads running,
+		// ** a new thread will be created only if the queue is full **.
+		// ``
+
+		executor.setCorePoolSize(newPoolSize);
 
 Review comment:
   You shouldn't be increasing core pool size:
   1. in order to let the `ThreadPoolExecutor` shutdown idling threads more quickly
   2. currently `newPoolSize` is initialised from `executor.getCorePoolSize()` so you would exponentially increase the pool size with subsequent `executeBlocking` calls: 1, 2, 4, 7, 11, 16, ...
   
   I think you should just keep the `setCorePoolSize` size constant.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services