diff --git a/oap-stdlib-test/src/test/java/oap/concurrent/ExecutorsTest.java b/oap-stdlib-test/src/test/java/oap/concurrent/ExecutorsTest.java index 52666c11e8..078bac9715 100644 --- a/oap-stdlib-test/src/test/java/oap/concurrent/ExecutorsTest.java +++ b/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(); + } } diff --git a/oap-stdlib/src/main/java/oap/concurrent/Executors.java b/oap-stdlib/src/main/java/oap/concurrent/Executors.java index 715acc7c14..48839d962d 100644 --- a/oap-stdlib/src/main/java/oap/concurrent/Executors.java +++ b/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, diff --git a/oap-stdlib/src/main/java/oap/concurrent/FixedBlockingVirtualThreadPerTaskExecutor.java b/oap-stdlib/src/main/java/oap/concurrent/FixedBlockingVirtualThreadPerTaskExecutor.java new file mode 100644 index 0000000000..bb9e9a3ee4 --- /dev/null +++ b/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 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 ); + } +} diff --git a/oap-storage/oap-storage-cloud-ftp/pom.xml b/oap-storage/oap-storage-cloud-ftp/pom.xml index 766b386938..899879ff9d 100644 --- a/oap-storage/oap-storage-cloud-ftp/pom.xml +++ b/oap-storage/oap-storage-cloud-ftp/pom.xml @@ -24,6 +24,12 @@ commons-net + + org.apache.commons + commons-pool2 + 2.12.1 + + org.projectlombok lombok diff --git a/oap-storage/oap-storage-cloud-ftp/src/main/java/oap/storage/cloud/ftp/AbstractFileSystemCloudApiFtp.java b/oap-storage/oap-storage-cloud-ftp/src/main/java/oap/storage/cloud/ftp/AbstractFileSystemCloudApiFtp.java index ab6b6cce8d..0ce1da3f41 100644 --- a/oap-storage/oap-storage-cloud-ftp/src/main/java/oap/storage/cloud/ftp/AbstractFileSystemCloudApiFtp.java +++ b/oap-storage/oap-storage-cloud-ftp/src/main/java/oap/storage/cloud/ftp/AbstractFileSystemCloudApiFtp.java @@ -14,6 +14,8 @@ import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; import org.apache.commons.net.ftp.FTPReply; +import org.apache.commons.pool2.impl.GenericObjectPool; +import org.apache.commons.pool2.impl.GenericObjectPoolConfig; import org.joda.time.DateTime; import org.joda.time.DateTimeZone; @@ -28,6 +30,7 @@ import java.nio.ByteBuffer; import java.nio.file.Files; import java.nio.file.Path; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; @@ -40,6 +43,9 @@ @Slf4j public abstract class AbstractFileSystemCloudApiFtp implements FileSystemCloudApi { + private static final int DEFAULT_POOL_MAX_SIZE = 8; + private static final long DEFAULT_POOL_MAX_WAIT_MILLIS = 30_000; + protected final String host; protected final int port; protected final String username; @@ -47,6 +53,8 @@ public abstract class AbstractFileSystemCloudApiFtp implements FileSystemCloudAp protected final boolean passiveMode; protected final boolean removeEmptyFolders; + private final GenericObjectPool pool; + protected AbstractFileSystemCloudApiFtp( FileSystemConfiguration fileSystemConfiguration, String scheme, String container ) { Object hostObj = fileSystemConfiguration.get( scheme, container, "jclouds.host" ); if( hostObj == null ) { @@ -68,6 +76,20 @@ protected AbstractFileSystemCloudApiFtp( FileSystemConfiguration fileSystemConfi Object removeEmptyFolders = fileSystemConfiguration.get( scheme, container, "jclouds.remove-empty-folders" ); this.removeEmptyFolders = removeEmptyFolders != null && Boolean.parseBoolean( removeEmptyFolders.toString() ); + + Object poolMaxSizeObj = fileSystemConfiguration.get( scheme, container, "jclouds.pool-max-size" ); + int poolMaxSize = poolMaxSizeObj != null ? Integer.parseInt( poolMaxSizeObj.toString() ) : DEFAULT_POOL_MAX_SIZE; + + Object poolMaxWaitObj = fileSystemConfiguration.get( scheme, container, "jclouds.pool-max-wait-millis" ); + long poolMaxWaitMillis = poolMaxWaitObj != null ? Long.parseLong( poolMaxWaitObj.toString() ) : DEFAULT_POOL_MAX_WAIT_MILLIS; + + GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig<>(); + poolConfig.setMaxTotal( poolMaxSize ); + poolConfig.setMaxWait( Duration.ofMillis( poolMaxWaitMillis ) ); + poolConfig.setBlockWhenExhausted( true ); + poolConfig.setTestOnBorrow( true ); + + this.pool = new GenericObjectPool<>( new FtpClientPooledObjectFactory( this ), poolConfig ); } protected abstract FTPClient createClient() throws IOException; @@ -75,7 +97,7 @@ protected AbstractFileSystemCloudApiFtp( FileSystemConfiguration fileSystemConfi protected void afterLogin( FTPClient client ) throws IOException { } - protected FTPClient connect() { + FTPClient createAndLoginClient() { try { FTPClient client = createClient(); client.connect( host, port ); @@ -105,6 +127,28 @@ protected FTPClient connect() { } } + private FTPClient borrow() { + try { + return pool.borrowObject(); + } catch( CloudException e ) { + throw e; + } catch( Exception e ) { + throw new CloudException( e ); + } + } + + private void release( FTPClient client, boolean healthy ) { + try { + if( healthy ) { + pool.returnObject( client ); + } else { + pool.invalidateObject( client ); + } + } catch( Exception e ) { + log.debug( "error releasing ftp client", e ); + } + } + protected static void disconnect( FTPClient client ) { try { if( client.isConnected() ) { @@ -190,21 +234,24 @@ private URI buildUri( CloudURI path ) { @Override public CompletableFuture blobExistsAsync( CloudURI path ) { - FTPClient client = connect(); + FTPClient client = borrow(); + boolean healthy = false; try { - return CompletableFuture.completedFuture( findFile( client, absolute( path.path ) ) != null ); + boolean exists = findFile( client, absolute( path.path ) ) != null; + healthy = true; + return CompletableFuture.completedFuture( exists ); } catch( IOException e ) { return CompletableFuture.failedFuture( new CloudException( e ) ); } finally { - disconnect( client ); + release( client, healthy ); } } @Override public CompletableFuture containerExistsAsync( CloudURI path ) { try { - FTPClient client = connect(); - disconnect( client ); + FTPClient client = borrow(); + release( client, true ); return CompletableFuture.completedFuture( true ); } catch( CloudException e ) { return CompletableFuture.completedFuture( false ); @@ -213,9 +260,11 @@ public CompletableFuture containerExistsAsync( CloudURI path ) { @Override public CompletableFuture deleteBlobAsync( CloudURI path ) { - FTPClient client = connect(); + FTPClient client = borrow(); + boolean healthy = false; try { if( !client.deleteFile( absolute( path.path ) ) ) { + healthy = true; return CompletableFuture.failedFuture( new CloudException( "cannot delete " + path ) ); } @@ -223,11 +272,12 @@ public CompletableFuture deleteBlobAsync( CloudURI path ) { removeEmptyParents( client, parentOf( absolute( path.path ) ) ); } + healthy = true; return CompletableFuture.completedFuture( null ); } catch( IOException e ) { return CompletableFuture.failedFuture( new CloudException( e ) ); } finally { - disconnect( client ); + release( client, healthy ); } } @@ -262,22 +312,25 @@ public CompletableFuture deleteContainerIfEmptyAsync( CloudURI path ) { @Override public CompletableFuture getMetadataAsync( CloudURI path ) { - FTPClient client = connect(); + FTPClient client = borrow(); + boolean healthy = false; try { FTPFile file = findFile( client, absolute( path.path ) ); + healthy = true; if( file == null ) return CompletableFuture.completedFuture( null ); return CompletableFuture.completedFuture( toStorageItem( path, file ) ); } catch( IOException e ) { return CompletableFuture.failedFuture( new CloudException( e ) ); } finally { - disconnect( client ); + release( client, healthy ); } } @Override public CompletableFuture downloadFileAsync( CloudURI source, Path destination ) { - FTPClient client = connect(); + FTPClient client = borrow(); + boolean healthy = false; try { oap.io.Files.ensureFile( destination ); try( OutputStream out = Files.newOutputStream( destination ) ) { @@ -285,11 +338,12 @@ public CompletableFuture downloadFileAsync( CloudURI source, Path destinat return CompletableFuture.failedFuture( new CloudException( "cannot download " + source ) ); } } + healthy = true; return CompletableFuture.completedFuture( null ); } catch( IOException e ) { return CompletableFuture.failedFuture( new CloudException( e ) ); } finally { - disconnect( client ); + release( client, healthy ); } } @@ -297,11 +351,15 @@ public CompletableFuture downloadFileAsync( CloudURI source, Path destinat public CompletableFuture copyAsync( CloudURI source, CloudURI destination ) { Preconditions.checkArgument( source.scheme.equals( destination.scheme ) ); - FTPClient sourceClient = connect(); - FTPClient destinationClient = connect(); + FTPClient sourceClient = borrow(); + FTPClient destinationClient = borrow(); + boolean sourceHealthy = false; + boolean destinationHealthy = false; try { InputStream in = sourceClient.retrieveFileStream( absolute( source.path ) ); if( in == null ) { + sourceHealthy = true; + destinationHealthy = true; return CompletableFuture.failedFuture( new CloudException( "cannot open source stream " + source ) ); } @@ -310,7 +368,11 @@ public CompletableFuture copyAsync( CloudURI source, CloudURI destination boolean stored = destinationClient.storeFile( absolute( destination.path ), in ); in.close(); - if( !stored || !sourceClient.completePendingCommand() ) { + boolean completed = sourceClient.completePendingCommand(); + sourceHealthy = completed; + destinationHealthy = stored; + + if( !stored || !completed ) { return CompletableFuture.failedFuture( new CloudException( "cannot copy " + source + " to " + destination ) ); } @@ -318,48 +380,49 @@ public CompletableFuture copyAsync( CloudURI source, CloudURI destination } catch( IOException e ) { return CompletableFuture.failedFuture( new CloudException( e ) ); } finally { - disconnect( sourceClient ); - disconnect( destinationClient ); + release( sourceClient, sourceHealthy ); + release( destinationClient, destinationHealthy ); } } @Override public CompletableFuture getInputStreamAsync( CloudURI path ) { - FTPClient client = connect(); + FTPClient client = borrow(); try { InputStream in = client.retrieveFileStream( absolute( path.path ) ); if( in == null ) { - disconnect( client ); + release( client, true ); return CompletableFuture.failedFuture( new CloudException( "cannot open " + path ) ); } - return CompletableFuture.completedFuture( new FtpInputStream( client, in ) ); + return CompletableFuture.completedFuture( new FtpInputStream( this, client, in ) ); } catch( IOException e ) { - disconnect( client ); + release( client, false ); return CompletableFuture.failedFuture( new CloudException( e ) ); } } @Override public OutputStream getOutputStream( CloudURI path, Map tags ) { - FTPClient client = connect(); + FTPClient client = borrow(); try { ensureRemoteDirectory( client, parentOf( absolute( path.path ) ) ); OutputStream out = client.storeFileStream( absolute( path.path ) ); if( out == null ) { - disconnect( client ); + release( client, true ); throw new CloudException( "cannot open output stream for " + path ); } - return new FtpOutputStream( client, out ); + return new FtpOutputStream( this, client, out ); } catch( IOException e ) { - disconnect( client ); + release( client, false ); throw new CloudException( e ); } } @Override public CompletableFuture uploadAsync( CloudURI destination, BlobData blobData ) { - FTPClient client = connect(); + FTPClient client = borrow(); + boolean healthy = false; try { ensureRemoteDirectory( client, parentOf( absolute( destination.path ) ) ); @@ -387,17 +450,19 @@ public CompletableFuture uploadAsync( CloudURI destination, BlobData blobD return CompletableFuture.failedFuture( new CloudException( "cannot upload to " + destination ) ); } + healthy = true; return CompletableFuture.completedFuture( null ); } catch( IOException e ) { return CompletableFuture.failedFuture( new CloudException( e ) ); } finally { - disconnect( client ); + release( client, healthy ); } } @Override public CompletableFuture> listAsync( CloudURI path, ListOptions listOptions ) { - FTPClient client = connect(); + FTPClient client = borrow(); + boolean healthy = false; try { List all = new ArrayList<>(); walk( client, path, absolute( path.path ), all ); @@ -416,11 +481,12 @@ public CompletableFuture> listAsync( C String nextToken = listOptions.maxKeys != null ? String.valueOf( skip + result.size() ) : null; + healthy = true; return CompletableFuture.completedFuture( new PageSet<>( nextToken, result ) ); } catch( IOException e ) { return CompletableFuture.failedFuture( new CloudException( e ) ); } finally { - disconnect( client ); + release( client, healthy ); } } @@ -445,13 +511,16 @@ private void walk( FTPClient client, CloudURI base, String dirPath, List { + private final AbstractFileSystemCloudApiFtp owner; + + FtpClientPooledObjectFactory( AbstractFileSystemCloudApiFtp owner ) { + this.owner = owner; + } + + @Override + public FTPClient create() { + return owner.createAndLoginClient(); + } + + @Override + public PooledObject wrap( FTPClient client ) { + return new DefaultPooledObject<>( client ); + } + + @Override + public void destroyObject( PooledObject pooledObject ) { + AbstractFileSystemCloudApiFtp.disconnect( pooledObject.getObject() ); + } + + @Override + public boolean validateObject( PooledObject pooledObject ) { + FTPClient client = pooledObject.getObject(); + try { + return client.isConnected() && client.sendNoOp(); + } catch( IOException e ) { + return false; + } + } +} diff --git a/oap-storage/oap-storage-cloud-test/src/main/java/oap/storage/cloud/FtpFixture.java b/oap-storage/oap-storage-cloud-test/src/main/java/oap/storage/cloud/FtpFixture.java index e277227a8b..8be4dc408b 100644 --- a/oap-storage/oap-storage-cloud-test/src/main/java/oap/storage/cloud/FtpFixture.java +++ b/oap-storage/oap-storage-cloud-test/src/main/java/oap/storage/cloud/FtpFixture.java @@ -117,6 +117,10 @@ public FileSystemConfiguration getFileSystemConfiguration( @Nullable String cont } public FileSystemConfiguration getFileSystemConfiguration( @Nullable String container, boolean removeEmptyFolders ) { + return getFileSystemConfiguration( container, removeEmptyFolders, null ); + } + + public FileSystemConfiguration getFileSystemConfiguration( @Nullable String container, boolean removeEmptyFolders, @Nullable Integer poolMaxSize ) { String scheme = tls ? "ftps" : "ftp"; LinkedHashMap map = new LinkedHashMap<>(); @@ -130,6 +134,10 @@ public FileSystemConfiguration getFileSystemConfiguration( @Nullable String cont map.put( "fs." + scheme + ".clouds.remove-empty-folders", true ); } + if( poolMaxSize != null ) { + map.put( "fs." + scheme + ".clouds.pool-max-size", poolMaxSize ); + } + map.put( "fs.default.clouds.scheme", scheme ); if( container != null ) { map.put( "fs.default.clouds.container", container ); diff --git a/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFtpTest.java b/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFtpTest.java index c7a2d3ac11..9a379e13f1 100644 --- a/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFtpTest.java +++ b/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFtpTest.java @@ -17,7 +17,12 @@ import java.io.InputStream; import java.io.OutputStream; import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import static java.nio.charset.StandardCharsets.UTF_8; @@ -213,4 +218,74 @@ public void testDeleteFileAndParentFolderIfEmpty() { assertThat( ftpFixture.resolve( "case3/folder1" ) ).doesNotExist(); } } + + @Test + public void testPoolReusesConnectionSequentially() { + ftpFixture.writeFile( "logs/file1.txt", "1" ); + + try( FileSystem fileSystem = new FileSystem( ftpFixture.getFileSystemConfiguration( null, false, 1 ) ) ) { + for( int i = 0; i < 5; i++ ) { + assertTrue( fileSystem.blobExists( new CloudURI( "ftp://logs/file1.txt" ) ) ); + } + } + } + + @Test + public void testPoolHandlesConcurrentUploads() { + int poolMaxSize = 2; + int uploads = 10; + + try( FileSystem fileSystem = new FileSystem( ftpFixture.getFileSystemConfiguration( null, false, poolMaxSize ) ) ) { + ExecutorService executor = Executors.newFixedThreadPool( uploads ); + try { + List> futures = new ArrayList<>(); + for( int i = 0; i < uploads; i++ ) { + int idx = i; + futures.add( CompletableFuture.runAsync( () -> + fileSystem.upload( new CloudURI( "ftp://concurrent/file" + idx + ".txt" ), + BlobData.builder().content( "content" + idx ).build() ), executor ) ); + } + + assertThat( CompletableFuture.allOf( futures.toArray( new CompletableFuture[0] ) ) ) + .succeedsWithin( 30, TimeUnit.SECONDS ); + } finally { + executor.shutdown(); + } + } + + for( int i = 0; i < uploads; i++ ) { + assertThat( ftpFixture.readFile( "concurrent/file" + i + ".txt" ) ).isEqualTo( "content" + i ); + } + } + + @Test + public void testPoolHandlesManyParallelUploads() { + int poolMaxSize = 8; + int uploads = 1000; + + try( FileSystem fileSystem = new FileSystem( ftpFixture.getFileSystemConfiguration( null, false, poolMaxSize ) ) ) { + ExecutorService executor = Executors.newFixedThreadPool( 50 ); + try { + List> futures = new ArrayList<>(); + for( int i = 0; i < uploads; i++ ) { + int idx = i; + futures.add( CompletableFuture.runAsync( () -> + fileSystem.upload( new CloudURI( "ftp://bulk/file" + idx + ".txt" ), + BlobData.builder().content( "content" + idx ).build() ), executor ) ); + } + + assertThat( CompletableFuture.allOf( futures.toArray( new CompletableFuture[0] ) ) ) + .succeedsWithin( 120, TimeUnit.SECONDS ); + } finally { + executor.shutdown(); + } + + PageSet list = fileSystem.list( new CloudURI( "ftp://bulk/" ), ListOptions.builder().build() ); + assertThat( list.size() ).isEqualTo( uploads ); + } + + for( int idx : new int[] { 0, 1, 500, 998, 999 } ) { + assertThat( ftpFixture.readFile( "bulk/file" + idx + ".txt" ) ).isEqualTo( "content" + idx ); + } + } } diff --git a/oap-storage/oap-storage-cloud/README.md b/oap-storage/oap-storage-cloud/README.md index 85f31daae8..8366cd5b64 100644 --- a/oap-storage/oap-storage-cloud/README.md +++ b/oap-storage/oap-storage-cloud/README.md @@ -185,6 +185,8 @@ Add the `oap-storage-cloud-ftp` artifact to your dependencies. The `ftp://` and Like `file`, FTP/FTPS have no bucket/container concept — a single connection (one server, one remote tree) serves the whole scheme. `container` is always empty; the URI's authority segment (if present) is folded into the path rather than treated as a host, so `ftp://reports/2024-06-01.json` addresses the remote path `reports/2024-06-01.json`, not a host named `reports`. The actual server to connect to always comes from config — `fs.ftp.clouds.host` is required. +FTP control connections (TCP connect + login) are pooled per backend instance using [Apache Commons Pool 2](https://commons.apache.org/proper/commons-pool/) — operations borrow a connection from the pool and return it when done instead of reconnecting/logging in on every call. Pooled connections are validated with an FTP `NOOP` before reuse, so idle connections dropped by the server/firewall are transparently replaced. + Required/optional configuration keys: | Key | Description | @@ -195,6 +197,8 @@ Required/optional configuration keys: | `fs.ftp.clouds.credential` | FTP password | | `fs.ftp.clouds.passive-mode` | `true`/`false` (default `true`) | | `fs.ftp.clouds.remove-empty-folders` | `true` to delete now-empty parent directories after a blob delete (default `false`) | +| `fs.ftp.clouds.pool-max-size` | Max pooled FTP connections per backend instance (default `8`) | +| `fs.ftp.clouds.pool-max-wait-millis` | Max time to wait for a pooled connection before failing, in milliseconds (default `30000`) | | `fs.ftps.clouds.tls-mode` | `explicit` (default) or `implicit` | | `fs.ftps.clouds.trust-all` | `true` to skip server certificate validation (e.g. self-signed certs in tests) | diff --git a/pom.xml b/pom.xml index 35c0d20d7a..de5b1cee57 100644 --- a/pom.xml +++ b/pom.xml @@ -67,7 +67,7 @@ - 25.10.1 + 25.10.2 25.0.1 25.0.0