001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, 013 * software distributed under the License is distributed on an 014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 015 * KIND, either express or implied. See the License for the 016 * specific language governing permissions and limitations 017 * under the License. 018 */ 019package org.eclipse.aether.util.concurrency; 020 021import java.util.concurrent.Callable; 022import java.util.concurrent.CompletableFuture; 023import java.util.concurrent.ExecutorService; 024import java.util.concurrent.Future; 025import java.util.concurrent.RejectedExecutionException; 026import java.util.concurrent.Semaphore; 027 028/** 029 * Utilities for executors and sizing them. 030 * <em>Big fat note:</em> Do not use this class outside of resolver. This and related classes are not meant as "drop 031 * in replacement" for Jave Executors, is used in very controlled fashion only. 032 * 033 * @since 2.0.11 034 */ 035public interface SmartExecutor extends AutoCloseable { 036 /** 037 * Submits a {@link Runnable} to execution. 038 * 039 * @throws RejectedExecutionException If this executor cannot accept the task. 040 */ 041 void submit(Runnable runnable) throws RejectedExecutionException; 042 043 /** 044 * Submits a {@link Callable} to execution, returns a {@link CompletableFuture}. 045 */ 046 <T> Future<T> submit(Callable<T> callable); 047 048 /** 049 * Shut down this instance (ideally used in try-with-resource construct). 050 */ 051 void close(); 052 053 /** 054 * Direct executor (caller executes). 055 */ 056 class Direct implements SmartExecutor { 057 @Override 058 public void submit(Runnable runnable) { 059 runnable.run(); 060 } 061 062 @Override 063 public <T> CompletableFuture<T> submit(Callable<T> callable) { 064 CompletableFuture<T> future = new CompletableFuture<>(); 065 try { 066 future.complete(callable.call()); 067 } catch (Exception e) { 068 future.completeExceptionally(e); 069 } 070 return future; 071 } 072 073 @Override 074 public void close() {} 075 } 076 077 /** 078 * Pooled executor backed by {@link ExecutorService}. 079 */ 080 class Pooled implements SmartExecutor { 081 private final ExecutorService executor; 082 083 Pooled(ExecutorService executor) { 084 this.executor = executor; 085 } 086 087 @Override 088 public void submit(Runnable runnable) { 089 ClassLoader tccl = Thread.currentThread().getContextClassLoader(); 090 try { 091 executor.submit(() -> { 092 ClassLoader old = Thread.currentThread().getContextClassLoader(); 093 Thread.currentThread().setContextClassLoader(tccl); 094 try { 095 runnable.run(); 096 } finally { 097 Thread.currentThread().setContextClassLoader(old); 098 } 099 }); 100 } catch (RejectedExecutionException e) { 101 try { 102 runnable.run(); 103 } catch (RuntimeException | Error t) { 104 // swallow to match async submit() semantics where exceptions 105 // are captured by the Future; callers like RunnableErrorForwarder 106 // already record the error before re-throwing 107 } 108 } 109 } 110 111 @Override 112 public <T> Future<T> submit(Callable<T> callable) { 113 ClassLoader tccl = Thread.currentThread().getContextClassLoader(); 114 CompletableFuture<T> future = new CompletableFuture<>(); 115 try { 116 executor.submit(() -> { 117 ClassLoader old = Thread.currentThread().getContextClassLoader(); 118 Thread.currentThread().setContextClassLoader(tccl); 119 try { 120 future.complete(callable.call()); 121 } catch (Exception e) { 122 future.completeExceptionally(e); 123 } finally { 124 Thread.currentThread().setContextClassLoader(old); 125 } 126 }); 127 } catch (RejectedExecutionException e) { 128 try { 129 future.complete(callable.call()); 130 } catch (Exception ex) { 131 future.completeExceptionally(ex); 132 } 133 } 134 return future; 135 } 136 137 @Override 138 public void close() { 139 executor.shutdown(); 140 } 141 } 142 143 /** 144 * Limited executor, where the actual goal is to protect accessed resource, like when virtual threads 145 * are being used, so the "pool" itself does not provide any kind of back-pressure. 146 */ 147 class Limited implements SmartExecutor { 148 private final SmartExecutor executor; 149 private final Semaphore semaphore; 150 151 Limited(SmartExecutor executor, int limit) { 152 this.executor = executor; 153 this.semaphore = new Semaphore(limit); 154 } 155 156 @Override 157 public void submit(Runnable runnable) { 158 try { 159 semaphore.acquire(); 160 try { 161 executor.submit(() -> { 162 try { 163 runnable.run(); 164 } finally { 165 semaphore.release(); 166 } 167 }); 168 } catch (RejectedExecutionException e) { 169 try { 170 runnable.run(); 171 } catch (RuntimeException | Error t) { 172 // swallow to match async submit() semantics where exceptions 173 // are captured by the Future; callers like RunnableErrorForwarder 174 // already record the error before re-throwing 175 } finally { 176 semaphore.release(); 177 } 178 } 179 } catch (InterruptedException e) { 180 Thread.currentThread().interrupt(); 181 throw new RejectedExecutionException(e); 182 } 183 } 184 185 @Override 186 public <T> Future<T> submit(Callable<T> callable) { 187 try { 188 semaphore.acquire(); 189 CompletableFuture<T> future = new CompletableFuture<>(); 190 try { 191 executor.submit(() -> { 192 try { 193 future.complete(callable.call()); 194 } catch (Exception e) { 195 future.completeExceptionally(e); 196 } finally { 197 semaphore.release(); 198 } 199 }); 200 } catch (RejectedExecutionException e) { 201 try { 202 future.complete(callable.call()); 203 } catch (Exception ex) { 204 future.completeExceptionally(ex); 205 } finally { 206 semaphore.release(); 207 } 208 } 209 return future; 210 } catch (InterruptedException e) { 211 Thread.currentThread().interrupt(); 212 CompletableFuture<T> failed = new CompletableFuture<>(); 213 failed.completeExceptionally(e); 214 return failed; 215 } 216 } 217 218 @Override 219 public void close() { 220 executor.close(); 221 } 222 } 223 224 /** 225 * Wrapper to prevent closing. 226 */ 227 class NonClosing implements SmartExecutor { 228 private final SmartExecutor smartExecutor; 229 230 NonClosing(SmartExecutor smartExecutor) { 231 this.smartExecutor = smartExecutor; 232 } 233 234 @Override 235 public void submit(Runnable runnable) { 236 smartExecutor.submit(runnable); 237 } 238 239 @Override 240 public <T> Future<T> submit(Callable<T> callable) { 241 return smartExecutor.submit(callable); 242 } 243 244 @Override 245 public void close() { 246 // nope; delegate is managed 247 } 248 } 249}