* START -> waiting -> config -> output -> waiting -> config ... ** * @author Josh Moore, josh at glencoesoftware.com * @since 4.1 */ public class ExporterI extends AbstractAmdServant implements _ExporterOperations, ServiceFactoryAware, BlitzOnly { private final static Log log = LogFactory.getLog(ExporterI.class); private final static int MAX_SIZE = 1024 * 1024; /** * Utility enum for asserting the state of Exporter instances. */ private enum State { config, output; static State check(ExporterI self) { if (self.file != null && self.retrieve != null) { throw new InternalException("Doing 2 things at once"); } if (self.retrieve == null) { return output; } return config; } } private/* final */ServiceFactoryI factory; private volatile OmeroMetadata retrieve; /** * Reference to the temporary file which is currently being output. If null, * then no generate method has been called. */ private volatile File file; /** * Encapsulates the logic for creating new LSIDs and comparing existing ones * to the internal value for this DB. */ private final DatabaseIdentity databaseIdentity; /** LOCI OME-XML service for working with OME-XML. */ private OMEXMLService service; public ExporterI(BlitzExecutor be, DatabaseIdentity databaseIdentity) throws DependencyException { super(null, be); this.databaseIdentity = databaseIdentity; retrieve = new OmeroMetadata(databaseIdentity); loci.common.services.ServiceFactory sf = new loci.common.services.ServiceFactory(); service = sf.getInstance(OMEXMLService.class); } public void setServiceFactory(ServiceFactoryI sf) throws ServerError { this.factory = sf; } // Interface methods // ========================================================================= public void addImage_async(AMD_Exporter_addImage __cb, final long id, Current __current) throws ServerError { State state = State.check(this); ServerError se = assertConfig(state); if (se != null) { __cb.ice_exception(se); } retrieve.addImage(new ImageI(id, false)); __cb.ice_response(); return; } /** * Generate XML and return the length */ public void generateXml_async(AMD_Exporter_generateXml __cb, Current __current) throws ServerError { State state = State.check(this); ServerError se = assertConfig(state); if (se != null) { __cb.ice_exception(se); } // sets retrieve, then file, then unsets retrieve do_xml(__cb); return; } /** * */ public void generateTiff_async(AMD_Exporter_generateTiff __cb, Current __current) throws ServerError { State state = State.check(this); ServerError se = assertConfig(state); if (se != null) { __cb.ice_exception(se); } // sets retrieve, then file, then unsets retrieve do_tiff(__cb); return; } public void read_async(AMD_Exporter_read __cb, long pos, int size, Current __current) throws ServerError { omero.ApiUsageException aue; State state = State.check(this); switch (state) { case config: aue = new omero.ApiUsageException(null, null, "Call a generate method first"); __cb.ice_exception(aue); return; case output: try { __cb.ice_response(read(pos, size)); } catch (Exception e) { if (e instanceof ServerError) { __cb.ice_exception(e); } else { omero.InternalException ie = new omero.InternalException(null, null, "Error during read"); IceMapper.fillServerError(ie, e); __cb.ice_exception(ie); } } return; default: throw new InternalException("Unknown state: " + state); } } // State methods // ========================================================================= /** * Transition from waiting to config */ private ServerError assertConfig(State state) { switch (state) { case config: return null; case output: return new omero.ApiUsageException(null, null, "Cannot configure during output"); default: return new omero.InternalException(null, null, "Unknown state: " + state); } } /** * Transition from waiting to config */ private void startConfig() { if (file != null) { file.delete(); file = null; } } /** * Transitions from config to output. */ private void do_xml(final AMD_Exporter_generateXml __cb) { try { factory.executor.execute(factory.principal, new Executor.SimpleWork(this, "generateXml") { @Transactional(readOnly = true) public Object doWork(Session session, ServiceFactory sf) { retrieve.initialize(session); IMetadata xmlMetadata = null; try { xmlMetadata = convertXml(retrieve); } catch (ServiceException e) { log.error(e); return null; } if (xmlMetadata != null) { Object root = xmlMetadata.getRoot(); if (root instanceof OME) { OME node = (OME) root; try { DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance(); DocumentBuilder parser = factory.newDocumentBuilder(); Document document = parser.newDocument(); Element element = node.asXMLElement(document); document.appendChild(element); file = TempFileManager.create_path( "__omero_export__", ".ome.xml"); file.deleteOnExit(); FileOutputStream fos = new FileOutputStream( file); DOMUtil.writeXML(fos, document); fos.close(); retrieve = null; __cb.ice_response(file.length()); return null; // ONLY VALID EXIT } catch (IOException ioe) { log.error(ioe); } catch (TransformerException e) { log.error(e); } catch (ParserConfigurationException e) { log.error(e); } } } return null; } }); } catch (Exception e) { IceMapper mapper = new IceMapper(); Ice.UserException ue = mapper.handleException(e, factory.getExecutor().getContext()); __cb.ice_exception(ue); } } /** * Transitions from config to output. */ private void do_tiff(final AMD_Exporter_generateTiff __cb) { try { factory.executor.execute(factory.principal, new Executor.SimpleWork(this, "generateTiff") { @Transactional(readOnly = true) public Object doWork(Session session, ServiceFactory sf) { retrieve.initialize(session); int num = retrieve.sizeImages(); if (num != 1) { omero.ApiUsageException a = new omero.ApiUsageException( null, null, "Only one image supported for TIFF, not "+num); __cb.ice_exception(a); return null; } RawPixelsStore raw = null; OmeroReader reader = null; ImageWriter writer = null; BackgroundWriter backgroundWriter = null; try { Image image = retrieve.getImage(0); Pixels pix = image.getPixels(0); file = TempFileManager.create_path("__omero_export__", ".ome.tiff"); raw = sf.createRawPixelsStore(); raw.setPixelsId(pix.getId().getValue(), true); reader = new OmeroReader(raw, pix); reader.setId("OMERO"); writer = new ImageWriter(); writer.setMetadataRetrieve(retrieve); writer.setWriteSequentially(true); writer.setId(file.getAbsolutePath()); long start = System.nanoTime(); int planeCount = reader.planes; int planeSize = raw.getPlaneSize(); //byte[] plane = new byte[planeSize]; long checkpoint = System.nanoTime(); long t = 0; long readTime = 0; long writeTime = 0; log.info("Creating backgroundWriter"); backgroundWriter = new BackgroundWriter(writer); Thread thread = new Thread(backgroundWriter); log.info("Starting backgroundWriter"); thread.start(); log.info("Reading image"); 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]); // This method eventually calls: // raw.getPlane(zct[0], zct[1], zct[2]) // on the RawPixelsStore log.info("Reading plane " + i); byte[] plane = new byte[planeSize]; reader.openBytes(readerIndex, plane); t = System.nanoTime(); readTime += t - checkpoint; checkpoint = t; //writer.saveBytes(i, plane); log.info("Saving plane " + i); backgroundWriter.saveBytes(plane); t = System.nanoTime(); writeTime += t - checkpoint; checkpoint = t; } log.info("Finallising backgroundWriter"); try{ backgroundWriter.finalise(); } finally { // Nulling to prevent another exception in cleanup(...) backgroundWriter = null; } // Wait for background writer while (thread.isAlive()) { log.info("Waiting for backgroundWriter"); thread.join(4000); } log.info("backgroundWriter done"); try { writer.close(); } finally { // Nulling to prevent another exception writer = null; } long end = System.nanoTime(); String msg = String.format("Saved image %s in %f ms. openBytes %f ms. saveBytes %f ms\n", retrieve.getImageName(0), (end-start) / 1000000.0, readTime / 1000000.0, writeTime / 1000000.0); log.info(msg); retrieve = null; __cb.ice_response(file.length()); } catch (Exception e) { omero.InternalException ie = new omero.InternalException( null, null, "Error during TIFF generation"); IceMapper.fillServerError(ie, e); __cb.ice_exception(ie); } finally { cleanup(raw, reader, writer, backgroundWriter); } return null; // see calls to __cb above } private void cleanup(RawPixelsStore raw, OmeroReader reader, ImageWriter writer, BackgroundWriter backgroundWriter) { try { if (raw != null) { raw.close(); } } catch (Exception e) { log.error("Error closing pix", e); } try { if (reader != null) { reader.close(); } } catch (Exception e) { 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(); } } catch (Exception e) { log.error("Error closing writer", e); } } }); } catch (Exception e) { IceMapper mapper = new IceMapper(); Ice.UserException ue = mapper.handleException(e, factory.getExecutor().getContext()); __cb.ice_exception(ue); } } /** * Read size bytes, and transition to "waiting" If any exception is thrown, * the offset for the current file will not be updated. */ private byte[] read(long pos, int size) throws ServerError { if (size > MAX_SIZE) { throw new ApiUsageException("Max read size is: " + MAX_SIZE); } byte[] buf = new byte[size]; RandomAccessFile ra = null; try { ra = new RandomAccessFile(file, "r"); long l = ra.length(); if (pos + size > l) { size = (int) (l - pos); } ra.seek(pos); int read = ra.read(buf); // Handle end of file if (read < 0) { buf = new byte[0]; } else if (read < size) { byte[] newBuf = new byte[read]; System.arraycopy(buf, 0, newBuf, 0, read); buf = newBuf; } } catch (IOException io) { throw new RuntimeException(io); } finally { if (ra != null) { try { ra.close(); } catch (IOException e) { log.warn("IOException on file close"); } } } return buf; } // XML Generation (public for testing) // ========================================================================= public IMetadata convertXml(MetadataRetrieve retrieve) throws ServiceException { IMetadata xmlMeta = service.createOMEXMLMetadata(); xmlMeta.createRoot(); service.convertMetadata(retrieve, xmlMeta); return xmlMeta; } public String generateXml(MetadataRetrieve retrieve) throws ServiceException { IMetadata xmlMeta = convertXml(retrieve); return service.getOMEXML(xmlMeta); } // Stateful interface methods // ========================================================================= public void preClose() { retrieve = null; if (file != null) { file.delete(); file = null; } } } components/blitz/src/ome/services/blitz/impl/BackgroundWriter.java 0000644 0135356 0001750 00000011413 11632125030 024210 0 ustar ah403 ah403 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.ImageWriter; /** * Simple wrapper object for the BioFormats ImageWriter class that allows the * {@link loci.formats.ImageWriter#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): * *
ImageWriter writer = new ImageWriter()
// TODO: Initialise writer here
BackgroundWriter backgroundWriter = new BackgroundWriter(writer);
Thread thread = new Thread(backgroundWriter);
thread.start();
try {
for ( <get planes loop> ) {
byte[] plane = new byte[planeSize];
// TODO: fill byte[] array
backgroundWriter.saveBytes(plane);
}
backgroundWriter.finalise();
thread.join();
}
catch (Exception e) {
backgroundWriter.shutdown();
}
finally {
writer.close();
}
*
* * 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 ImageWriter. 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 ImageWriter class. * *
* The BackgroundWriter expects the {@link loci.formats.ImageWriter } class * to be correctly initialised to accept bytes to the * {@link loci.formats.ImageWriter#saveBytes(int, byte[])} method. * *
* Note: The ImageWriter is not closed by this class.
*
* @author Alex Herbert, GDSC
*/
public class BackgroundWriter implements Runnable {
private final static Log log = LogFactory.getLog(BackgroundWriter.class);
private ImageWriter writer;
private BlockingQueue