Xxhash integration

azure/pom.xml

@@ -6,12 +6,12 @@
     <parent>
         <groupId>com.instaclustr</groupId>
         <artifactId>esop-parent</artifactId>
-        <version>4.0.1</version>
+        <version>4.0.2-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
     <artifactId>esop-azure</artifactId>
-    <version>4.0.1</version>
+    <version>4.0.2-SNAPSHOT</version>
 
     <name>esop-azure</name>
     <description>Backup and restoration tooling for Cassandra for Azure</description>

core/pom.xml

@@ -6,12 +6,12 @@
     <parent>
         <groupId>com.instaclustr</groupId>
         <artifactId>esop-parent</artifactId>
-        <version>4.0.1</version>
+        <version>4.0.2-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
     <artifactId>esop-core</artifactId>
-    <version>4.0.1</version>
+    <version>4.0.2-SNAPSHOT</version>
 
     <name>esop-core</name>
     <description>Core of backup and restoration tooling for Cassandra</description>
@@ -25,6 +25,7 @@
         <commons-io.version>2.20.0</commons-io.version>
         <awaitility.version>3.1.6</awaitility.version>
         <jackson.bom.version>2.19.2</jackson.bom.version>
+        <lz4.version>1.8.0</lz4.version>
 
         <slf4j.version>2.0.17</slf4j.version>
         <logback.version>1.5.19</logback.version>
