maobaolong commented on code in PR #2168:
URL:
https://github.com/apache/incubator-uniffle/pull/2168#discussion_r1802286914
##########
common/src/main/java/org/apache/uniffle/common/executor/ThreadPoolManager.java:
##########
@@ -90,36 +99,145 @@ public static ThreadPoolExecutor newThreadPool(
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
+ return newThreadPool(
+ name,
+ () -> corePoolSize,
+ () -> maximumPoolSize,
+ () -> keepAliveTime,
+ unit,
+ workQueue,
+ threadFactory,
+ handler);
+ }
+
+ /**
+ * Add a thread pool.
+ *
+ * @param name the name of the thread pool
+ * @param corePoolSizeSupplier the core pool size supplier
+ * @param maximumPoolSizeSupplier the maximum pool size supplier
+ * @param keepAliveTimeSupplier the keep alive time supplier
+ * @param unit the unit
+ * @param workQueue the work queue
+ * @param threadFactory the thread factory
+ * @return the registered thread pool
+ */
+ public static ThreadPoolExecutor newThreadPool(
+ String name,
+ Supplier<Integer> corePoolSizeSupplier,
+ Supplier<Integer> maximumPoolSizeSupplier,
+ Supplier<Long> keepAliveTimeSupplier,
+ TimeUnit unit,
+ BlockingQueue<Runnable> workQueue,
+ ThreadFactory threadFactory) {
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
- corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, handler);
- registerThreadPool(name, corePoolSize, maximumPoolSize, keepAliveTime,
threadPoolExecutor);
+ corePoolSizeSupplier.get(),
+ maximumPoolSizeSupplier.get(),
+ keepAliveTimeSupplier.get(),
+ unit,
+ workQueue,
+ threadFactory);
+ registerThreadPool(
+ name,
+ corePoolSizeSupplier,
+ maximumPoolSizeSupplier,
+ () -> keepAliveTimeSupplier.get() * unit.toMillis(1),
+ threadPoolExecutor);
return threadPoolExecutor;
}
/**
- * Register a thread pool to THREAD_POOL_MAP.
+ * Add a thread pool.
+ *
+ * @param name the name of the thread pool
+ * @param corePoolSizeSupplier the core pool size supplier
+ * @param maximumPoolSizeSupplier the maximum pool size supplier
+ * @param keepAliveTimeSupplier the keep alive time supplier
+ * @param unit the unit
+ * @param workQueue the work queue
+ * @param threadFactory the thread factory
+ * @param handler the handler to use when execution is blocked because the
thread bounds and queue
+ * capacities are reached
+ * @return the registered thread pool
+ */
+ public static ThreadPoolExecutor newThreadPool(
+ String name,
+ Supplier<Integer> corePoolSizeSupplier,
+ Supplier<Integer> maximumPoolSizeSupplier,
+ Supplier<Long> keepAliveTimeSupplier,
+ TimeUnit unit,
+ BlockingQueue<Runnable> workQueue,
+ ThreadFactory threadFactory,
+ RejectedExecutionHandler handler) {
+ ThreadPoolExecutor threadPoolExecutor =
+ new ThreadPoolExecutor(
+ corePoolSizeSupplier.get(),
+ maximumPoolSizeSupplier.get(),
+ keepAliveTimeSupplier.get(),
+ unit,
+ workQueue,
+ threadFactory,
+ handler);
+ registerThreadPool(
+ name,
+ corePoolSizeSupplier,
+ maximumPoolSizeSupplier,
+ keepAliveTimeSupplier,
+ threadPoolExecutor);
+ return threadPoolExecutor;
+ }
+
+ /**
+ * Add a thread pool.
*
* @param name the name of the thread pool
- * @param corePoolSize the core pool size supplier
- * @param maximumPoolSize the maximum pool size supplier
- * @param keepAliveTime the keep alive time supplier
+ * @param corePoolSize the core pool size
+ * @param maximumPoolSize the maximum pool size
+ * @param keepAliveTime the keep alive time
* @param threadPoolExecutor the thread pool which will be registered
+ * @return the registered thread pool
*/
+ @VisibleForTesting
public static void registerThreadPool(
String name,
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
ThreadPoolExecutor threadPoolExecutor) {
+ registerThreadPool(
+ name, () -> corePoolSize, () -> maximumPoolSize, () -> keepAliveTime,
threadPoolExecutor);
+ }
+
+ /**
+ * Register a thread pool to THREAD_POOL_MAP.
+ *
+ * @param name the name of the thread pool
+ * @param corePoolSizeSupplier the core pool size supplier
+ * @param maximumPoolSizeSupplier the maximum pool size supplier
+ * @param keepAliveTimeSupplier the keep alive time supplier
+ * @param threadPoolExecutor the thread pool which will be registered
+ */
+ public static void registerThreadPool(
+ String name,
+ Supplier<Integer> corePoolSizeSupplier,
+ Supplier<Integer> maximumPoolSizeSupplier,
+ Supplier<Long> keepAliveTimeSupplier,
+ ThreadPoolExecutor threadPoolExecutor) {
THREAD_POOL_MAP.put(
- threadPoolExecutor, new MeasurableThreadPoolExecutor(name,
threadPoolExecutor));
+ threadPoolExecutor,
+ new MeasurableThreadPoolExecutor(
+ name,
+ threadPoolExecutor,
+ corePoolSizeSupplier,
+ maximumPoolSizeSupplier,
+ keepAliveTimeSupplier));
LOG.info(
"{} thread pool, core size:{}, max size:{}, keep alive time:{}",
name,
- corePoolSize,
- maximumPoolSize,
- keepAliveTime);
+ corePoolSizeSupplier,
Review Comment:
Done.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
