iRODS: TEST THIS. IF WORKING, SPLIT IT INTO CLEAN COMMITS

korydraughn · korydraughn · commit 660266c1e349 · 2025-08-14T17:50:53.000-04:00
diff --git a/defaults/src/main/resources/default.properties b/defaults/src/main/resources/default.properties
@@ -776,3 +776,16 @@ deepbox.listing.box.trash=false
 # 1 min
 deepbox.download.interrupt.ms=60000
 deepbox.download.interval.ms=0=50 2=200 5=500 15=2000
+
+# iRODS configuration
+# -------------------
+irods.client_negotiation_policy=CS_NEG_REFUSE
+# Local files exceeding the following size will trigger use of parallel transfer.
+# Normally defaults to 32MB in iRODS clients.
+irods.parallel_transfer.size_threshold=33554432
+# The number of threads used for parallel transfer.
+irods.parallel_transfer.thread_count=3
+# The size of the buffer used for reading during parallel transfer.
+irods.parallel_transfer.rbuffer_size=4194304
+# The size of the buffer used for writing during parallel transfer.
+irods.parallel_transfer.wbuffer_size=4194304
diff --git a/irods/pom.xml b/irods/pom.xml
@@ -27,7 +27,7 @@
         <dependency>
             <groupId>org.irods</groupId>
             <artifactId>irods4j</artifactId>
-            <version>0.3.0-java8-SNAPSHOT</version>
+            <version>0.3.0-java8</version>
         </dependency>
         <dependency>
             <groupId>ch.cyberduck</groupId>
diff --git a/irods/src/main/java/ch/cyberduck/core/irods/IRODSChunkWorker.java b/irods/src/main/java/ch/cyberduck/core/irods/IRODSChunkWorker.java
@@ -15,7 +15,6 @@
  * GNU General Public License for more details.
  */
 
-import org.irods.irods4j.high_level.connection.IRODSConnectionPool;
 import org.irods.irods4j.high_level.io.IRODSDataObjectInputStream;
 import org.irods.irods4j.high_level.io.IRODSDataObjectOutputStream;
 import org.irods.irods4j.high_level.io.IRODSDataObjectStream;
