View Javadoc
1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *   http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing,
13   * software distributed under the License is distributed on an
14   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15   * KIND, either express or implied.  See the License for the
16   * specific language governing permissions and limitations
17   * under the License.
18   */
19  package org.eclipse.aether.util.concurrency;
20  
21  import java.util.concurrent.Callable;
22  import java.util.concurrent.CompletableFuture;
23  import java.util.concurrent.ExecutorService;
24  import java.util.concurrent.Future;
25  import java.util.concurrent.RejectedExecutionException;
26  import java.util.concurrent.Semaphore;
27  
28  /**
29   * Utilities for executors and sizing them.
30   * <em>Big fat note:</em> Do not use this class outside of resolver. This and related classes are not meant as "drop
31   * in replacement" for Jave Executors, is used in very controlled fashion only.
32   *
33   * @since 2.0.11
34   */
35  public interface SmartExecutor extends AutoCloseable {
36      /**
37       * Submits a {@link Runnable} to execution.
38       *
39       * @throws RejectedExecutionException If this executor cannot accept the task.
40       */
41      void submit(Runnable runnable) throws RejectedExecutionException;
42  
43      /**
44       * Submits a {@link Callable} to execution, returns a {@link CompletableFuture}.
45       */
46      <T> Future<T> submit(Callable<T> callable);
47  
48      /**
49       * Shut down this instance (ideally used in try-with-resource construct).
50       */
51      void close();
52  
53      /**
54       * Direct executor (caller executes).
55       */
56      class Direct implements SmartExecutor {
57          @Override
58          public void submit(Runnable runnable) {
59              runnable.run();
60          }
61  
62          @Override
63          public <T> CompletableFuture<T> submit(Callable<T> callable) {
64              CompletableFuture<T> future = new CompletableFuture<>();
65              try {
66                  future.complete(callable.call());
67              } catch (Exception e) {
68                  future.completeExceptionally(e);
69              }
70              return future;
71          }
72  
73          @Override
74          public void close() {}
75      }
76  
77      /**
78       * Pooled executor backed by {@link ExecutorService}.
79       */
80      class Pooled implements SmartExecutor {
81          private final ExecutorService executor;
82  
83          Pooled(ExecutorService executor) {
84              this.executor = executor;
85          }
86  
87          @Override
88          public void submit(Runnable runnable) {
89              ClassLoader tccl = Thread.currentThread().getContextClassLoader();
90              try {
91                  executor.submit(() -> {
92                      ClassLoader old = Thread.currentThread().getContextClassLoader();
93                      Thread.currentThread().setContextClassLoader(tccl);
94                      try {
95                          runnable.run();
96                      } finally {
97                          Thread.currentThread().setContextClassLoader(old);
98                      }
99                  });
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 }