Ticket #6746: alex.patch

new file components/blitz/src/ome/services/blitz/impl/BackgroundTiffWriter.java

@@ +1 @@
+package ome.services.blitz.impl;
+
+import java.io.IOException;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import loci.formats.FormatException;
+import loci.formats.out.TiffWriter;
+
+/**
+ * Simple wrapper object for the BioFormats TiffWriter class that allows the 
+ * {@link loci.formats.out.TiffWriter#saveBytes(int, byte[])} method to be called 
+ * in a background thread. This allows a producer of image bytes to operate 
+ * simultaneously with the consumer (BioFormats):
+ * 
+ * <pre>
+    TiffWriter writer = new TiffWriter()
+    
+    // TODO: Initialise writer here
+
+    BackgroundTiffWriter BackgroundTiffWriter = new BackgroundTiffWriter(writer);
+    Thread thread = new Thread(BackgroundTiffWriter);
+    thread.start();
+    
+    try {
+        for ( &lt;get planes loop&gt; ) {
+            byte[] plane = new byte[planeSize];
+            
+            // TODO: fill byte[] array
+
+            BackgroundTiffWriter.saveBytes(plane);
+        }
+        BackgroundTiffWriter.finalise();
+        thread.join();
+    }
+    catch (Exception e) {
+        BackgroundTiffWriter.shutdown();
+    } 
+    finally {
+        // The original writer is not closed by the BackgroundTiffWriter 
+        writer.close();
+    }
+        </pre>
+ * 
+ * <p>
+ * The class uses a BlockingQueue to store successive byte arrays passed to 
+ * the {@link #saveBytes(byte[])} method. These are sequentially dequeued and 
+ * passed to the TiffWriter. The size of the blocking queue can be set in the
+ * constructor. Consequently the producer of the bytes should not get too far
+ * ahead of the consumer TiffWriter class.    
+ * 
+ * <p>
+ * The BackgroundTiffWriter expects the {@link loci.formats.TiffWriter } class
+ * to be correctly initialised to accept bytes to the 
+ * {@link loci.formats.TiffWriter#saveBytes(int, byte[])} method. 
+ * 
+ * <p>
+ * Note: The TiffWriter is not closed by this class. 
+ * 
+ * @author Alex Herbert, GDSC
+ */
+public class BackgroundTiffWriter implements Runnable {
+
+        private final static Log log = LogFactory.getLog(BackgroundTiffWriter.class);
+
+        private TiffWriter writer;
+        private BlockingQueue<byte[]> queue;
+        private volatile boolean closed = false;
+        
+        private final Object lock = new Object();
+        
+        /**
+         * Default constructor
+         * @param writer An opened and initialised TiffWriter
+         */
+        public BackgroundTiffWriter(TiffWriter writer) {
+                this.writer = writer;
+                init(5);
+        }
+        
+        /**
+         * Constructor
+         * @param writer An opened and initialised TiffWriter
+         * @param queueSize The size of the blocking queue
+         */
+        public BackgroundTiffWriter(TiffWriter writer, int queueSize) {
+                this.writer = writer;
+                init(queueSize);
+        }
+        
+        /**
+         * Initialise the blocking queue
+         * @param queueSize
+         */
+        private void init(int queueSize) {
+                queue = new ArrayBlockingQueue<byte[]>(queueSize);
+        }
+
+        /* (non-Javadoc)
+         * @see java.lang.Runnable#run()
+         */
+        @Override
+        public void run() {
+                Thread.currentThread();
+                try {
+                        int no = 0;
+                        byte[] bytes;
+                        while (!closed) {
+                                bytes = queue.take();
+                                if (bytes == null || bytes.length == 0 || closed)
+                                        break;
+                                synchronized (lock) {
+                                        writer.saveBytes(no, bytes);
+                                }
+                                no++;
+                        }
+                }
+                catch (InterruptedException e) {
+                        log.info(e.toString());
+                        throw new RuntimeException(e);
+                } catch (FormatException e) {
+                        log.error(e.toString());
+                        throw new RuntimeException(e);
+                } catch (IOException e) {
+                        log.error(e.toString());
+                        throw new RuntimeException(e);
+                } finally {
+                        closed = true;
+
+                        // Free remaining memory
+                        queue.clear();
+                        queue = null;
+                        writer = null;
+
+                        notifyAll();
+                }
+        }
+
+        /**
+         * Adds the bytes to the processing queue. A zero length byte array indicates
+         * that no more input is expected.
+         * @param bytes The bytes
+         * @throws RuntimeException If the writer is closed or the thread is interrupted
+         *                          adding to the queue.
+         */
+        public void saveBytes(byte[] bytes) throws RuntimeException {
+                if (closed)
+                        throw new RuntimeException("The writer has been closed");
+                try {
+                        queue.put(bytes);
+                } catch (InterruptedException iex) {
+                        Thread.currentThread().interrupt();
+                        throw new RuntimeException("Unexpected interruption", iex);
+                }
+        }
+        
+        /**
+         * Send a signal to the queue to indicate no more input is expected.
+         */
+        public void finalise() {
+                saveBytes(new byte[0]);
+        }
+
+        /**
+         * Finish the current write process and stop further writing. 
+         * Use this method to end the processing in the event of error.
+         */
+        public void shutdown() {
+                if (closed)
+                        return;
+                
+                // Ensure that the current write process is over. This allows the 
+                // TiffWriter to be closed after shutting down the BackgroundTiffWriter.
+                synchronized (lock) {
+                        closed = true;
+                        log.info("Terminating ...");
+                }
+        }
+
+        /**
+         * @return True if no further writes are possible
+         */
+        public boolean isClosed() {
+                return closed;
+        }
+}

