Optimizing a thread safe Java NIO / Serialization / FIFO Queue

Question

I've written a thread safe, persistent FIFO for Serializable items. The reason for reinventing the wheel is that we simply can't afford any third party dependencies in this project and want to keep this really simple.

The problem is it isn't fast enough. Most of it is undoubtedly due to reading and writing directly to disk but I think we should be able to squeeze a bit more out of it anyway. Any ideas on how to improve the performance of the 'take'- and 'add'-methods?

/**
 * <code>DiskQueue</code> Persistent, thread safe FIFO queue for
 * <code>Serializable</code> items.
 */
public class DiskQueue<ItemT extends Serializable>
{
   public static final int EMPTY_OFFS = -1;
   public static final int LONG_SIZE = 8;
   public static final int HEADER_SIZE = LONG_SIZE * 2;

   private InputStream inputStream;
   private OutputStream outputStream;
   private RandomAccessFile file;
   private FileChannel channel;
   private long offs = EMPTY_OFFS;
   private long size = 0;

   public DiskQueue(String filename)
   {
      try
      {
         boolean fileExists = new File(filename).exists();
         file = new RandomAccessFile(filename, "rwd");
         if (fileExists)
         {
            size = file.readLong();
            offs = file.readLong();
         }
         else
         {
            file.writeLong(size);
            file.writeLong(offs);
         }
      } catch (FileNotFoundException e)
      {
         throw new RuntimeException(e);
      } catch (IOException e)
      {
         throw new RuntimeException(e);
      }

      channel = file.getChannel();
      inputStream = Channels.newInputStream(channel);
      outputStream = Channels.newOutputStream(channel);
   }

   /**
    * Add item to end of queue.
    */
   public void add(ItemT item)
   {
      try
      {
         synchronized (this)
         {
            channel.position(channel.size());

            ObjectOutputStream s = new ObjectOutputStream(outputStream);
            s.writeObject(item);
            s.flush();

            size++;
            file.seek(0);
            file.writeLong(size);
            if (offs == EMPTY_OFFS)
            {
               offs = HEADER_SIZE;
               file.writeLong(offs);
            }
            notify();
         }
      } catch (IOException e)
      {
         throw new RuntimeException(e);
      }
   }

   /**
    * Clears overhead by moving the remaining items up and shortening the file.
    */
   public synchronized void defrag()
   {

      if (offs > HEADER_SIZE && size > 0)
      {
         try
         {
            long totalBytes = channel.size() - offs;
            ByteBuffer buffer = ByteBuffer.allocateDirect((int) totalBytes);
            channel.position(offs);

            for (int bytes = 0; bytes < totalBytes;)
            {
               int res = channel.read(buffer);
               if (res == -1)
               {
                  throw new IOException("Failed to read data into buffer");
               }
               bytes += res;
            }

            channel.position(HEADER_SIZE);
            buffer.flip();
            for (int bytes = 0; bytes < totalBytes;)
            {
               int res = channel.write(buffer);
               if (res == -1)
               {
                  throw new IOException("Failed to write buffer to file");
               }
               bytes += res;
            }

            offs = HEADER_SIZE;
            file.seek(LONG_SIZE);
            file.writeLong(offs);
            file.setLength(HEADER_SIZE + totalBytes);
         } catch (IOException e)
         {
            throw new RuntimeException(e);
         }
      }
   }

   /**
    * Returns the queue overhead in bytes.
    */
   public synchronized long overhead()
   {
      return (offs == EMPTY_OFFS) ? 0 : offs - HEADER_SIZE;
   }

   /**
    * Returns the first item in the queue, blocks if queue is empty.
    */
   public ItemT peek() throws InterruptedException
   {
      block();

      synchronized (this)
      {
         if (offs != EMPTY_OFFS)
         {
            return readItem();
         }
      }
      return peek();
   }

   /**
    * Returns the number of remaining items in queue.
    */
   public synchronized long size()
   {
      return size;
   }

   /**
    * Removes and returns the first item in the queue, blocks if queue is empty.
    */
   public ItemT take() throws InterruptedException
   {
      block();

      try
      {
         synchronized (this)
         {
            if (offs != EMPTY_OFFS)
            {
               ItemT result = readItem();
               size--;
               offs = channel.position();
               file.seek(0);
               if (offs == channel.size())
               {
                  truncate();
               }
               file.writeLong(size);
               file.writeLong(offs);
               return result;
            }
         }
         return take();
      } catch (IOException e)
      {
         throw new RuntimeException(e);
      }
   }

   /**
    * Throw away all items and reset the file.
    */
   public synchronized void truncate()
   {
      try
      {
         offs = EMPTY_OFFS;
         file.setLength(HEADER_SIZE);
         size = 0;
      } catch (IOException e)
      {
         throw new RuntimeException(e);
      }
   }

