title: Java-Multithreaded Compression with Deflater
body: <p>So this is an assignment that I've sort of hit a wall on. At this point, I'm mostly hoping to have more eyes since I honestly can't see anything wrong with my code anymore after redoing and refining it so many times.</p>
  
  <p>We needed to write a multithreaded compressor in Java that can be decompressed correctly with the gzip -d call. We cannot use the GZIPOutputStream call. Instead, we have generate a header and trailer manually and use a Deflater to compress the data. We read from standard input and write to standard output.</p>
  
  <p>Basically, I used and Executor to maintain a thread pool. I read the input as it comes in and write it into a buffer of a set size. Once the buffer is full, I pass that block of data to a thread (put a task in the queue). Each thread has it's own Deflater and is passed the input and any other information it needs to compress that data. I also use the last 32Kb of each block as a dictionary for the next block.</p>
  
  <p>I've confirmed that my header and trailer are correct. I used GZIPOutputStream to compress a file and used hexdump to get the bytes so I can compare it with my output. I've checked of files of different sizes and the header and trailer are identical, so in all likelyhood the issue is in the compressed data. The error that I get is: invalid compressed data--crc error </p>
  
  <p>I have confirmed that when I pass in an input that is relatively small (so that there is only one thread since I never fill up the buffer, there is only one task in the queue) the output is correct. I can call gzip -d on the compressed data and get back the exact same input.</p>
  
  <p>In other words, the problem is in when there is enough data that more than one thread is up and running. I used a hexdump on my output for the large file and compared it with GZIPOutputStream's hexdump an they're quite similar (not exactly the same, but even the case with the small file the hexdump was also slightly different for the compressed data. And in that case, gzip -d still worked). That's also how I know the header and trailer are correct.</p>
  
  <p>Incoming code dump</p>
  
  <pre><code>import java.lang.Runtime;
  import java.lang.String;
  import java.lang.Integer;
  import java.util.Arrays;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.nio.ByteBuffer;
  import java.io.*;
  import java.util.zip.*;
  
  /*Warning: Do not compress files larger than 2GB please. Since this is just
    an assignment and not meant to replace an actual parallel compressor, I cut corners
    by casting longs to ints, since it's easier to convert to 4 bytes*/
  
  public class Main {
      private static final int BLOCK_SIZE = 128*1024;
      private static final int DICT_SIZE = 32*1024;
      private static byte[] header = {(byte)0x1f, (byte)0x8b, (byte)0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
  
      public static void main(String[] args){
  
      class workerThread implements Callable&lt;byte[]&gt; {
          private boolean lastBlock;
          private boolean dictAvailable;
          private byte[] input;
          private byte[] dictionary;
          private int lastSize;       
  
          private byte[] output = new byte[BLOCK_SIZE];
          private int compressedLength;
          private ByteArrayOutputStream bOut = new ByteArrayOutputStream();
          Deflater compress = new Deflater (Deflater.DEFAULT_COMPRESSION, true);
  
          workerThread(boolean last, byte[] blockIn, byte[] dict, boolean useDictionary, int lastBSize){
              this.lastBlock = last;
              this.input = blockIn;
              this.dictionary = dict;
              this.dictAvailable = useDictionary;
              this.lastSize = lastBSize;
          }
  
          public byte[] call() {
              //System.out.println("running thread ");
              if (lastBlock) {
                  // System.out.println("Last block!");
                  compress.setInput(input,0,lastSize);
                  if(dictAvailable) {
                      compress.setDictionary(dictionary);
                  }
                  compress.finish();
                  compressedLength = compress.deflate(output,0,BLOCK_SIZE,Deflater.SYNC_FLUSH);
              }
              else {
                  //System.out.println("Not last block!");
                  compress.setInput(input,0,BLOCK_SIZE);
                  if(dictAvailable) {
                      compress.setDictionary(dictionary);
                  }
                  compressedLength = compress.deflate(output,0,BLOCK_SIZE,Deflater.SYNC_FLUSH);
              }
              byte[] finalOut = Arrays.copyOfRange(output,0,compressedLength);
              return finalOut;
          }
      }
  
      getProcessors p = new getProcessors();
      boolean useDict = true;
      int numProcs = p.getNumProcs();
      boolean customProcs = false;
      boolean foundProcs = false;
      boolean foundDict = false;
  
      /*Checking if arguments are correct*/
      ........
      /*Correct arguments, proceeding*/
  
      BufferedInputStream inBytes = new BufferedInputStream(System.in);
      byte[] buff = new byte[BLOCK_SIZE];
      byte[] dict = new byte[DICT_SIZE];
      int bytesRead = 0;
      int offset = 0;
      int uncompressedLength = 0;
      int lastBlockSize = 0;
      boolean isLastBlock = false;
      boolean firstBlockDone = false;
  
      /*Using an executor with a fixed thread pool size in order to manage threads
      as well as obtain future results to maintain synchronization*/
      ExecutorService exec = Executors.newFixedThreadPool(numProcs);
      CRC32 checksum = new CRC32();
      checksum.reset();
      List&lt;Future&lt;byte[]&gt;&gt; results = new ArrayList&lt;Future&lt;byte[]&gt;&gt;();
  
      //byte[] temp;
  
      System.out.write(header,0,header.length);
      try{
          bytesRead = inBytes.read(buff,0, BLOCK_SIZE);
          while (bytesRead != -1)  {
              uncompressedLength += bytesRead;
              checksum.update(buff,offset,bytesRead);
              offset += bytesRead;
  
              if (offset == BLOCK_SIZE) {
                  offset = 0;
                  if(!firstBlockDone){ 
                  firstBlockDone = true;
                  results.add(exec.submit(new workerThread(isLastBlock,buff,dict,false,lastBlockSize)));
                  }
                   else {
                       results.add(exec.submit(new workerThread(isLastBlock,buff,dict,useDict,lastBlockSize)));
                   }
  
                   if (useDict) {
                       System.arraycopy(buff, BLOCK_SIZE-DICT_SIZE, dict, 0, DICT_SIZE);
                   }
              }
  
               /*Implementation warning! Because of the way bytes are read in, this program will fail if
               the file being zipped is exactly a multiple of 128*1024*/
               if((bytesRead=inBytes.read(buff,offset,BLOCK_SIZE-offset)) == -1) {
                   isLastBlock = true;
                   lastBlockSize = offset;
                   results.add(exec.submit(new workerThread(isLastBlock,buff,dict,useDict,lastBlockSize)));
               }
          }    
          try {
              for(Future&lt;byte[]&gt; result: results) {
              //System.out.println("Got result!");
              System.out.write(result.get(),0,result.get().length);
              //temp = result.get();
              }
          }
          catch (InterruptedException ex) {
              ex.printStackTrace();
              System.err.println("Interrupted thread!");
          } 
          catch (ExecutionException ex) {
              ex.printStackTrace();
              System.err.println("Interrupted thread!");
          }
          finally{ 
              exec.shutdownNow();
          }
  
      /*Converting CRC sum and total length to bytes for trailer*/
      byte[] trailer = new byte[8];
      getTrailer trail = new getTrailer(checksum.getValue(),uncompressedLength);
      trail.writeTrailer(trailer,0);
      System.out.write(trailer);
  
      }
      catch (IOException ioe) {
          ioe.printStackTrace();
          System.out.println("IO error.");
          System.exit(-1);    
      }
      catch (Throwable e) {
          System.out.println("Unexpected exception or error.");
          System.exit(-1);
      }
      }
  }
  </code></pre>
  
  <p>Ah whoops, the formatting got thrown off a bit by the code block format. </p>
  
  <p>As you can see, I keep reading from input until buff is full. The reason being since this is not a file, there's a chance that the first call to read does not read enough bytes to fill the array (leaving me with a bunch of nulls that I don't want messing up anything). Once it is full, I give it to the Executor so the thread will execute the task. I implemented Callable instead of Runnable so that I could return the ouput as a byte array AND because I need the future interface. The exec.get() method allows me to the keep the threads synchornized. I've tested that with an arbitrary case (printing out numbers 1 - 100 to make sure, and they do indeed print in order).</p>
  
  <p>There is a flaw where this program won't work with files that are a multiple of BLOCK_SIZE, but that's not even the issue I'm on right now. This program works when the input is small enough that I only run one thread.</p>
  
  <p>For each block except for the last one, I call deflate with the SYNC_FLUSH option. This is so I can end on a byte boundary. The last block I compress normally and call finish on.</p>
  
  <p>Sorry for the really long post. I just honestly need more opinions besides my own because I can't seem to find the mistake. In case anyone is wonderful enough to want to compile and run it to see for themselves, here are the other classes I had (just to get the number of processes and generate the trailer. Both of these work fine).</p>
  
  <pre><code>import java.io.*;
  
  public class getTrailer {
      private long crc;
      private int total;
      public getTrailer (long crcVal, int totalIn) {
          this.crc = crcVal;
          this.total = totalIn;
      } 
      public void writeTrailer(byte[] buf, int offset) throws IOException {
          writeInt((int)crc, buf, offset); // CRC-32 of uncompr. data
          writeInt(total, buf, offset + 4); // Number of uncompr. bytes
      }
  
      /* Writes integer in Intel byte order to a byte array, starting at a
      * given offset
      */
  
      public void writeInt(int i, byte[] buf, int offset) throws IOException {
          writeShort(i &amp; 0xffff, buf, offset);
          writeShort((i &gt;&gt; 16) &amp; 0xffff, buf, offset + 2);
      }
  
     /*
      * Writes short integer in Intel byte order to a byte array, starting
      * at a given offset
      */
  
      public void writeShort(int s, byte[] buf, int offset) throws IOException {
          buf[offset] = (byte)(s &amp; 0xff);
          buf[offset + 1] = (byte)((s &gt;&gt; 8) &amp; 0xff);
      }
  }
  </code></pre>
  
  <p>The trailer function is literally copy pasted from JAva's docs</p>
  
  <pre><code>public class getProcessors {
      private Runtime runner = Runtime.getRuntime();
      private int nProcs = runner.availableProcessors();
  
      int getNumProcs() {
          return nProcs;
      }
  }
  </code></pre>
  
  <p>I realize how long this is, but I really need someone else's opinion. If anyone sees anything they think might be causing issues, please tell me. I don't need someone to write the program for me (I think I'm almost there) but I just...can't see anything wrong.</p>