components/blitz/src/ome/services/blitz/impl/ExporterI.java

@@ -98 +98 @@
      * @see ticket:6520
      */
     private final static long BIG_TIFF_SIZE = 2L * Integer.MAX_VALUE;
+    
+    /**
+     * The size above which a tiff file should be written using a background
+     * thread. This is to avoid the thread overhead when small files are
+     * exported.
+     */
+    private final static long MULTI_THREAD_EXPORT_SIZE = 1024L * 1024;  // 1MB
 
 
     /**
@@ -357 +364 @@
                             RawPixelsStore raw = null;
                             OmeroReader reader = null;
                             TiffWriter writer = null;
+                            BackgroundTiffWriter backgroundWriter = null;
                             try {
 
                                 Image image = retrieve.getImage(0);
@@ -387 +395 @@
                                 long dSize = getDataBytes(reader);
                                 final boolean bigtiff =
                                     ( ( mSize + dSize ) > BIG_TIFF_SIZE );
+                                final boolean multiThread = ( dSize > MULTI_THREAD_EXPORT_SIZE );
                                 if (bigtiff) {
                                     writer.setBigTiff(true);
                                 }
@@ -397 +406 @@
                                 log.info(String.format(
                                             "Using big TIFF? %s mSize=%d " +
                                             "dSize=%d planeCount=%d " +
-                                            "planeSize=%d",
+                                            "planeSize=%d multiThread=%s",
                                             bigtiff, mSize, dSize,
-                                            planeCount, planeSize));
-                                byte[] plane = new byte[planeSize];
-                                for (int i = 0; i < planeCount; i++) {
-                                    int[] zct = FormatTools.getZCTCoords(
-                                        retrieve.getPixelsDimensionOrder(0).getValue(),
-                                        reader.getSizeZ(), reader.getSizeC(), reader.getSizeT(),
-                                        planeCount, i);
-                                    int readerIndex = reader.getIndex(zct[0], zct[1], zct[2]);
-                                    reader.openBytes(readerIndex, plane);
-                                    writer.saveBytes(i, plane);
+                                            planeCount, planeSize,
+                                            multiThread));
+
+                                // Perform some debug timing for the export
+                                long start = System.nanoTime();
+                                long checkpoint1, checkpoint2;
+                                long openTime = 0, saveTime = 0, waitTime = 0;
+                                
+                                if (multiThread)
+                                {
+                                    // Save the image using a different thread
+                                    backgroundWriter = new BackgroundTiffWriter(writer, 10);
+                                    Thread thread = new Thread(backgroundWriter);
+                                    thread.start();
+                                    
+                                    for (int i = 0; i < planeCount; i++) {
+                                        int[] zct = FormatTools.getZCTCoords(
+                                            retrieve.getPixelsDimensionOrder(0).getValue(),
+                                            reader.getSizeZ(), reader.getSizeC(), reader.getSizeT(),
+                                            planeCount, i);
+                                        int readerIndex = reader.getIndex(zct[0], zct[1], zct[2]);
+                                        byte[] plane = new byte[planeSize];
+                                        checkpoint1 = System.nanoTime();
+                                        reader.openBytes(readerIndex, plane);
+                                        checkpoint2 = System.nanoTime();
+                                        openTime += checkpoint2 - checkpoint1;
+                                        backgroundWriter.saveBytes(plane);
+                                        checkpoint1 = System.nanoTime();
+                                        saveTime += checkpoint1 - checkpoint2;
+                                    }
+                                    
+                                    // Wait for background writer
+                                    checkpoint1 = System.nanoTime();
+                                        backgroundWriter.finalise();
+                                        thread.join();
+                                    waitTime = System.nanoTime() - checkpoint1;
+                                }
+                                else
+                                {
+                                    byte[] plane = new byte[planeSize];
+                                    
+                                    for (int i = 0; i < planeCount; i++) {
+                                        int[] zct = FormatTools.getZCTCoords(
+                                            retrieve.getPixelsDimensionOrder(0).getValue(),
+                                            reader.getSizeZ(), reader.getSizeC(), reader.getSizeT(),
+                                            planeCount, i);
+                                        int readerIndex = reader.getIndex(zct[0], zct[1], zct[2]);
+                                        checkpoint1 = System.nanoTime();
+                                        reader.openBytes(readerIndex, plane);
+                                        checkpoint2 = System.nanoTime();
+                                        openTime += checkpoint2 - checkpoint1;
+                                        writer.saveBytes(i, plane);
+                                        checkpoint1 = System.nanoTime();
+                                        saveTime += checkpoint1 - checkpoint2;
+                                    }
                                 }
+                                
+                                long end = System.nanoTime();
+                                String msg = String.format("Saved image %s in %f ms. open = %f ms. save = %f ms. wait = %f ms\n", 
+                                                retrieve.getImageName(0), 
+                                                (end-start) / 1000000.0, 
+                                                openTime / 1000000.0, 
+                                                saveTime / 1000000.0,
+                                                waitTime / 1000000.0);
+                                
+                                log.info(msg);
+
                                 retrieve = null;
 
                                 try {
@@ -418 +483 @@
                                     // Nulling to prevent another exception
                                     writer = null;
                                 }
-
+                                
                                     __cb.ice_response(file.length());
                                 } catch (Exception e) {
                                     omero.InternalException ie = new omero.InternalException(
@@ -427 +492 @@
                                     IceMapper.fillServerError(ie, e);
                                     __cb.ice_exception(ie);
                                 } finally {
-                                    cleanup(raw, reader, writer);
+                                    cleanup(raw, reader, writer, backgroundWriter);
                                 }
 
                             return null; // see calls to __cb above
                         }
 
                         private void cleanup(RawPixelsStore raw,
-                                OmeroReader reader, IFormatWriter writer) {
+                                OmeroReader reader, IFormatWriter writer,
+                                BackgroundTiffWriter backgroundWriter) {
                             try {
                                 if (raw != null) {
                                     raw.close();
@@ -450 +516 @@
                                 log.error("Error closing reader", e);
                             }
                             try {
+                                if (backgroundWriter != null) {
+                                        backgroundWriter.shutdown();
+                                        backgroundWriter = null;
+                                }
+                            } catch (Exception e) {
+                                log.error("Error closing background writer", e);
+                            }
+                            try {
                                 if (writer != null) {
                                     writer.close();
                                 }