@@ -120,6 +121,12 @@
             <version>${awaitility.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.lz4</groupId>
+            <artifactId>lz4-java</artifactId>
+            <version>${lz4.version}</version>
+        </dependency>
+
         <!-- logging -->
 
         <dependency>

core/src/main/java/com/instaclustr/esop/cli/BackupApplication.java

@@ -59,7 +59,7 @@ public void run() {
         List<Module> additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
         additionalModules.add(new BackupModule());
 
-        Esop.init(this, jmxSpec, hashSpec, additionalModules);
+        Esop.init(this, jmxSpec, hashSpec, request, additionalModules);
 
         final Operation<?> operation = operationsService.submitOperationRequest(request);
 

core/src/main/java/com/instaclustr/esop/cli/CommitLogBackupApplication.java

@@ -53,7 +53,7 @@ public void run() {
         List<Module> additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
         additionalModules.add(new CommitlogBackupModule());
 
-        Esop.init(this, jmxSpec, hashSpec, additionalModules);
+        Esop.init(this, jmxSpec, hashSpec, request, additionalModules);
 
         final Operation<?> operation = operationsService.submitOperationRequest(request);
 

core/src/main/java/com/instaclustr/esop/cli/CommitLogRestoreApplication.java

@@ -49,7 +49,7 @@ public void run() {
         List<Module> additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
         additionalModules.add(new RestoreCommitlogModule());
 
-        Esop.init(this, null, hashSpec, additionalModules);
+        Esop.init(this, null, hashSpec, request, additionalModules);
 
         final Operation<?> operation = operationsService.submitOperationRequest(request);
 

core/src/main/java/com/instaclustr/esop/cli/Esop.java

@@ -12,6 +12,7 @@
 import com.google.inject.Stage;
 import com.instaclustr.cassandra.CassandraModule;
 import com.instaclustr.esop.SPIModule;
+import com.instaclustr.esop.impl.AbstractOperationRequest;
 import com.instaclustr.esop.impl.backup.BackupModules.UploadingModule;
 import com.instaclustr.esop.impl.hash.HashModule;
 import com.instaclustr.esop.impl.hash.HashSpec;
@@ -70,6 +71,7 @@ public static void main(String[] args, boolean exit) {
     static void init(final Runnable command,
                      final CassandraJMXSpec jmxSpec,
                      final HashSpec hashSpec,
+                     final AbstractOperationRequest request,
                      final List<Module> additionalModules) {
 
         final List<Module> modules = new ArrayList<>();
@@ -92,7 +94,7 @@ protected void configure() {
         modules.add(new ExecutorsModule());
         modules.add(new UploadingModule());
         modules.add(new DownloadingModule());
-        modules.add(new HashModule(hashSpec));
+        modules.add(new HashModule(hashSpec, request.concurrentConnections));
         modules.addAll(additionalModules);
 
         final Injector injector = Guice.createInjector(

core/src/main/java/com/instaclustr/esop/cli/ListApplication.java

@@ -54,7 +54,7 @@ public void run() {
         List<Module> additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
         additionalModules.add(new ListModule());
 
-        Esop.init(this, jmxSpec, new HashSpec(), additionalModules);
+        Esop.init(this, jmxSpec, new HashSpec(), request, additionalModules);
 
         final Operation<?> operation = operationsService.submitOperationRequest(request);
 

core/src/main/java/com/instaclustr/esop/cli/RemoveBackupApplication.java

@@ -78,7 +78,7 @@ public void run() {
         List<Module> additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
         additionalModules.add(new RemoveBackupModule());
 
-        Esop.init(this, jmxSpec, new HashSpec(), additionalModules);
+        Esop.init(this, jmxSpec, new HashSpec(), request, additionalModules);
 
         if (rate.value == 0) {
             final Operation<?> operation = operationsService.submitOperationRequest(request);

core/src/main/java/com/instaclustr/esop/cli/RestoreApplication.java

@@ -61,7 +61,7 @@ public void run() {
         additionalModules.add(new RestoreModule());
         additionalModules.add(new RestorationStrategyModule());
 
-        Esop.init(this, jmxSpec, hashSpec, additionalModules);
+        Esop.init(this, jmxSpec, hashSpec, request, additionalModules);
 
         final Operation<?> operation = operationsService.submitOperationRequest(request);
 

core/src/main/java/com/instaclustr/esop/impl/AbstractOperationRequest.java

@@ -22,7 +22,7 @@ public abstract class AbstractOperationRequest extends OperationRequest {
         converter = StorageLocationTypeConverter.class,
         description = "Location to which files will be backed up or restored from, in form " +
             "cloudProvider://bucketName/clusterId/datacenterId/nodeId or file:///some/path/bucketName/clusterId/datacenterId/nodeId. " +
-            "'cloudProvider' is one of 's3', 'oracle', 'ceph', 'minio', 'azure' or 'gcp'.",
+            "'cloudProvider' is one of 's3', 'azure' or 'gcp'.",
         required = true)
     @JsonSerialize(using = StorageLocationSerializer.class)
     @JsonDeserialize(using = StorageLocationDeserializer.class)
@@ -47,9 +47,8 @@ public abstract class AbstractOperationRequest extends OperationRequest {
     @JsonProperty("retry")
     public RetrySpec retry = new RetrySpec();
 
-    @Option(names = {"--cc", "--concurrent-connections"},
-            description = "Number of files (or file parts) to download concurrently. Higher values will increase throughput. Default is 10.",
-            defaultValue = "10"
+    @Option(names = {"--cc", "--concurrent-connections", "--parallelism"},
+            description = "Number of files (or file parts) to download / upload / hash concurrently. Higher values will increase throughput. Default is 50% of available CPUs."
     )
     @JsonProperty("concurrentConnections")
     public Integer concurrentConnections;
@@ -60,6 +59,8 @@ public abstract class AbstractOperationRequest extends OperationRequest {
 
     public AbstractOperationRequest() {
         // for picocli
+        if (concurrentConnections == null)
+            concurrentConnections = getDefaultConcurrentConnections();
     }
 
     public AbstractOperationRequest(final StorageLocation storageLocation,
@@ -74,7 +75,7 @@ public AbstractOperationRequest(final StorageLocation storageLocation,
         this.skipBucketVerification = skipBucketVerification;
         this.proxySettings = proxySettings;
         this.retry = retry == null ? new RetrySpec() : retry;
-        this.concurrentConnections = concurrentConnections == null ? 10 : concurrentConnections;
+        this.concurrentConnections = concurrentConnections == null ? getDefaultConcurrentConnections() : concurrentConnections;
         this.kmsKeyId = kmsKeyId;
     }
 
@@ -96,5 +97,25 @@ public void validate(final Set<String> storageProviders) {
         if (storageProviders != null && !storageProviders.contains(storageLocation.storageProvider)) {
             throw new IllegalStateException(format("Available storage providers: %s", Arrays.toString(storageProviders.toArray())));
         }
+
+        validateConcurrentConnections();
+    }
+
+    /**
+     * Get default number of concurrent connections based on 50% of available processors.
+     */
+    private static int getDefaultConcurrentConnections() {
+        return Runtime.getRuntime().availableProcessors() / 2;
+    }
+
+    private void validateConcurrentConnections() {
+        if (concurrentConnections <= 0) {
+            throw new IllegalStateException("--parallelism must be greater than 0");
+        }
+
+        if (concurrentConnections > Runtime.getRuntime().availableProcessors()) {
+            throw new IllegalStateException("--parallelism value cannot be greater than number of available processors: "
+                + Runtime.getRuntime().availableProcessors());
+        }
     }
 }

core/src/main/java/com/instaclustr/esop/impl/SSTableUtils.java

@@ -41,7 +41,6 @@ public class SSTableUtils {
     private static final int SSTABLE_PREFIX_IDX = 1;
     private static final int SSTABLE_GENERATION_IDX = 2;
     private static final Pattern CHECKSUM_RE = Pattern.compile("^([a-zA-Z0-9]+).*");
-    private static final HashService hashService = new HashServiceImpl(new HashSpec());
 
     public static String sstableHash(Path path) throws IOException {
         final Matcher matcher = SSTABLE_RE.matcher(path.getFileName().toString());
@@ -102,14 +101,11 @@ public static String calculateChecksum(final Path filePath) throws IOException {
     public static Map<String, List<ManifestEntry>> getSSTables(String keyspace,
                                                                String table,
                                                                Path snapshotDirectory,
-                                                               Path tableBackupPath,
-                                                               HashSpec hashSpec) throws IOException {
+                                                               Path tableBackupPath) throws IOException {
         if (!Files.exists(snapshotDirectory)) {
             return Collections.emptyMap();
         }
 
-        final HashService hashService = new HashServiceImpl(hashSpec);
-
         return Files.list(snapshotDirectory)
                     .flatMap(path -> {
                         if (isCassandra22SecIndex(path)) {
@@ -148,12 +144,11 @@ public static Map<String, List<ManifestEntry>> getSSTables(String keyspace,
                                 }
 
                                 backupPath = backupPath.resolve(hash).resolve(manifestComponentFileName.getFileName());
-                                final String hashOfFile = hashService.hash(sstableComponent);
 
                                 entries.add(new ManifestEntry(backupPath,
                                                               sstableComponent,
                                                               ManifestEntry.Type.FILE,
-                                                              hashOfFile,
+                                                              null,
                                                               new KeyspaceTable(keyspace, table),
                                                               null));
                             }

core/src/main/java/com/instaclustr/esop/impl/Snapshots.java

@@ -46,8 +46,6 @@
 
 public class Snapshots implements Cloneable {
 
-    public static HashSpec hashSpec;
-
     private final Map<String, Snapshot> snapshots = new HashMap<>();
 
     public static Snapshots of(Map<String, Snapshot> snapshots) {
@@ -569,7 +567,7 @@ public static Table parse(final String keyspace, final String table, final List<
                     final Path tablePath = Paths.get("data").resolve(Paths.get(keyspace, table));
 
                     for (final Path path : value) {
-                        tb.sstables.putAll(SSTableUtils.getSSTables(keyspace, table, path, tablePath, Snapshots.hashSpec));
+                        tb.sstables.putAll(SSTableUtils.getSSTables(keyspace, table, path, tablePath));
                     }
 
                     final Optional<Path> schemaPath = value.stream().map(p -> p.resolve("schema.cql")).filter(Files::exists).findFirst();
@@ -750,10 +748,6 @@ public static synchronized Snapshots merge(final List<Snapshots> scannedSnapshot
     }
 
     public static synchronized Snapshots parse(final Path cassandraDir, final String snapshot) throws Exception {
-        if (Snapshots.hashSpec == null) {
-            Snapshots.hashSpec = new HashSpec();
-        }
-
         final Snapshots snapshots = new Snapshots();
         final SnapshotLister lister = new SnapshotLister();
         Files.walkFileTree(cassandraDir, lister);

core/src/main/java/com/instaclustr/esop/impl/backup/coordination/BaseBackupOperationCoordinator.java

@@ -28,6 +28,8 @@
 import com.instaclustr.esop.impl.backup.coordination.ClearSnapshotOperation.ClearSnapshotOperationRequest;
 import com.instaclustr.esop.impl.backup.coordination.TakeSnapshotOperation.TakeSnapshotOperationRequest;
 import com.instaclustr.esop.impl.hash.HashSpec;
+import com.instaclustr.esop.impl.hash.ParallelHashService;
+import com.instaclustr.esop.impl.hash.ParallelHashServiceImpl;
 import com.instaclustr.esop.impl.interaction.CassandraSchemaVersion;
 import com.instaclustr.esop.impl.interaction.CassandraTokens;
 import com.instaclustr.esop.topology.CassandraClusterTopology;
@@ -126,8 +128,8 @@ public void coordinate(final Operation<BackupOperationRequest> operation) {
                                       new TakeSnapshotOperationRequest(request.entities, request.snapshotTag),
                                       cassandraVersionProvider).run0();
 
-            Snapshots.hashSpec = hashSpec;
             final Snapshots snapshots = Snapshots.parse(request.dataDirs, request.snapshotTag);
+
             final Optional<Snapshot> snapshot = snapshots.get(request.snapshotTag);
 
             if (!snapshot.isPresent()) {
@@ -139,6 +141,11 @@ public void coordinate(final Operation<BackupOperationRequest> operation) {
             manifest.setSchemaVersion(request.schemaVersion);
             manifest.setTokens(tokens);
 
+            // Compute hashes and populate it to manifest entries
+            try (ParallelHashService parallelHashService = new ParallelHashServiceImpl(hashSpec, request.concurrentConnections)) {
+                parallelHashService.hashAndPopulate(manifest.getManifestEntries(false));
+            }
+
             // manifest
             final Path localManifestPath = getLocalManifestPath(request.snapshotTag);
             manifest.setManifest(getManifestAsManifestEntry(localManifestPath, request));

core/src/main/java/com/instaclustr/esop/impl/hash/HashModule.java

@@ -5,14 +5,17 @@
 public class HashModule extends AbstractModule {
 
     private final HashSpec hashSpec;
+    private final int parallelHashingThreads;
 
-    public HashModule(final HashSpec hashSpec) {
+    public HashModule(final HashSpec hashSpec, final int parallelHashingThreads) {
         this.hashSpec = hashSpec;
+        this.parallelHashingThreads = parallelHashingThreads;
     }
 
     @Override
     protected void configure() {
         bind(HashSpec.class).toInstance(this.hashSpec);
         bind(HashService.class).toInstance(new HashServiceImpl(this.hashSpec));
+        bind(ParallelHashService.class).toProvider(() -> new ParallelHashServiceImpl(this.hashSpec, this.parallelHashingThreads));
     }
 }

core/src/main/java/com/instaclustr/esop/impl/hash/HashService.java

@@ -14,7 +14,7 @@ public interface HashService {
 
     void verify(Path file, String hash) throws HashVerificationException;
 
-    class HashingException extends Exception {
+    class HashingException extends RuntimeException {
 
         public HashingException(final String message) {
             super(message);
@@ -25,7 +25,7 @@ public HashingException(final String message, final Throwable cause) {
         }
     }
 
-    class HashVerificationException extends Exception {
+    class HashVerificationException extends RuntimeException {
 
         public HashVerificationException(final String message) {
             super(message);