@@ -29,15 +28,13 @@
 
 public class IRODSChunkWorker implements Runnable {
 
-    private final IRODSConnectionPool.PoolConnection conn;
     private final InputStream in;
     private final OutputStream out;
     private final long offset;
     private final long chunkSize;
     private final byte[] buffer;
 
-    public IRODSChunkWorker(IRODSConnectionPool.PoolConnection conn, InputStream in, OutputStream out, long offset, long chunkSize, int bufferSize) throws IOException, IRODSException {
-        this.conn = conn;
+    public IRODSChunkWorker(InputStream in, OutputStream out, long offset, long chunkSize, int bufferSize) {
         this.in = in;
         this.out = out;
         this.offset = offset;
@@ -67,16 +64,6 @@ public void run() {
         catch(IOException | IRODSException e) {
             // TODO Log error
         }
-        finally {
-            try {
-                in.close();
-            }
-            catch(Exception e) { /* Ignored */ }
-            try {
-                out.close();
-            }
-            catch(Exception e) { /* Ignored */ }
-        }
     }
 
     private void seek(InputStream in) throws IRODSException, IOException {
diff --git a/irods/src/main/java/ch/cyberduck/core/irods/IRODSConnectionUtils.java b/irods/src/main/java/ch/cyberduck/core/irods/IRODSConnectionUtils.java
@@ -0,0 +1,67 @@
+package ch.cyberduck.core.irods;
+
+/*
+ * Copyright (c) 2002-2025 iterate GmbH. All rights reserved.
+ * https://cyberduck.io/
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ */
+
+import org.apache.commons.lang3.StringUtils;
+import org.irods.irods4j.authentication.AuthPlugin;
+import org.irods.irods4j.authentication.NativeAuthPlugin;
+import org.irods.irods4j.authentication.PamInteractiveAuthPlugin;
+import org.irods.irods4j.authentication.PamPasswordAuthPlugin;
+import org.irods.irods4j.high_level.connection.IRODSConnectionPool;
+import org.irods.irods4j.high_level.connection.QualifiedUsername;
+import org.irods.irods4j.low_level.api.IRODSApi;
+import org.irods.irods4j.low_level.api.IRODSException;
+
+import java.io.IOException;
+
+final class IRODSConnectionUtils {
+
+    public static void startIRODSConnectionPool(IRODSSession session, IRODSConnectionPool connPool) throws IRODSException, IOException {
+        String host = session.getHost().getHostname();
+        int port = session.getHost().getPort();
+        String zone = session.getRegion();
+        String username = session.getHost().getCredentials().getUsername();
+        String password = session.getHost().getCredentials().getPassword();
+
+        connPool.start(
+                host,
+                port,
+                new QualifiedUsername(username, zone),
+                conn -> {
+                    try {
+                        final String authScheme = StringUtils.defaultIfBlank(session.getHost().getProtocol().getAuthorization(), "native");
+                        AuthPlugin plugin = null;
+                        if("native".equals(authScheme)) {
+                            plugin = new NativeAuthPlugin();
+                        }
+                        else if("pam_password".equals(authScheme)) {
+                            plugin = new PamPasswordAuthPlugin(true);
+                        }
+                        else if("pam_interactive".equals(authScheme)) {
+                            plugin = new PamInteractiveAuthPlugin(true);
+                        }
+                        else {
+                            throw new IllegalArgumentException(String.format("Authentication scheme not recognized: %s", authScheme));
+                        }
+                        IRODSApi.rcAuthenticateClient(conn, plugin, password);
+                        return true;
+                    }
+                    catch(Exception e) {
+                        return false;
+                    }
+                });
+    }
+}
diff --git a/irods/src/main/java/ch/cyberduck/core/irods/IRODSDownloadFeature.java b/irods/src/main/java/ch/cyberduck/core/irods/IRODSDownloadFeature.java
@@ -24,21 +24,21 @@
 import ch.cyberduck.core.features.Read;
 import ch.cyberduck.core.io.BandwidthThrottle;
 import ch.cyberduck.core.io.StreamListener;
+import ch.cyberduck.core.preferences.HostPreferencesFactory;
+import ch.cyberduck.core.preferences.PreferencesReader;
 import ch.cyberduck.core.transfer.TransferStatus;
 
 import org.irods.irods4j.high_level.connection.IRODSConnectionPool;
 import org.irods.irods4j.high_level.connection.IRODSConnectionPool.PoolConnection;
-import org.irods.irods4j.high_level.connection.QualifiedUsername;
 import org.irods.irods4j.high_level.io.IRODSDataObjectInputStream;
 import org.irods.irods4j.high_level.vfs.IRODSFilesystem;
-import org.irods.irods4j.low_level.api.IRODSApi;
 import org.irods.irods4j.low_level.api.IRODSApi.RcComm;
 
-import java.io.BufferedOutputStream;
 import java.io.FileOutputStream;
-import java.io.RandomAccessFile;
+import java.io.OutputStream;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -49,8 +49,6 @@
 public class IRODSDownloadFeature implements Download {
 
     private final IRODSSession session;
-    private final int numThread = 3; // TODO Make this configurable
-    private static final int BUFFER_SIZE = 4 * 1024 * 1024; // TODO Make configurable
 
     public IRODSDownloadFeature(final IRODSSession session) {
         this.session = session;
@@ -61,6 +59,8 @@ public void download(final Path file, final Local local, final BandwidthThrottle
                          final StreamListener listener, final TransferStatus status,
                          final ConnectionCallback callback) throws BackgroundException {
         try {
+            final PreferencesReader preferences = HostPreferencesFactory.get(session.getHost());
+
             final RcComm primaryConn = session.getClient().getRcComm();
             final String logicalPath = file.getAbsolute();
 
@@ -72,9 +72,8 @@ public void download(final Path file, final Local local, final BandwidthThrottle
 
             // Transfer the bytes over multiple connections if the size of the data object
             // exceeds a certain threshold - e.g. 32MB.
-            // TODO Consider making this configurable.
-            if(dataObjectSize < 32 * 1024 * 1024) {
-                byte[] buffer = new byte[4 * 1024 * 1024];
+            if(dataObjectSize < 32 * 1024 * 1024) { //preferences.getInteger("irods.parallel_transfer.size_threshold")) {
+                byte[] buffer = new byte[4 * 1024 * 1024]; //preferences.getInteger("irods.parallel_transfer.rbuffer_size")];
 
                 try(IRODSDataObjectInputStream in = new IRODSDataObjectInputStream(primaryConn, logicalPath);
                     FileOutputStream out = new FileOutputStream(local.getAbsolute())) {
@@ -92,58 +91,77 @@ public void download(final Path file, final Local local, final BandwidthThrottle
             // The data object is larger than the threshold so use parallel transfer.
             //
 
-            // Step 1: Get replica token & number via primary stream
-            try(IRODSDataObjectInputStream primary = new IRODSDataObjectInputStream(primaryConn, logicalPath)) {
-                final String replicaToken = primary.getReplicaToken();
-                final long replicaNumber = primary.getReplicaNumber();
-
-                // Step 2: Setup connection pool
-                try(final IRODSConnectionPool pool = new IRODSConnectionPool(numThread)) {
-                    pool.start(
-                            session.getHost().getHostname(),
-                            session.getHost().getPort(),
-                            new QualifiedUsername(session.getHost().getCredentials().getUsername(), session.getRegion()),
-                            conn -> {
-                                try {
-                                    // TODO Needs to take the value of the Authorization profile value.
-                                    IRODSApi.rcAuthenticateClient(conn, "native", session.getHost().getCredentials().getPassword());
-                                    return true;
-                                }
-                                catch(Exception e) {
-                                    return false;
-                                }
-                            });
-
-                    final long chunkSize = dataObjectSize / numThread;
-                    final long remainChunkSize = dataObjectSize % numThread;
-
-                    // Step 3: Create empty target file
-                    try(RandomAccessFile out = new RandomAccessFile(local.getAbsolute(), "rw")) {
-                        out.setLength(dataObjectSize);
-                    }
+            // TODO Clamp the value so that users do not specify something ridiculous.
+            final int threadCount = 3; //preferences.getInteger("irods.parallel_transfer.thread_count");
+            final ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+
+            final long chunkSize = dataObjectSize / threadCount;
+            final long remainingBytes = dataObjectSize % threadCount;
+
+            final List<IRODSDataObjectInputStream> secondaryIrodsStreams = new ArrayList<>();
+            final List<OutputStream> localFileStreams = new ArrayList<>();
+
+            // Open the primary iRODS input stream.
+            // TODO Needs to pass the target resource name if provided.
+            try(IRODSDataObjectInputStream primaryStream = new IRODSDataObjectInputStream(primaryConn, logicalPath)) {
+                // Initialize connections for secondary streams.
+                try(IRODSConnectionPool pool = new IRODSConnectionPool(threadCount - 1)) {
+                    IRODSConnectionUtils.startIRODSConnectionPool(session, pool);
 
-                    // Step 4: Parallel readers
-                    final ExecutorService executor = Executors.newFixedThreadPool(numThread);
+                    // Holds handles to tasks running on the thread pool. This allows us to wait for
+                    // all tasks to complete before shutting down everything.
                     List<Future<?>> tasks = new ArrayList<>();
-                    for(int i = 0; i < numThread; i++) {
-                        final PoolConnection conn = pool.getConnection();
-                        tasks.add(executor.submit(new IRODSChunkWorker(
-                                conn,
-                                new IRODSDataObjectInputStream(conn.getRcComm(), replicaToken, replicaNumber),
-                                new FileOutputStream(local.getAbsolute()),
-                                i * chunkSize,
-                                (numThread - 1 == i) ? chunkSize + remainChunkSize : chunkSize,
-                                BUFFER_SIZE
-                        )));
-                    }
 
-                    for(Future<?> task : tasks) {
-                        task.get();
-                    }
+                    // Open the first output stream for the local file and store it for processing.
+                    // This also guarantees the target file exists and is empty (i.e. truncated to zero
+                    // if it exists).
+                    final java.nio.file.Path localFilePath = Paths.get(local.getAbsolute());
+                    localFileStreams.add(Files.newOutputStream(localFilePath, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING));
+
+                    // Launch the first IO task.
+                    tasks.add(executor.submit(new IRODSChunkWorker(
+                            primaryStream,
+                            localFileStreams.get(0),
+                            0,
+                            chunkSize,
+                            4 * 1024 * 1024//preferences.getInteger("irods.parallel_transfer.rbuffer_size")
+                    )));
+
+                    try {
+                        // Launch remaining IO tasks.
+                        for(int i = 1; i < threadCount; ++i) {
+                            PoolConnection conn = pool.getConnection();
+                            secondaryIrodsStreams.add(new IRODSDataObjectInputStream(conn.getRcComm(), logicalPath));
+                            localFileStreams.add(Files.newOutputStream(localFilePath, StandardOpenOption.WRITE));
+                            tasks.add(executor.submit(new IRODSChunkWorker(
+                                    secondaryIrodsStreams.get(secondaryIrodsStreams.size() - 1),
+                                    localFileStreams.get(localFileStreams.size() - 1),
+                                    i * chunkSize,
+                                    (threadCount - 1 == i) ? chunkSize + remainingBytes : chunkSize,
+                                    4 * 1024 * 1024//preferences.getInteger("irods.parallel_transfer.rbuffer_size")
+                            )));
+                        }
 
-                    executor.shutdown();
+                        // Wait for all tasks on the thread pool to complete.
+                        for(Future<?> task : tasks) {
+                            try {
+                                task.get();
+                            }
+                            catch(Exception e) { /* Ignored */ }
+                        }
+                    }
+                    finally {
+                        closeInputStreams(secondaryIrodsStreams);
+                    }
                 }
             }
+            finally {
+                closeOutputStreams(localFileStreams);
+            }
+
+            executor.shutdown();
+            // TODO Make this configurable.
+            executor.awaitTermination(5, TimeUnit.SECONDS);
         }
         catch(Exception e) {
             throw new IRODSExceptionMappingService().map("Download {0} failed", e);
@@ -159,4 +177,22 @@ public boolean offset(final Path file) {
     public Download withReader(final Read reader) {
         return this;
     }
+
+    private static void closeInputStreams(List<IRODSDataObjectInputStream> streams) {
+        streams.forEach(in -> {
+            try {
+                in.close();
+            }
+            catch(Exception e) { /* Ignored */ }
+        });
+    }
+
+    private static void closeOutputStreams(List<OutputStream> streams) {
+        streams.forEach(out -> {
+            try {
+                out.close();
+            }
+            catch(Exception e) { /* Ignored */ }
+        });
+    }
 }
diff --git a/irods/src/main/java/ch/cyberduck/core/irods/IRODSExceptionMappingService.java b/irods/src/main/java/ch/cyberduck/core/irods/IRODSExceptionMappingService.java
@@ -22,6 +22,7 @@
 import org.apache.logging.log4j.Logger;
 
 public class IRODSExceptionMappingService extends AbstractExceptionMappingService<Exception> {
+
     private static final Logger log = LogManager.getLogger(IRODSExceptionMappingService.class);
 
     @Override
@@ -32,5 +33,4 @@ public BackgroundException map(final Exception e) {
         this.append(buffer, e.getMessage());
         return this.wrap(e, buffer);
     }
-
 }
diff --git a/irods/src/main/java/ch/cyberduck/core/irods/IRODSIntegerUtils.java b/irods/src/main/java/ch/cyberduck/core/irods/IRODSIntegerUtils.java
@@ -0,0 +1,32 @@
+package ch.cyberduck.core.irods;
+
+/*
+ * Copyright (c) 2002-2025 iterate GmbH. All rights reserved.
+ * https://cyberduck.io/
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ */
+
+public final class IRODSIntegerUtils {
+
+    static <T extends Comparable<T>> T clamp(T value, T low, T high) {
+        if(value.compareTo(low) < 0) {
+            return low;
+        }
+
+        if(value.compareTo(high) > 0) {
+            return high;
+        }
+
+        return value;
+    }
+
+}
diff --git a/irods/src/main/java/ch/cyberduck/core/irods/IRODSSession.java b/irods/src/main/java/ch/cyberduck/core/irods/IRODSSession.java
diff --git a/irods/src/main/java/ch/cyberduck/core/irods/IRODSUploadFeature.java b/irods/src/main/java/ch/cyberduck/core/irods/IRODSUploadFeature.java

Original file line number	Diff line number	Diff line change
`@@ -22,6 +22,7 @@`
`22`	`22`	`import org.apache.logging.log4j.Logger;`
`23`	`23`
`24`	`24`	`public class IRODSExceptionMappingService extends AbstractExceptionMappingService<Exception> {`
	`25`	`+`
`25`	`26`	`private static final Logger log = LogManager.getLogger(IRODSExceptionMappingService.class);`
`26`	`27`
`27`	`28`	`@Override`
`@@ -32,5 +33,4 @@ public BackgroundException map(final Exception e) {`
`32`	`33`	`this.append(buffer, e.getMessage());`
`33`	`34`	`return this.wrap(e, buffer);`
`34`	`35`	`}`
`35`		`-`
`36`	`36`	`}`