   /**
    * Block until an item is available.
    */
   protected void block() throws InterruptedException
   {
      while (offs == EMPTY_OFFS)
      {
         try
         {
            synchronized (this)
            {
               wait();
               file.seek(LONG_SIZE);
               offs = file.readLong();
            }
         } catch (IOException e)
         {
            throw new RuntimeException(e);
         }
      }
   }

   /**
    * Read and return item.
    */
   @SuppressWarnings("unchecked")
   protected ItemT readItem()
   {
      try
      {
         channel.position(offs);
         return (ItemT) new ObjectInputStream(inputStream).readObject();
      } catch (ClassNotFoundException e)
      {
         throw new RuntimeException(e);
      } catch (IOException e)
      {
         throw new RuntimeException(e);
      }
   }
}

Answer 1

A big performance killer in your code is the use of "rwd" mode to open the file. The "d" forces every write to be synchronously written on the physical disk.

A quick benchmark gives me a 500x speed improvement just by removing the "d" in your code. I did a profiling of your code, and it appears that the majority of the time is spent in the writeLong function. This is because writeLong internally calls write eight times, and each time a physical write is performed !

You should remove this "d" mode and insert flushing instructions at strategic places. I think FileChannel.force(false) is the right method for that. Adding this instruction in add, take and defrag gives me a 12x speed improvement compared to the original code.

Answer 2

A few random notes:

1. It does not seem completely thread-safe. The offs field sometimes is read outside any synchronized block.

   [...] synchronization has no effect unless both read and write operations are synchronized.

   From Effective Java, 2nd Edition, Item 66: Synchronize access to shared mutable data.

2. I'd consider using separate file for every item and one for the meta information (list of the name of the corresponding files).

3. If you don't use checked exceptions and rethrow every exception catching Exceptions directly would be simpler.

   } catch (ClassNotFoundException e) {
       throw new RuntimeException(e);
   } catch (IOException e) {
       throw new RuntimeException(e);
   }
   

   The following is very similar (not exactly the same):

   } catch (Exception e) {
       throw new RuntimeException(e);
   }
   

4. I'd prefer concurrency utilities to wait and notify. (Effective Java, 2nd Edition, Item 69) A Semaphore might be a good choice here.

5. public ItemT getSomething() {
       block();
   
       synchronized (this) {
           if (queue if not empty) {
               return readItem();
           }
       }
       return getSomething();
   }
   

   Instead of the above recursive structure I'd use a loop. The recursion might cause StackOverflowErrors.

   public ItemT getSomething() {
       block();
   
       while (true) {
           synchronized (this) {
               if (queue if not empty) {
                   return readItem();
               }
           }
       }
   }
   

Answer 3

In addition to the great points by palacsint and barjak I have a few things to consider.

There are two ways you could increase the concurrency. The only thing that must be serialized is the head/tail pointer update.

This assumes that the order in which items are written to and read from the disk does not need to match the order in which the calls to add and take occur. This seems to be the case given the structure as it is now.

First, serialize in parallel, no pun intended. Each thread can perform its own object serialization without making the other threads wait.

Here's how add would look:

add ( ItemT item )
    serialize item to a byte array
    lock file
        write byte count
        write bytes
        update size

It's been a long time since I played with NIO, and I never used it in a real project, but I'm pretty sure it provides facilities for efficiently copying between ByteBuffers. You can see I've added a byte count to each item. This is so you can know how many bytes to read from the file and advance the pointer.

Second, let the I/O subsystem decide how best to order the seek/read/write steps by moving the read/write of the byte arrays outside synchronization. Since you've already serialized the object during add, you know exactly how many bytes to read/write and advance the pointers.

Now add is maximally concurrent:

add ( ItemT item )
    serialize item to byte array
    insert entry size in front of byte array
    lock tail pointer
        advance tail pointer by entry size
    write entry

You can avoid byte shuffling by inserting a 0 for the size before serializing instead of performing an insertion after calculating the size. The key is that "write entry" should be an atomic "append byte buffer to file" operation which again I believe NIO provides.

Here are some random notes I made while reading the code:

1. Do you have to instantiate new object input/output streams every time? Depending on how expensive this is, you may want to store them in a ThreadLocal, especially when using the byte arrays above.

2. You are seeking twice for every operation which could have a large cost if the block containing the header is far from the head/tail of the queue. Could you tolerate writing the size of the queue less frequently or in a separate thread? Are you writing to disk for fault tolerance or because it might grow too large to fit in memory? If the latter, keep it in memory only and write it when the application terminates or every x operations.

3. Why do you need EMPTY_OFFS? Can you use HEADER_SIZE as the initial offset for an empty queue?

4. Abbreviating offset to offs is terribad! It looks like the plural form of off and doesn't add any value. Pay those extra two keystrokes for clarity and call it exercise if you need an excuse. :)

Answer 4

The problem is that you are always seeking. This slows you down very much on a conventional hard disk. The solution is to only append to a file and do compaction later. Such data structures are called journals. See Journal.IO for more details.