Take the 2-minute tour ×
Code Review Stack Exchange is a question and answer site for peer programmer code reviews. It's 100% free, no registration required.

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);
      }
   }
}
share|improve this question
    
So why can't you afford 3rd party dependencies? It's actually simpler to use them in a case like this. Performance tuning and error handling for I/O is... hard. –  Martijn Verburg Oct 24 '12 at 8:32
    
Yeah I know it's hard, been there done that. I have no choice right here though and solving problems is what programming is all about. –  trialcodr Oct 24 '12 at 9:58
    
Any ideas on how to get rid of the ObjectStream instantiations in add() / readItem()? –  trialcodr Oct 25 '12 at 8:50
add comment

migrated from programmers.stackexchange.com Oct 24 '12 at 9:53

This question came from our site for professional programmers interested in conceptual questions about software development.

4 Answers

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();
                }
            }
        }
    }
    
share|improve this answer
    
1. Thanks, missed that one. 2. I'm afraid separate files for each item won't scale to hundreds of thousands of items. Maybe one file for headers and one for data would be a good compromise? 3. It's not that I want to catch everything, I want to catch everything the code is throwing right now. 4. I tried changing to a semaphore and it works like before but wait()/notify() is a simpler construct and better fit as I don't really need multiple permits. Care to elaborate on why you prefer the Semaphore? 5. You're right of course, thanks. –  trialcodr Oct 25 '12 at 8:43
    
@trialcodr: 2. Another idea: you could use separate folders (one folder for every 100 file). 3: Semaphore seemed the most suitable from the concurrency framework, I've not thought too much about this. Effective Java is very convincing about not using wait/notify. –  palacsint Oct 25 '12 at 11:20
    
The problem with separate files and folders is that you normally would require more seeking as there is a different position on the disk for every file/folder and the file system data needs to be updated. There was a good discussion on the HornetQ Journal and why this is fast. But I can't find it at the moment. –  SpaceTrucker Oct 25 '12 at 11:54
    
Just for the record, I did a quick test of splitting the file into two. One for the book keeping and one for data. I was very surprised to find that it was actually slower than the naive single file solution. That might change when/if the data file grows really big though. –  trialcodr Oct 26 '12 at 12:08
add comment

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.

share|improve this answer
add comment

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. :)

share|improve this answer
add comment

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.

share|improve this answer
    
Am not :) add() appends at eof, take() picks the first non-taken from the start and compaction happens whenever defrag() is called. If you add or take several items in a row the seeks shouldn't make a difference as it's already in the right place. –  trialcodr Oct 24 '12 at 9:57
add comment

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

Not the answer you're looking for? Browse other questions tagged or ask your own question.