[teiid-commits] teiid SVN: r3000 - in trunk: engine/src/main/java/org/teiid/common/buffer/impl and 7 other directories.

Author: shawkins Date: 2011-03-17 10:25:04 -0400 (Thu, 17 Mar 2011) New Revision: 3000 Added: trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java Removed: trunk/engine/src/main/java/org/teiid/query/processor/xml/XMLUtil.java Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java trunk/engine/src/main/java/org/teiid/query/function/aggregate/TextAgg.java trunk/engine/src/main/java/org/teiid/query/function/source/XMLSystemFunctions.java trunk/engine/src/main/java/org/teiid/query/processor/xml/DocumentInProgress.java trunk/engine/src/main/java/org/teiid/query/xquery/saxon/SaxonXQueryExpression.java trunk/runtime/src/main/java/org/teiid/transport/ObjectDecoder.java trunk/runtime/src/test/java/org/teiid/transport/TestCommSockets.java Log: TEIID-1511 ensuring that remote streams get proactively cleaned up Added: trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -0,0 +1,82 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + +package org.teiid.common.buffer; + +import java.io.BufferedInputStream; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStreamWriter; +import java.io.Writer; +import java.nio.charset.Charset; + +import org.teiid.common.buffer.FileStore.FileStoreOutputStream; +import org.teiid.core.types.DataTypeManager; +import org.teiid.core.types.InputStreamFactory; + +public final class FileStoreInputStreamFactory extends InputStreamFactory { + private final FileStore lobBuffer; + private FileStoreOutputStream fsos; + private String encoding; + + public FileStoreInputStreamFactory(FileStore lobBuffer, String encoding) { + this.encoding = encoding; + this.lobBuffer = lobBuffer; + this.lobBuffer.setCleanupReference(this); + } + + @Override + public InputStream getInputStream() throws IOException { + if (fsos != null && !fsos.bytesWritten()) { + return new ByteArrayInputStream(fsos.getBuffer(), 0, fsos.getCount()); + } + //TODO: adjust the buffer size, and/or develop a shared buffer strategy + return new BufferedInputStream(lobBuffer.createInputStream(0)); + } + + @Override + public long getLength() { + return lobBuffer.getLength(); + } + + public Writer getWriter() { + return new OutputStreamWriter(getOuputStream(), Charset.forName(encoding)); + } + + public FileStoreOutputStream getOuputStream() { + if (fsos == null) { + fsos = lobBuffer.createOutputStream(DataTypeManager.MAX_LOB_MEMORY_BYTES); + } + return fsos; + } + + @Override + public void free() throws IOException { + lobBuffer.remove(); + } + + @Override + public boolean isPersistent() { + return true; + } +} \ No newline at end of file Property changes on: trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -23,6 +23,7 @@ package org.teiid.common.buffer.impl; import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicInteger; import org.teiid.common.buffer.FileStore; import org.teiid.common.buffer.StorageManager; @@ -31,11 +32,15 @@ public class MemoryStorageManager implements StorageManager { + private AtomicInteger created = new AtomicInteger(); + private AtomicInteger removed = new AtomicInteger(); + public void initialize() { } @Override public FileStore createFileStore(String name) { + created.incrementAndGet(); return new FileStore() { private ByteBuffer buffer = ByteBuffer.allocate(1 << 16); @@ -53,6 +58,7 @@ @Override public synchronized void removeDirect() { + removed.incrementAndGet(); buffer = ByteBuffer.allocate(0); } @@ -70,4 +76,12 @@ } }; } + + public int getCreated() { + return created.get(); + } + + public int getRemoved() { + return removed.get(); + } } \ No newline at end of file Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -35,6 +35,7 @@ import org.teiid.client.SourceWarning; import org.teiid.common.buffer.BlockedException; import org.teiid.common.buffer.FileStore; +import org.teiid.common.buffer.FileStoreInputStreamFactory; import org.teiid.common.buffer.TupleSource; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidProcessingException; @@ -55,8 +56,7 @@ import org.teiid.dqp.internal.process.DQPCore.FutureWork; import org.teiid.dqp.message.AtomicRequestMessage; import org.teiid.dqp.message.AtomicResultsMessage; -import org.teiid.query.processor.xml.XMLUtil; -import org.teiid.query.processor.xml.XMLUtil.FileStoreInputStreamFactory; +import org.teiid.query.function.source.XMLSystemFunctions; import org.teiid.query.sql.symbol.SingleElementSymbol; import org.teiid.translator.DataNotAvailableException; import org.teiid.translator.TranslatorException; @@ -192,7 +192,7 @@ StandardXMLTranslator sxt = new StandardXMLTranslator((Source)value); SQLXMLImpl sqlxml; try { - sqlxml = XMLUtil.saveToBufferManager(dtm.getBufferManager(), sxt); + sqlxml = XMLSystemFunctions.saveToBufferManager(dtm.getBufferManager(), sxt); } catch (TeiidComponentException e) { throw new TeiidRuntimeException(e); } catch (TeiidProcessingException e) { Modified: trunk/engine/src/main/java/org/teiid/query/function/aggregate/TextAgg.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/function/aggregate/TextAgg.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/engine/src/main/java/org/teiid/query/function/aggregate/TextAgg.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -31,13 +31,13 @@ import javax.sql.rowset.serial.SerialBlob; import org.teiid.common.buffer.FileStore; +import org.teiid.common.buffer.FileStoreInputStreamFactory; import org.teiid.common.buffer.FileStore.FileStoreOutputStream; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidProcessingException; import org.teiid.core.types.BlobImpl; import org.teiid.core.types.BlobType; import org.teiid.core.types.Streamable; -import org.teiid.query.processor.xml.XMLUtil.FileStoreInputStreamFactory; import org.teiid.query.sql.symbol.DerivedColumn; import org.teiid.query.sql.symbol.ElementSymbol; import org.teiid.query.sql.symbol.TextLine; Modified: trunk/engine/src/main/java/org/teiid/query/function/source/XMLSystemFunctions.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/function/source/XMLSystemFunctions.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/engine/src/main/java/org/teiid/query/function/source/XMLSystemFunctions.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -78,17 +78,20 @@ import org.json.simple.parser.ContentHandler; import org.json.simple.parser.JSONParser; import org.json.simple.parser.ParseException; +import org.teiid.common.buffer.BufferManager; +import org.teiid.common.buffer.FileStore; +import org.teiid.common.buffer.FileStoreInputStreamFactory; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidProcessingException; import org.teiid.core.types.ClobImpl; import org.teiid.core.types.ClobType; import org.teiid.core.types.SQLXMLImpl; +import org.teiid.core.types.Streamable; import org.teiid.core.types.XMLTranslator; import org.teiid.core.types.XMLType; import org.teiid.core.types.XMLType.Type; import org.teiid.query.eval.Evaluator; import org.teiid.query.function.CharsetUtils; -import org.teiid.query.processor.xml.XMLUtil; import org.teiid.query.util.CommandContext; import org.teiid.translator.WSConnection.Util; @@ -245,7 +248,7 @@ final Transformer transformer = factory.newTransformer(styleSource); //this creates a non-validated sqlxml - it may not be valid xml/root-less xml - SQLXMLImpl result = XMLUtil.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { + SQLXMLImpl result = XMLSystemFunctions.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { @Override public void translate(Writer writer) throws TransformerException { @@ -273,7 +276,7 @@ return null; } - XMLType result = new XMLType(XMLUtil.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { + XMLType result = new XMLType(XMLSystemFunctions.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { @Override public void translate(Writer writer) throws TransformerException, @@ -309,7 +312,7 @@ */ public static XMLType xmlElement(CommandContext context, final String name, final Evaluator.NameValuePair<String>[] namespaces, final Evaluator.NameValuePair<?>[] attributes, final List<?> contents) throws TeiidComponentException, TeiidProcessingException { - XMLType result = new XMLType(XMLUtil.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { + XMLType result = new XMLType(XMLSystemFunctions.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { @Override public void translate(Writer writer) throws TransformerException, @@ -383,7 +386,7 @@ return singleValue; } - XMLType result = new XMLType(XMLUtil.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { + XMLType result = new XMLType(XMLSystemFunctions.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { @Override public void translate(Writer writer) throws TransformerException, @@ -671,7 +674,7 @@ private static SQLXML jsonToXml(CommandContext context, final String rootName, final Reader r) throws TeiidComponentException, TeiidProcessingException { - XMLType result = new XMLType(XMLUtil.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { + XMLType result = new XMLType(XMLSystemFunctions.saveToBufferManager(context.getBufferManager(), new XMLTranslator() { @Override public void translate(Writer writer) throws TransformerException, @@ -700,9 +703,31 @@ result.setType(Type.DOCUMENT); return result; } + + /** + * This method saves the given XML object to the buffer manager's disk process + * Documents less than the maxMemorySize will be held directly in memory + */ + public static SQLXMLImpl saveToBufferManager(BufferManager bufferMgr, XMLTranslator translator) + throws TeiidComponentException, TeiidProcessingException { + boolean success = false; + final FileStore lobBuffer = bufferMgr.createFileStore("xml"); //$NON-NLS-1$ + FileStoreInputStreamFactory fsisf = new FileStoreInputStreamFactory(lobBuffer, Streamable.ENCODING); + try{ + Writer writer = fsisf.getWriter(); + translator.translate(writer); + writer.close(); + success = true; + return new SQLXMLImpl(fsisf); + } catch(IOException e) { + throw new TeiidComponentException(e); + } catch(TransformerException e) { + throw new TeiidProcessingException(e); + } finally { + if (!success && lobBuffer != null) { + lobBuffer.remove(); + } + } + } - public static void main(String[] args) { - System.out.println(Charset.availableCharsets()); - } - } Modified: trunk/engine/src/main/java/org/teiid/query/processor/xml/DocumentInProgress.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/xml/DocumentInProgress.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/engine/src/main/java/org/teiid/query/processor/xml/DocumentInProgress.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -33,12 +33,12 @@ import net.sf.saxon.TransformerFactoryImpl; import org.teiid.common.buffer.FileStore; +import org.teiid.common.buffer.FileStoreInputStreamFactory; import org.teiid.core.TeiidComponentException; import org.teiid.core.types.SQLXMLImpl; import org.teiid.logging.LogManager; import org.teiid.logging.MessageLevel; import org.teiid.query.mapping.xml.MappingNodeConstants; -import org.teiid.query.processor.xml.XMLUtil.FileStoreInputStreamFactory; import org.xml.sax.SAXException; Deleted: trunk/engine/src/main/java/org/teiid/query/processor/xml/XMLUtil.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/xml/XMLUtil.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/engine/src/main/java/org/teiid/query/processor/xml/XMLUtil.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -1,122 +0,0 @@ -/* - * JBoss, Home of Professional Open Source. - * See the COPYRIGHT.txt file distributed with this work for information - * regarding copyright ownership. Some portions may be licensed - * to Red Hat, Inc. under one or more contributor license agreements. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA - * 02110-1301 USA. - */ - -package org.teiid.query.processor.xml; - -import java.io.BufferedInputStream; -import java.io.ByteArrayInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStreamWriter; -import java.io.Writer; -import java.nio.charset.Charset; - -import javax.xml.transform.TransformerException; - -import org.teiid.common.buffer.BufferManager; -import org.teiid.common.buffer.FileStore; -import org.teiid.common.buffer.FileStore.FileStoreOutputStream; -import org.teiid.core.TeiidComponentException; -import org.teiid.core.TeiidProcessingException; -import org.teiid.core.types.DataTypeManager; -import org.teiid.core.types.InputStreamFactory; -import org.teiid.core.types.SQLXMLImpl; -import org.teiid.core.types.Streamable; -import org.teiid.core.types.XMLTranslator; - -/** - * Utility methods to be used with the XML and XQuery processing. - */ -public class XMLUtil { - - public static final class FileStoreInputStreamFactory extends InputStreamFactory { - private final FileStore lobBuffer; - private final FileStoreOutputStream fsos; - private String encoding; - - public FileStoreInputStreamFactory(FileStore lobBuffer, String encoding) { - this.encoding = encoding; - this.lobBuffer = lobBuffer; - fsos = lobBuffer.createOutputStream(DataTypeManager.MAX_LOB_MEMORY_BYTES); - this.lobBuffer.setCleanupReference(this); - } - - @Override - public InputStream getInputStream() throws IOException { - if (!fsos.bytesWritten()) { - return new ByteArrayInputStream(fsos.getBuffer(), 0, fsos.getCount()); - } - //TODO: adjust the buffer size, and/or develop a shared buffer strategy - return new BufferedInputStream(lobBuffer.createInputStream(0)); - } - - @Override - public long getLength() { - return lobBuffer.getLength(); - } - - public Writer getWriter() { - return new OutputStreamWriter(fsos, Charset.forName(encoding)); - } - - public FileStoreOutputStream getOuputStream() { - return fsos; - } - - @Override - public void free() throws IOException { - lobBuffer.remove(); - } - - @Override - public boolean isPersistent() { - return true; - } - } - - /** - * This method saves the given XML object to the buffer manager's disk process - * Documents less than the maxMemorySize will be held directly in memory - */ - public static SQLXMLImpl saveToBufferManager(BufferManager bufferMgr, XMLTranslator translator) - throws TeiidComponentException, TeiidProcessingException { - boolean success = false; - final FileStore lobBuffer = bufferMgr.createFileStore("xml"); //$NON-NLS-1$ - FileStoreInputStreamFactory fsisf = new FileStoreInputStreamFactory(lobBuffer, Streamable.ENCODING); - try{ - Writer writer = fsisf.getWriter(); - translator.translate(writer); - writer.close(); - success = true; - return new SQLXMLImpl(fsisf); - } catch(IOException e) { - throw new TeiidComponentException(e); - } catch(TransformerException e) { - throw new TeiidProcessingException(e); - } finally { - if (!success && lobBuffer != null) { - lobBuffer.remove(); - } - } - } - -} \ No newline at end of file Modified: trunk/engine/src/main/java/org/teiid/query/xquery/saxon/SaxonXQueryExpression.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/xquery/saxon/SaxonXQueryExpression.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/engine/src/main/java/org/teiid/query/xquery/saxon/SaxonXQueryExpression.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -83,7 +83,6 @@ import org.teiid.query.QueryPlugin; import org.teiid.query.analysis.AnalysisRecord; import org.teiid.query.function.source.XMLSystemFunctions; -import org.teiid.query.processor.xml.XMLUtil; import org.teiid.query.sql.lang.XMLTable; import org.teiid.query.sql.lang.XMLTable.XMLColumn; import org.teiid.query.sql.symbol.DerivedColumn; @@ -446,7 +445,7 @@ if (next != null) { type = Type.CONTENT; } - SQLXMLImpl xml = XMLUtil.saveToBufferManager(bufferManager, new XMLTranslator() { + SQLXMLImpl xml = XMLSystemFunctions.saveToBufferManager(bufferManager, new XMLTranslator() { @Override public void translate(Writer writer) throws TransformerException, Modified: trunk/runtime/src/main/java/org/teiid/transport/ObjectDecoder.java =================================================================== --- trunk/runtime/src/main/java/org/teiid/transport/ObjectDecoder.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/runtime/src/main/java/org/teiid/transport/ObjectDecoder.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -22,10 +22,7 @@ */ package org.teiid.transport; -import java.io.BufferedInputStream; import java.io.BufferedOutputStream; -import java.io.IOException; -import java.io.InputStream; import java.io.ObjectOutputStream; import java.io.OutputStream; import java.io.StreamCorruptedException; @@ -40,8 +37,9 @@ import org.jboss.netty.handler.codec.serialization.CompatibleObjectEncoder; import org.jboss.netty.handler.codec.serialization.ObjectEncoder; import org.teiid.common.buffer.FileStore; +import org.teiid.common.buffer.FileStoreInputStreamFactory; import org.teiid.common.buffer.StorageManager; -import org.teiid.core.types.InputStreamFactory; +import org.teiid.core.types.Streamable; import org.teiid.core.types.InputStreamFactory.StreamFactoryReference; import org.teiid.core.util.ExternalizeUtil; import org.teiid.netty.handler.codec.serialization.CompactObjectInputStream; @@ -144,14 +142,7 @@ if (stream == null) { store = storageManager.createFileStore("temp-stream"); //$NON-NLS-1$ StreamFactoryReference sfr = streams.get(streamIndex); - store.setCleanupReference(sfr); - sfr.setStreamFactory(new InputStreamFactory() { - FileStore fs = store; - @Override - public InputStream getInputStream() throws IOException { - return new BufferedInputStream(fs.createInputStream(0)); - } - }); + sfr.setStreamFactory(new FileStoreInputStreamFactory(store, Streamable.ENCODING)); this.stream = new BufferedOutputStream(store.createOutputStream()); } if (dataLen == 0) { Modified: trunk/runtime/src/test/java/org/teiid/transport/TestCommSockets.java =================================================================== --- trunk/runtime/src/test/java/org/teiid/transport/TestCommSockets.java 2011-03-17 14:04:37 UTC (rev 2999) +++ trunk/runtime/src/test/java/org/teiid/transport/TestCommSockets.java 2011-03-17 14:25:04 UTC (rev 3000) @@ -41,6 +41,7 @@ import org.teiid.client.security.SessionToken; import org.teiid.client.util.ResultsFuture; import org.teiid.common.buffer.BufferManagerFactory; +import org.teiid.common.buffer.impl.MemoryStorageManager; import org.teiid.core.ComponentNotFoundException; import org.teiid.core.TeiidComponentException; import org.teiid.core.crypto.NullCryptor; @@ -62,6 +63,7 @@ SocketListener listener; private SocketServerConnectionFactory sscf; private InetSocketAddress addr; + private MemoryStorageManager storageManager; @Before public void setUp() { addr = new InetSocketAddress(0); @@ -125,8 +127,14 @@ SocketServerConnection conn = helpEstablishConnection(false); FakeService fs = conn.getService(FakeService.class); assertEquals(150, fs.lobMethod(new ByteArrayInputStream(new byte[100]), new StringReader(new String(new char[50])))); + assertEquals(2, storageManager.getCreated()); + assertEquals(2, storageManager.getRemoved()); assertEquals(0, fs.lobMethod(new ByteArrayInputStream(new byte[0]), new StringReader(new String(new char[0])))); + assertEquals(4, storageManager.getCreated()); + assertEquals(4, storageManager.getRemoved()); assertEquals((1 << 17) + 50, fs.lobMethod(new ByteArrayInputStream(new byte[1 << 17]), new StringReader(new String(new char[50])))); + assertEquals(6, storageManager.getCreated()); + assertEquals(6, storageManager.getRemoved()); } @Test public void testServerRemoteStreaming() throws Exception { @@ -173,7 +181,8 @@ }, null); server.registerClientService(FakeService.class, new TestSocketRemoting.FakeServiceImpl(), null); - listener = new SocketListener(addr.getPort(), addr.getAddress().getHostAddress(), 1024, 1024, 1, config, server, BufferManagerFactory.getStandaloneBufferManager()); + storageManager = new MemoryStorageManager(); + listener = new SocketListener(addr.getPort(), addr.getAddress().getHostAddress(), 1024, 1024, 1, config, server, storageManager); SocketListenerStats stats = listener.getStats(); assertEquals(0, stats.maxSockets);