CE-182 ftp connection pool

oap-stdlib-test/src/test/java/oap/concurrent/ExecutorsTest.java

@@ -27,22 +27,26 @@
 import org.testng.annotations.Test;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.IntStream;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 public class ExecutorsTest {
     @Test
     public void test() throws InterruptedException, ExecutionException, TimeoutException {
-        var executor = Executors.newFixedBlockingThreadPool( 2 );
+        ThreadPoolExecutor executor = Executors.newFixedBlockingThreadPool( 2 );
 
-        var c = new AtomicInteger();
+        AtomicInteger c = new AtomicInteger();
 
-        var start = System.currentTimeMillis();
+        long start = System.currentTimeMillis();
 
-        var f = IntStream.range( 0, 5 ).mapToObj( i -> {
+        CompletableFuture[] f = IntStream.range( 0, 5 ).mapToObj( i -> {
             System.out.println( ( System.currentTimeMillis() - start ) + " prerun - " + i );
             return CompletableFuture.runAsync( () -> {
                 System.out.println( ( System.currentTimeMillis() - start ) + " run - " + i + " -> " + Thread.currentThread().getName() + "..." );
@@ -55,4 +59,36 @@ public void test() throws InterruptedException, ExecutionException, TimeoutExcep
         CompletableFuture.allOf( f ).get( 10, TimeUnit.SECONDS );
     }
 
+    @Test
+    public void testNewFixedBlockingVirtualThreadPerTaskExecutorLimitsConcurrency() throws InterruptedException {
+        int threads = 4;
+        int tasks = 20;
+
+        ExecutorService executor = Executors.newFixedBlockingVirtualThreadPerTaskExecutor( threads );
+
+        AtomicInteger running = new AtomicInteger();
+        AtomicInteger maxRunning = new AtomicInteger();
+        CountDownLatch done = new CountDownLatch( tasks );
+
+        for( int i = 0; i < tasks; i++ ) {
+            executor.execute( () -> {
+                int current = running.incrementAndGet();
+                maxRunning.updateAndGet( max -> Math.max( max, current ) );
+                try {
+                    Thread.sleep( 20 );
+                } catch( InterruptedException e ) {
+                    Thread.currentThread().interrupt();
+                } finally {
+                    running.decrementAndGet();
+                    done.countDown();
+                }
+            } );
+        }
+
+        assertThat( done.await( 10, TimeUnit.SECONDS ) ).isTrue();
+        assertThat( maxRunning.get() ).isLessThanOrEqualTo( threads );
+
+        executor.shutdown();
+        assertThat( executor.awaitTermination( 10, TimeUnit.SECONDS ) ).isTrue();
+    }
 }

oap-stdlib/src/main/java/oap/concurrent/Executors.java

@@ -75,6 +75,10 @@ public static ThreadPoolExecutor newFixedBlockingThreadPool( int nThreads, Threa
             0, TimeUnit.SECONDS, new SynchronousQueue<>(), threadFactory, new ThreadPoolExecutor.BlockingPolicy() );
     }
 
+    public static ExecutorService newFixedBlockingVirtualThreadPerTaskExecutor( int threads ) {
+        return new FixedBlockingVirtualThreadPerTaskExecutor( threads );
+    }
+
     public static ScheduledExecutorService newScheduledThreadPool( int corePoolSize, String threadPrefix ) {
         return new ScheduledExecutorService( java.util.concurrent.Executors.newScheduledThreadPool(
             corePoolSize,

oap-stdlib/src/main/java/oap/concurrent/FixedBlockingVirtualThreadPerTaskExecutor.java

@@ -0,0 +1,88 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) Open Application Platform Authors
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package oap.concurrent;
+
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Virtual-thread-per-task executor bounded to a fixed number of concurrently running tasks.
+ * {@code newVirtualThreadPerTaskExecutor()} spawns an unbounded number of threads, so a
+ * {@link Semaphore} gates task start: {@code execute}/{@code submit}/{@code invokeAll}/{@code invokeAny}
+ * block while {@code threads} tasks are already running, and unblock as tasks complete.
+ */
+class FixedBlockingVirtualThreadPerTaskExecutor extends AbstractExecutorService {
+    private final ExecutorService delegate = java.util.concurrent.Executors.newVirtualThreadPerTaskExecutor();
+    private final Semaphore semaphore;
+
+    FixedBlockingVirtualThreadPerTaskExecutor( int threads ) {
+        this.semaphore = new Semaphore( threads );
+    }
+
+    @Override
+    public void execute( Runnable command ) {
+        semaphore.acquireUninterruptibly();
+        try {
+            delegate.execute( () -> {
+                try {
+                    command.run();
+                } finally {
+                    semaphore.release();
+                }
+            } );
+        } catch( RuntimeException | Error e ) {
+            semaphore.release();
+            throw e;
+        }
+    }
+
+    @Override
+    public void shutdown() {
+        delegate.shutdown();
+    }
+
+    @Override
+    public List<Runnable> shutdownNow() {
+        return delegate.shutdownNow();
+    }
+
+    @Override
+    public boolean isShutdown() {
+        return delegate.isShutdown();
+    }
+
+    @Override
+    public boolean isTerminated() {
+        return delegate.isTerminated();
+    }
+
+    @Override
+    public boolean awaitTermination( long timeout, TimeUnit unit ) throws InterruptedException {
+        return delegate.awaitTermination( timeout, unit );
+    }
+}

oap-storage/oap-storage-cloud-ftp/pom.xml

@@ -24,6 +24,12 @@
             <artifactId>commons-net</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-pool2</artifactId>
+            <version>2.12.1</version>
+        </dependency>
+
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>