ParallelBlockCompressedOutputStream

README.md

@@ -3,6 +3,7 @@
 [![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.github.samtools/htsjdk/badge.svg)](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22com.github.samtools%22%20AND%20a%3A%22htsjdk%22)
 [![License](http://img.shields.io/badge/license-MIT-blue.svg)](https://github.com/samtools/htsjdk)
 [![Language](http://img.shields.io/badge/language-java-brightgreen.svg)](https://www.java.com/)
+[![Join the chat at https://gitter.im/samtools/htsjdk](https://badges.gitter.im/samtools/htsjdk.svg)](https://gitter.im/samtools/htsjdk?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 
 Status of downstream projects automatically built on top of the current htsjdk master branch. See [gatk-jenkins](https://gatk-jenkins.broadinstitute.org/view/HTSJDK%20Release%20Tests/) for detailed logs. Failure may indicate problems  in htsjdk, but may also be due to expected incompatibilities between versions, or unrelated failures in downstream projects.
 - [Picard](https://github.com/broadinstitute/picard):  [![Build Status](https://gatk-jenkins.broadinstitute.org/buildStatus/icon?job=picard-on-htsjdk-master)](https://gatk-jenkins.broadinstitute.org/job/picard-on-htsjdk-master/)

src/main/java/htsjdk/samtools/AbstractSAMHeaderRecord.java

@@ -59,8 +59,6 @@ public void setAttribute(final String key, final Object value) {
     /**
      * Set the given value for the attribute named 'key'.  Replaces an existing value, if any.
      * If value is null, the attribute is removed.
-     * Supported types are Character, Integer, Float and String.  Byte and Short may also be
-     * passed in but they will be converted to Integer.
      * @param key attribute name
      * @param value attribute value
      */
@@ -71,6 +69,7 @@ public void setAttribute(final String key, final String value) {
             mAttributes.put(key, value);
         }
     }
+
     /**
      * Returns the Set of attributes.
      */

src/main/java/htsjdk/samtools/BAMFileWriter.java

@@ -23,8 +23,10 @@
  */
 package htsjdk.samtools;
 
+import htsjdk.samtools.util.AbstractBlockCompressedOutputStream;
 import htsjdk.samtools.util.BinaryCodec;
 import htsjdk.samtools.util.BlockCompressedOutputStream;
+import htsjdk.samtools.util.ParallelBlockCompressedOutputStream;
 import htsjdk.samtools.util.RuntimeIOException;
 import htsjdk.samtools.util.zip.DeflaterFactory;
 
@@ -42,35 +44,50 @@ class BAMFileWriter extends SAMFileWriterImpl {
 
     private final BinaryCodec outputBinaryCodec;
     private BAMRecordCodec bamRecordCodec = null;
-    private final BlockCompressedOutputStream blockCompressedOutputStream;
+    private final AbstractBlockCompressedOutputStream blockCompressedOutputStream;
     private BAMIndexer bamIndexer = null;
 
-    protected BAMFileWriter(final File path) {
-        blockCompressedOutputStream = new BlockCompressedOutputStream(path);
+    protected BAMFileWriter(final File path, boolean createIndex) {
+        blockCompressedOutputStream = Defaults.ZIP_THREADS > 0 && !createIndex ?
+                new ParallelBlockCompressedOutputStream(path) :
+                new BlockCompressedOutputStream(path);
+
         outputBinaryCodec = new BinaryCodec(new DataOutputStream(blockCompressedOutputStream));
         outputBinaryCodec.setOutputFileName(path.getAbsolutePath());
     }
 
-    protected BAMFileWriter(final File path, final int compressionLevel) {
-        blockCompressedOutputStream = new BlockCompressedOutputStream(path, compressionLevel);
+    protected BAMFileWriter(final File path, final int compressionLevel, boolean createIndex) {
+        blockCompressedOutputStream = Defaults.ZIP_THREADS > 0 && !createIndex ?
+                new ParallelBlockCompressedOutputStream(path, compressionLevel) :
+                new BlockCompressedOutputStream(path, compressionLevel);
+
         outputBinaryCodec = new BinaryCodec(new DataOutputStream(blockCompressedOutputStream));
         outputBinaryCodec.setOutputFileName(path.getAbsolutePath());
     }
 
-    protected BAMFileWriter(final OutputStream os, final File file) {
-        blockCompressedOutputStream = new BlockCompressedOutputStream(os, file);
+    protected BAMFileWriter(final OutputStream os, final File file, boolean createIndex) {
+        blockCompressedOutputStream = Defaults.ZIP_THREADS > 0 && !createIndex ?
+                new ParallelBlockCompressedOutputStream(os, file) :
+                new BlockCompressedOutputStream(os, file);
+
         outputBinaryCodec = new BinaryCodec(new DataOutputStream(blockCompressedOutputStream));
         outputBinaryCodec.setOutputFileName(getPathString(file));
     }
 
-    protected BAMFileWriter(final OutputStream os, final File file, final int compressionLevel) {
-        blockCompressedOutputStream = new BlockCompressedOutputStream(os, file, compressionLevel);
+    protected BAMFileWriter(final OutputStream os, final File file, final int compressionLevel, boolean createIndex) {
+        blockCompressedOutputStream = Defaults.ZIP_THREADS > 0 && !createIndex ?
+                new ParallelBlockCompressedOutputStream(os, file, compressionLevel) :
+                new BlockCompressedOutputStream(os, file, compressionLevel);
+
         outputBinaryCodec = new BinaryCodec(new DataOutputStream(blockCompressedOutputStream));
         outputBinaryCodec.setOutputFileName(getPathString(file));
     }
 
-    protected BAMFileWriter(final OutputStream os, final File file, final int compressionLevel, final DeflaterFactory deflaterFactory) {
-        blockCompressedOutputStream = new BlockCompressedOutputStream(os, file, compressionLevel, deflaterFactory);
+    protected BAMFileWriter(final OutputStream os, final File file, final int compressionLevel,
+                            final DeflaterFactory deflaterFactory, boolean createIndex) {
+        blockCompressedOutputStream = Defaults.ZIP_THREADS > 0 && !createIndex ?
+                new ParallelBlockCompressedOutputStream(os, file, compressionLevel, deflaterFactory) :
+                new BlockCompressedOutputStream(os, file, compressionLevel, deflaterFactory);
         outputBinaryCodec = new BinaryCodec(new DataOutputStream(blockCompressedOutputStream));
         outputBinaryCodec.setOutputFileName(getPathString(file));
     }

src/main/java/htsjdk/samtools/Defaults.java

@@ -15,7 +15,7 @@
  */
 public class Defaults {
     private static Log log = Log.getInstance(Defaults.class);
-    
+
     /** Should BAM index files be created when writing out coordinate sorted BAM files?  Default = false. */
     public static final boolean CREATE_INDEX;
 
@@ -84,6 +84,7 @@ public class Defaults {
      */
     public static final boolean SRA_LIBRARIES_DOWNLOAD;
 
+    public static final int ZIP_THREADS;
 
     static {
         CREATE_INDEX = getBooleanProperty("create_index", false);
@@ -104,6 +105,7 @@ public class Defaults {
         CUSTOM_READER_FACTORY = getStringProperty("custom_reader", "");
         SAM_FLAG_FIELD_FORMAT = SamFlagField.valueOf(getStringProperty("sam_flag_field_format", SamFlagField.DECIMAL.name()));
         SRA_LIBRARIES_DOWNLOAD = getBooleanProperty("sra_libraries_download", false);
+        ZIP_THREADS = getIntProperty("zip_threads", 0);
     }
 
     /**

src/main/java/htsjdk/samtools/SAMFileHeader.java

@@ -270,7 +270,7 @@ public SortOrder getSortOrder() {
 
     public void setSortOrder(final SortOrder so) {
         sortOrder = so;
-        setAttribute(SORT_ORDER_TAG, so.name());
+        super.setAttribute(SORT_ORDER_TAG, so.name());
     }
 
     public GroupOrder getGroupOrder() {
@@ -292,7 +292,42 @@ public GroupOrder getGroupOrder() {
 
     public void setGroupOrder(final GroupOrder go) {
         groupOrder = go;
-        setAttribute(GROUP_ORDER_TAG, go.name());
+        super.setAttribute(GROUP_ORDER_TAG, go.name());
+    }
+
+
+    /**
+     * Set the given value for the attribute named 'key'.  Replaces an existing value, if any.
+     * If value is null, the attribute is removed.
+     * Otherwise, the value will be converted to a String with toString.
+     * @param key attribute name
+     * @param value attribute value
+     * @deprecated Use {@link #setAttribute(String, String) instead
+     */
+    @Deprecated
+    @Override
+    public void setAttribute(final String key, final Object value) {
+        if (key.equals(SORT_ORDER_TAG) || key.equals(GROUP_ORDER_TAG)) {
+            this.setAttribute(key, value.toString());
+        } else {
+            super.setAttribute(key, value);
+        }
+    }
+
+    /**
+     * Set the given value for the attribute named 'key'.  Replaces an existing value, if any.
+     * If value is null, the attribute is removed.
+     * @param key attribute name
+     * @param value attribute value
+     */
+    @Override
+    public void setAttribute(final String key, final String value) {
+        if (key.equals(SORT_ORDER_TAG)) {
+            this.sortOrder = null;
+        } else if (key.equals(GROUP_ORDER_TAG)) {
+            this.groupOrder = null;
+        }
+        super.setAttribute(key, value);
     }
 
     /**

src/main/java/htsjdk/samtools/SAMFileWriterFactory.java

@@ -264,7 +264,7 @@ public SAMFileWriter makeBAMWriter(final SAMFileHeader header, final boolean pre
             }
             OutputStream os = IOUtil.maybeBufferOutputStream(new FileOutputStream(outputFile, false), bufferSize);
             if (createMd5File) os = new Md5CalculatingOutputStream(os, new File(outputFile.getAbsolutePath() + ".md5"));
-            final BAMFileWriter ret = new BAMFileWriter(os, outputFile, compressionLevel, deflaterFactory);
+            final BAMFileWriter ret = new BAMFileWriter(os, outputFile, compressionLevel, deflaterFactory, createIndex);
             final boolean createIndex = this.createIndex && IOUtil.isRegularPath(outputFile);
             if (this.createIndex && !createIndex) {
                 log.warn("Cannot create index for BAM because output file is not a regular file: " + outputFile.getAbsolutePath());
@@ -347,7 +347,7 @@ public SAMFileWriter makeSAMWriter(final SAMFileHeader header, final boolean pre
      */
 
     public SAMFileWriter makeBAMWriter(final SAMFileHeader header, final boolean presorted, final OutputStream stream) {
-        return initWriter(header, presorted, new BAMFileWriter(stream, null, this.getCompressionLevel(), this.deflaterFactory));
+        return initWriter(header, presorted, new BAMFileWriter(stream, null, this.getCompressionLevel(), this.deflaterFactory, false));
     }
 
     /**