[teiid-commits] teiid SVN: r3554 - in trunk: client/src/main/java/org/teiid/client and 7 other directories.

Author: shawkins Date: 2011-10-17 14:49:11 -0400 (Mon, 17 Oct 2011) New Revision: 3554 Added: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestLrfuEvictionQueue.java Modified: trunk/build/kits/jboss-container/deploy/teiid/teiid-jboss-beans.xml trunk/client/src/main/java/org/teiid/client/BatchSerializer.java trunk/client/src/main/java/org/teiid/client/ResultsMessage.java trunk/client/src/test/java/org/teiid/client/TestBatchSerializer.java trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/FunctionModifier.java trunk/documentation/admin-guide/src/main/docbook/en-US/content/performance.xml trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java trunk/engine/src/main/java/org/teiid/common/buffer/CacheKey.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferFrontedFileStoreCache.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/LrfuEvictionQueue.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java trunk/engine/src/test/java/org/teiid/query/processor/proc/TestProcedureProcessor.java Log: TEIID-1750 further refinement to serialization, improving queue precision, adding a doc entry and initial asynch cleaner for buffermanager Modified: trunk/build/kits/jboss-container/deploy/teiid/teiid-jboss-beans.xml =================================================================== --- trunk/build/kits/jboss-container/deploy/teiid/teiid-jboss-beans.xml 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/build/kits/jboss-container/deploy/teiid/teiid-jboss-beans.xml 2011-10-17 18:49:11 UTC (rev 3554) @@ -49,8 +49,8 @@ property if using older clients utilizing lobs. (default true) --> <property name="inlineLobs">true</property> <!-- Memory buffer space used by the buffer manager in MB. -1 determines the setting automatically from the maxReserveKB (default -1). - This value cannot be smaller than maxStorageObjectSize. --> - <property name="maxMemoryBufferSpace">-1</property> + This value cannot be smaller than maxStorageObjectSize.--> + <property name="memoryBufferSpace">-1</property> <!-- Set to true to hold the memory buffer off-heap. If true you must ensure that the VM can allocate that much direct memory (default false). --> <property name="memoryBufferOffHeap">false</property> <!-- The maximum size of a buffer managed object (typically a table page or a results batch) in bytes (default 8388608 or 8MB). Modified: trunk/client/src/main/java/org/teiid/client/BatchSerializer.java =================================================================== --- trunk/client/src/main/java/org/teiid/client/BatchSerializer.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/client/src/main/java/org/teiid/client/BatchSerializer.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -35,6 +35,7 @@ import java.util.Arrays; import java.util.Calendar; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.TimeZone; @@ -52,7 +53,60 @@ * @since 4.2 */ public class BatchSerializer { + + /* + * Public sharing part for the mapping between class and type in format of Map<class->Integer>. + */ + public static final int STRING = 0; + public static final int CHAR = 1; + public static final int BOOLEAN = 2; + public static final int BYTE = 3; + public static final int SHORT = 4; + public static final int INTEGER = 5; + public static final int LONG = 6; + public static final int BIGINTEGER = 7; + public static final int FLOAT = 8; + public static final int DOUBLE = 9; + public static final int BIGDECIMAL = 10; + public static final int DATE = 11; + public static final int TIME = 12; + public static final int TIMESTAMP = 13; + public static final int OBJECT = 14; + public static final int BLOB = 15; + public static final int CLOB = 16; + public static final int XML = 17; + public static final int NULL = 18; + + private static final Map<Class<?>, Integer> typeMap = new LinkedHashMap<Class<?>, Integer>(64); + private static final List<Class<?>> typeList; + static { + typeMap.put(DataTypeManager.DefaultDataClasses.STRING, STRING); + typeMap.put(DataTypeManager.DefaultDataClasses.CHAR, CHAR); + typeMap.put(DataTypeManager.DefaultDataClasses.BOOLEAN, BOOLEAN); + typeMap.put(DataTypeManager.DefaultDataClasses.BYTE, BYTE); + typeMap.put(DataTypeManager.DefaultDataClasses.SHORT, SHORT); + typeMap.put(DataTypeManager.DefaultDataClasses.INTEGER, INTEGER); + typeMap.put(DataTypeManager.DefaultDataClasses.LONG, LONG); + typeMap.put(DataTypeManager.DefaultDataClasses.BIG_INTEGER, BIGINTEGER); + typeMap.put(DataTypeManager.DefaultDataClasses.FLOAT, FLOAT); + typeMap.put(DataTypeManager.DefaultDataClasses.DOUBLE, DOUBLE); + typeMap.put(DataTypeManager.DefaultDataClasses.BIG_DECIMAL, BIGDECIMAL); + typeMap.put(DataTypeManager.DefaultDataClasses.DATE, DATE); + typeMap.put(DataTypeManager.DefaultDataClasses.TIME, TIME); + typeMap.put(DataTypeManager.DefaultDataClasses.TIMESTAMP, TIMESTAMP); + typeMap.put(DataTypeManager.DefaultDataClasses.OBJECT, OBJECT); + typeMap.put(DataTypeManager.DefaultDataClasses.BLOB, BLOB); + typeMap.put(DataTypeManager.DefaultDataClasses.CLOB, CLOB); + typeMap.put(DataTypeManager.DefaultDataClasses.XML, XML); + typeMap.put(DataTypeManager.DefaultDataClasses.NULL, NULL); + typeList = new ArrayList<Class<?>>(typeMap.keySet()); + } + + public static int getCode(Class<?> source) { + return typeMap.get(source).intValue(); + } + private BatchSerializer() {} // Uninstantiable private static ColumnSerializer defaultSerializer = new ColumnSerializer(); @@ -83,9 +137,49 @@ version1serializers.put(DataTypeManager.DefaultDataTypes.BLOB, new BlobColumnSerializer1()); version1serializers.put(DataTypeManager.DefaultDataTypes.XML, new XmlColumnSerializer1()); version1serializers.put(DataTypeManager.DefaultDataTypes.NULL, new NullColumnSerializer1()); - //TODO: do better with just an object column + version1serializers.put(DataTypeManager.DefaultDataTypes.OBJECT, new ObjectColumnSerializer1()); } + public static final class ObjectColumnSerializer1 extends ColumnSerializer { + + @Override + protected void writeObject(ObjectOutput out, Object obj) + throws IOException { + int code = getCode(obj.getClass()); + out.writeByte((byte)code); + if (code == BOOLEAN) { + if (Boolean.TRUE.equals(obj)) { + out.write((byte)1); + } else { + out.write((byte)0); + } + } else if (code != OBJECT) { + ColumnSerializer s = getSerializer(DataTypeManager.getDataTypeName(obj.getClass()), (byte)1); + s.writeObject(out, obj); + } else { + super.writeObject(out, obj); + } + } + + @Override + protected Object readObject(ObjectInput in) throws IOException, + ClassNotFoundException { + int code = in.readByte(); + if (code == BOOLEAN) { + if (in.readByte() == (byte)0) { + return Boolean.FALSE; + } + return Boolean.TRUE; + } + if (code != OBJECT) { + ColumnSerializer s = getSerializer(DataTypeManager.getDataTypeName(typeList.get(code)), (byte)1); + return s.readObject(in); + } + return super.readObject(in); + } + + } + private static final int MAX_UTF = 0xFFFF/3; //this is greater than the expected max length of Teiid Strings private static class StringColumnSerializer1 extends ColumnSerializer { Modified: trunk/client/src/main/java/org/teiid/client/ResultsMessage.java =================================================================== --- trunk/client/src/main/java/org/teiid/client/ResultsMessage.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/client/src/main/java/org/teiid/client/ResultsMessage.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -72,6 +72,8 @@ /** OPTION DEBUG log if OPTION DEBUG was used */ private String debugLog; + + private byte clientSerializationVersion; /** * Query plan annotations, if OPTION SHOWPLAN or OPTION PLANONLY was used: @@ -347,5 +349,13 @@ public boolean isUpdateResult() { return isUpdateResult; } + + public byte getClientSerializationVersion() { + return clientSerializationVersion; + } + + public void setClientSerializationVersion(byte clientSerializationVersion) { + this.clientSerializationVersion = clientSerializationVersion; + } } Modified: trunk/client/src/test/java/org/teiid/client/TestBatchSerializer.java =================================================================== --- trunk/client/src/test/java/org/teiid/client/TestBatchSerializer.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/client/src/test/java/org/teiid/client/TestBatchSerializer.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -76,7 +76,8 @@ DataTypeManager.DefaultDataTypes.SHORT, DataTypeManager.DefaultDataTypes.STRING, DataTypeManager.DefaultDataTypes.TIME, - DataTypeManager.DefaultDataTypes.TIMESTAMP + DataTypeManager.DefaultDataTypes.TIMESTAMP, + DataTypeManager.DefaultDataTypes.OBJECT, }; private static String sampleString(int length) { char[] chars = new char[length]; @@ -86,35 +87,10 @@ return new String(chars); } - private static List[] sampleBatch(int rows) { - List[] batch = new List[rows]; + private static List<?>[] sampleBatchWithNulls(int rows) { + List<?>[] batch = new List[rows]; for (int i = 0; i < rows; i++) { - java.util.Date d = new java.util.Date(); - Object[] data = { new BigDecimal("" + i), //$NON-NLS-1$ - new BigInteger(Integer.toString(i)), - (i%2 == 0) ? Boolean.FALSE: Boolean.TRUE, - new Byte((byte)i), - new Character((char)i), - TimestampWithTimezone.createDate(d), - new Double(i), - new Float(i), - new Integer(i), - new Long(i), - new Short((short)i), - sampleString(i), - TimestampWithTimezone.createTime(d), - TimestampWithTimezone.createTimestamp(d) - }; - batch[i] = Arrays.asList(data); - } - return batch; - } - - private static List[] sampleBatchWithNulls(int rows) { - List[] batch = new List[rows]; - - for (int i = 0; i < rows; i++) { java.util.Date d = new java.util.Date(); int mod = i%14; Object[] data = { (mod == 0) ? null : new BigDecimal("" + i), //$NON-NLS-1$ @@ -130,7 +106,8 @@ (mod == 10) ? null : new Short((short)i), (mod == 11) ? null : sampleString(i), (mod == 12) ? null : TimestampWithTimezone.createTime(d), - (mod == 13) ? null : TimestampWithTimezone.createTimestamp(d) + (mod == 13) ? null : TimestampWithTimezone.createTimestamp(d), + (mod == 14) ? null : TimestampWithTimezone.createTimestamp(d), }; batch[i] = Arrays.asList(data); } @@ -140,23 +117,16 @@ public void testSerializeBasicTypes() throws Exception { // The number 8 is important here because boolean isNull information is packed into bytes, // so we want to make sure the boundary cases are handled correctly - helpTestSerialization(sampleBatchTypes, sampleBatch(1)); // Less than 8 rows - helpTestSerialization(sampleBatchTypes, sampleBatch(8)); // Exactly 8 rows - helpTestSerialization(sampleBatchTypes, sampleBatch(17)); // More than 8 rows, but not a multiple of 8 - helpTestSerialization(sampleBatchTypes, sampleBatch(120)); // A multiple of 8 rows - helpTestSerialization(sampleBatchTypes, sampleBatch(833)); // A bunch of rows. This should also test large strings + helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(1)); // Less than 8 rows + helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(8)); // Exactly 8 rows + helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(17)); // More than 8 rows, but not a multiple of 8 + helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(120)); // A multiple of 8 rows + helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(833)); // A bunch of rows. This should also test large strings + helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(4096)); // A bunch of rows. This should also test large strings } - public void testSerializeBasicTypesWithNulls() throws Exception { - helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(1)); - helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(8)); - helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(17)); - helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(120)); - helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(833)); - } - public void testSerializeLargeStrings() throws Exception { - List row = Arrays.asList(new Object[] {sampleString(66666)}); + List<?> row = Arrays.asList(new Object[] {sampleString(66666)}); helpTestSerialization(new String[] {DataTypeManager.DefaultDataTypes.STRING}, new List[] {row}); } Modified: trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/FunctionModifier.java =================================================================== --- trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/FunctionModifier.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/FunctionModifier.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -24,13 +24,11 @@ */ package org.teiid.translator.jdbc; -import java.util.HashMap; import java.util.List; -import java.util.Map; +import org.teiid.client.BatchSerializer; import org.teiid.language.Function; import org.teiid.language.LanguageObject; -import org.teiid.translator.TypeFacility; /** @@ -43,52 +41,28 @@ /* * Public sharing part for the mapping between class and type in format of Map<class->Integer>. */ - public static final int STRING = 0; - public static final int CHAR = 1; - public static final int BOOLEAN = 2; - public static final int BYTE = 3; - public static final int SHORT = 4; - public static final int INTEGER = 5; - public static final int LONG = 6; - public static final int BIGINTEGER = 7; - public static final int FLOAT = 8; - public static final int DOUBLE = 9; - public static final int BIGDECIMAL = 10; - public static final int DATE = 11; - public static final int TIME = 12; - public static final int TIMESTAMP = 13; - public static final int OBJECT = 14; - public static final int BLOB = 15; - public static final int CLOB = 16; - public static final int XML = 17; - public static final int NULL = 18; + public static final int STRING = BatchSerializer.STRING; + public static final int CHAR = BatchSerializer.CHAR; + public static final int BOOLEAN = BatchSerializer.BOOLEAN; + public static final int BYTE = BatchSerializer.BYTE; + public static final int SHORT = BatchSerializer.SHORT; + public static final int INTEGER = BatchSerializer.INTEGER; + public static final int LONG = BatchSerializer.LONG; + public static final int BIGINTEGER = BatchSerializer.BIGINTEGER; + public static final int FLOAT = BatchSerializer.FLOAT; + public static final int DOUBLE = BatchSerializer.DOUBLE; + public static final int BIGDECIMAL = BatchSerializer.BIGDECIMAL; + public static final int DATE = BatchSerializer.DATE; + public static final int TIME = BatchSerializer.TIME; + public static final int TIMESTAMP = BatchSerializer.TIMESTAMP; + public static final int OBJECT = BatchSerializer.OBJECT; + public static final int BLOB = BatchSerializer.BLOB; + public static final int CLOB = BatchSerializer.CLOB; + public static final int XML = BatchSerializer.XML; + public static final int NULL = BatchSerializer.NULL; - private static final Map<Class<?>, Integer> typeMap = new HashMap<Class<?>, Integer>(); - - static { - typeMap.put(TypeFacility.RUNTIME_TYPES.STRING, STRING); - typeMap.put(TypeFacility.RUNTIME_TYPES.CHAR, CHAR); - typeMap.put(TypeFacility.RUNTIME_TYPES.BOOLEAN, BOOLEAN); - typeMap.put(TypeFacility.RUNTIME_TYPES.BYTE, BYTE); - typeMap.put(TypeFacility.RUNTIME_TYPES.SHORT, SHORT); - typeMap.put(TypeFacility.RUNTIME_TYPES.INTEGER, INTEGER); - typeMap.put(TypeFacility.RUNTIME_TYPES.LONG, LONG); - typeMap.put(TypeFacility.RUNTIME_TYPES.BIG_INTEGER, BIGINTEGER); - typeMap.put(TypeFacility.RUNTIME_TYPES.FLOAT, FLOAT); - typeMap.put(TypeFacility.RUNTIME_TYPES.DOUBLE, DOUBLE); - typeMap.put(TypeFacility.RUNTIME_TYPES.BIG_DECIMAL, BIGDECIMAL); - typeMap.put(TypeFacility.RUNTIME_TYPES.DATE, DATE); - typeMap.put(TypeFacility.RUNTIME_TYPES.TIME, TIME); - typeMap.put(TypeFacility.RUNTIME_TYPES.TIMESTAMP, TIMESTAMP); - typeMap.put(TypeFacility.RUNTIME_TYPES.OBJECT, OBJECT); - typeMap.put(TypeFacility.RUNTIME_TYPES.BLOB, BLOB); - typeMap.put(TypeFacility.RUNTIME_TYPES.CLOB, CLOB); - typeMap.put(TypeFacility.RUNTIME_TYPES.XML, XML); - typeMap.put(TypeFacility.RUNTIME_TYPES.NULL, NULL); - } - public static int getCode(Class<?> source) { - return typeMap.get(source).intValue(); + return BatchSerializer.getCode(source); } /** Modified: trunk/documentation/admin-guide/src/main/docbook/en-US/content/performance.xml =================================================================== --- trunk/documentation/admin-guide/src/main/docbook/en-US/content/performance.xml 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/documentation/admin-guide/src/main/docbook/en-US/content/performance.xml 2011-10-17 18:49:11 UTC (rev 3554) @@ -63,6 +63,38 @@ if your installation makes use of internal materialization, makes heavy use of SQL/XML, or processes large row counts. </para> + <section> + <title>Big Data/Memory</title> + <para>Usage of extremely large VM sizes and or datasets requires additional considerations. + Teiid has a non-negligible amount of overhead per batch/table page on the order of 100-200 bytes. Depending on the data types involved each + full batch/table page will represent between 64 and 4096 rows. If you are dealing with datasets with billions of rows and you run into OutOfMemory issues, consider increasing the processor + batch size in the &jboss-beans; file to force the allocation of larger batches and table pages. If the processor batch size is increased and/or you are dealing with extremely wide result sets (several hundred columns), + then the default setting of 8MB for the maxStorageObjectSize in the &jboss-beans; file may be too low. The sizing for maxStorageObjectSize is terms of serialized size, which will be much + closer to the raw data size then the Java memory footprint estimation used for maxReservedKB. + maxStorageObjectSize should not be set too large relative to memoryBufferSpace since it will reduce the performance of the memory buffer. The memory buffer supports only 1 concurrent writer for each maxStorageObjectSize of the memoryBufferSpace. + </para> + <note><para>Teiid temporary tables (also used for internal materialization) can only support 2^31-1 rows per table.</para></note> + <para> + The memoryBufferSpace setting controls the amount of on or off heap memory allocated as byte buffers for use by the Teiid buffer manager. This setting defaults to -1, which automatically determines a setting based upon whether it is on or off heap and the value for maxReserveKB. + </para> + <para>You can take advantage of the buffer manager memory buffer to access system memory without allocating it to the heap. Setting memoryBufferOffHeap to true in &jboss-beans; will allocate the + Teiid memory buffer off heap. Depending on whether your installation is dedicated to Teiid and the amount of system memory available, this may be perferable to on-heap allocation. The primary + benefit is additional memory usage for Teiid without additional garbage collection tuning. This becomes especially important in situations where more than 32GB of memory is desired for the VM. + Note that when using off-heap allocation, the memory must still be available to the java process and that setting the value of memoryBufferSpace too high may cause the VM to swap rather than reside in memory. + With large off-heap buffer sizes (greater than several gigabytes) you may also need to adjust VM settings. For Sun VMs the relevant VM settings are MaxDirectMemorySize and UseLargePages. For example adding: + <programlisting>-XX:MaxDirectMemorySize=12g -XX:+UseLargePages</programlisting> + to the VM process arguments would allow for an effective allocation of approximately an 11GB Teiid memory buffer (the memoryBufferSpace setting) accounting for any additional direct memory that may be needed + by the AS or applications running in the AS. + </section> + <section> + <title>Disk Usage</title> + <para>For table page and result batches the buffer manager will we a limited number of files that are dedicated to a particular storage size. + However, as mentioned in the installation, creation of Teiid lob values (for example through SQL/XML) will typically create one file per lob once the lob exceeds the allowable + in memory size of 8KB. In heavy usage scenarios, consider pointing the buffer directory on a partition that is routinely defragmented. + By default Teiid will use up to 50GB of disk space. This is tracked in terms of the number of bytes written by Teiid. For large data sets, you may need to increase the + maxBufferSpace setting in the &jboss-beans; file. + </para> + </section> </section> <section> <title>Threading</title> Modified: trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -22,6 +22,7 @@ package org.teiid.common.buffer; +import java.lang.ref.WeakReference; import java.util.Collection; import org.teiid.core.TeiidComponentException; @@ -74,7 +75,7 @@ * @return * @throws TeiidComponentException */ - CacheEntry get(T lock, Long oid, Serializer<?> serializer) throws TeiidComponentException; + CacheEntry get(T lock, Long oid, WeakReference<? extends Serializer<?>> ref) throws TeiidComponentException; /** * Adds an entry to the cache. Modified: trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -27,36 +27,32 @@ public class CacheEntry extends BaseCacheEntry { private boolean persistent; private Object object; - private int sizeEstimate; + private final int sizeEstimate; private WeakReference<? extends Serializer<?>> serializer; public CacheEntry(Long oid) { - super(new CacheKey(oid, 0, 0)); + this(new CacheKey(oid, 0, 0), 0, null, null, false); } - public CacheEntry(CacheKey key) { + public CacheEntry(CacheKey key, int sizeEstimate, Object object, WeakReference<? extends Serializer<?>> serializer, boolean persistent) { super(key); + this.sizeEstimate = sizeEstimate; + this.object = object; + this.serializer = serializer; + this.persistent = persistent; } public int getSizeEstimate() { return sizeEstimate; } - public void setSizeEstimate(int sizeEstimate) { - this.sizeEstimate = sizeEstimate; - } - public Object nullOut() { Object result = getObject(); - setObject(null); - setSerializer(null); + this.object = null; + this.serializer = null; return result; } - public void setObject(Object object) { - this.object = object; - } - public Object getObject() { return object; } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/CacheKey.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/CacheKey.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/engine/src/main/java/org/teiid/common/buffer/CacheKey.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -25,10 +25,10 @@ public class CacheKey implements Comparable<CacheKey> { private Long id; - protected float lastAccess; + protected int lastAccess; protected float orderingValue; - public CacheKey(Long id, float lastAccess, float orderingValue) { + public CacheKey(Long id, int lastAccess, float orderingValue) { this.id = id; this.lastAccess = lastAccess; this.orderingValue = orderingValue; @@ -59,7 +59,7 @@ return this.id.equals(((CacheKey)obj).getId()); } - public float getLastAccess() { + public int getLastAccess() { return lastAccess; } @@ -71,7 +71,7 @@ public int compareTo(CacheKey o) { int result = (int) Math.signum(orderingValue - o.orderingValue); if (result == 0) { - result = (int)Math.signum(lastAccess - o.lastAccess); + result = (int)Math.signum((lastAccess&0xffffffffl) - (o.lastAccess&0xffffffffl)); if (result == 0) { return Long.signum(id - o.id); } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferFrontedFileStoreCache.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferFrontedFileStoreCache.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferFrontedFileStoreCache.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -26,6 +26,7 @@ import java.io.InputStream; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; +import java.lang.ref.WeakReference; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; @@ -383,13 +384,13 @@ public void initialize() throws TeiidComponentException { storageManager.initialize(); memoryBufferSpace = Math.max(memoryBufferSpace, maxStorageObjectSize); - blocks = (int) Math.min(Integer.MAX_VALUE, (memoryBufferSpace>>LOG_BLOCK_SIZE)); + blocks = (int) Math.min(Integer.MAX_VALUE, (memoryBufferSpace>>LOG_BLOCK_SIZE)*ADDRESSES_PER_BLOCK/(ADDRESSES_PER_BLOCK+1)); inodesInuse = new ConcurrentBitSet(blocks+1, BufferManagerImpl.CONCURRENCY_LEVEL); blocksInuse = new ConcurrentBitSet(blocks, BufferManagerImpl.CONCURRENCY_LEVEL); this.blockByteBuffer = new BlockByteBuffer(30, blocks, LOG_BLOCK_SIZE, direct); //ensure that we'll run out of blocks first this.inodeByteBuffer = new BlockByteBuffer(30, blocks+1, LOG_INODE_SIZE, direct); - memoryWritePermits = new Semaphore(Math.max(1, (int)Math.min(memoryBufferSpace/maxStorageObjectSize, Integer.MAX_VALUE))); + memoryWritePermits = new Semaphore(Math.max(1, (int)Math.min((((long)blocks)<<LOG_BLOCK_SIZE)/maxStorageObjectSize, Integer.MAX_VALUE))); maxMemoryBlocks = Math.min(MAX_DOUBLE_INDIRECT, maxStorageObjectSize>>LOG_BLOCK_SIZE); //try to maintain enough freespace so that writers don't block in cleaning cleaningThreshold = Math.min(maxMemoryBlocks<<4, blocks>>1); @@ -547,10 +548,16 @@ } @Override - public CacheEntry get(PhysicalInfo info, Long oid, Serializer<?> serializer) throws TeiidComponentException { + public CacheEntry get(PhysicalInfo info, Long oid, + WeakReference<? extends Serializer<?>> ref) + throws TeiidComponentException { if (info == null) { return null; } + Serializer<?> serializer = ref.get(); + if (serializer == null) { + return null; + } long currentTime = readAttempts.incrementAndGet(); InputStream is = null; boolean inStorage = false; @@ -604,11 +611,8 @@ } } } - CacheEntry ce = new CacheEntry(new CacheKey(oid, 1, 1)); ObjectInputStream ois = new ObjectInputStream(is); - ce.setSizeEstimate(ois.readInt()); - ce.setObject(serializer.deserialize(ois)); - ce.setPersistent(true); + CacheEntry ce = new CacheEntry(new CacheKey(oid, 1, 1), ois.readInt(), serializer.deserialize(ois), ref, true); return ce; } catch(IOException e) { throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_reading", oid)); //$NON-NLS-1$ @@ -728,6 +732,9 @@ BlockInputStream is = new BlockInputStream(bm, memoryBlockCount); BlockStore blockStore = sizeBasedStores[sizeIndex]; block = getAndSetNextClearBit(blockStore); + if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) { + LogManager.logDetail(LogConstants.CTX_DQP, "Allocating storage data block", info.block, "of size", blockStore.blockSize, "to", info); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-1$ + } FileStore fs = blockStore.stores[block/blockStore.blocksInUse.getBitsPerSegment()]; long blockOffset = (block%blockStore.blocksInUse.getBitsPerSegment())*blockStore.blockSize; byte[] b = new byte[BLOCK_SIZE]; @@ -763,6 +770,9 @@ } else { BlockStore blockStore = sizeBasedStores[info.sizeIndex]; blockStore.blocksInUse.clear(info.block); + if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) { + LogManager.logDetail(LogConstants.CTX_DQP, "Freed storage data block", info.block, "of size", blockStore.blockSize); //$NON-NLS-1$ //$NON-NLS-2$ + } info.block = EMPTY_ADDRESS; } } @@ -915,6 +925,7 @@ public void setSize(int size) { this.memoryBlockCount = (size>>BufferFrontedFileStoreCache.LOG_BLOCK_SIZE) + ((size&BufferFrontedFileStoreCache.BLOCK_MASK)>0?1:0); int blocks = memoryBlockCount; + this.sizeIndex = 0; while (blocks >= 1) { this.sizeIndex++; blocks>>=2; Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -36,6 +36,8 @@ import java.util.List; import java.util.ListIterator; import java.util.Map; +import java.util.Timer; +import java.util.TimerTask; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -82,6 +84,42 @@ */ public class BufferManagerImpl implements BufferManager, StorageManager { + private static final class Cleaner extends TimerTask { + WeakReference<BufferManagerImpl> bufferRef; + + public Cleaner(BufferManagerImpl bufferManagerImpl) { + this.bufferRef = new WeakReference<BufferManagerImpl>(bufferManagerImpl); + } + + @Override + public void run() { + while (true) { + BufferManagerImpl impl = this.bufferRef.get(); + if (impl == null) { + this.cancel(); + return; + } + if (impl.reserveBatchKB.get() < impl.maxReserveKB.get()*.9 || impl.activeBatchKB.get() < (impl.maxReserveKB.get()>>3)) { + CacheEntry entry = impl.evictionQueue.firstEntry(false); + if (entry == null) { + return; + } + //we aren't holding too many memory entries, ensure that + //entries aren't old + int lastAccess = 0x1fffffff&entry.getKey().getLastAccess(); + int currentTime = 0x1fffffff&(int)impl.readAttempts.get(); + if (lastAccess > currentTime) { + currentTime += 1<<29; + } + if (currentTime - lastAccess < 1<<28) { + return; + } + } + impl.doEvictions(0, false); + } + } + } + final class BatchManagerImpl implements BatchManager, Serializer<List<? extends List<?>>> { final Long id; SizeUtility sizeUtility; @@ -150,10 +188,7 @@ } } CacheKey key = new CacheKey(oid, 0, old!=null?old.getKey().getOrderingValue():0); - CacheEntry ce = new CacheEntry(key); - ce.setObject(batch); - ce.setSizeEstimate(sizeEstimate); - ce.setSerializer(this.ref); + CacheEntry ce = new CacheEntry(key, sizeEstimate, batch, this.ref, false); if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.TRACE)) { LogManager.logTrace(LogConstants.CTX_BUFFER_MGR, "Add batch to BufferManager", ce.getId(), "with size estimate", ce.getSizeEstimate()); //$NON-NLS-1$ //$NON-NLS-2$ } @@ -226,15 +261,13 @@ if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, id, id, "reading batch", batch, "from storage, total reads:", count); //$NON-NLS-1$ //$NON-NLS-2$ } - ce = cache.get(o, batch, this); + ce = cache.get(o, batch, this.ref); if (ce == null) { throw new AssertionError("Batch not found in storage " + batch); //$NON-NLS-1$ } if (!retain) { cache.remove(this.id, batch); } - ce.setSerializer(this.ref); - ce.setPersistent(true); if (retain) { addMemoryEntry(ce); } @@ -284,8 +317,8 @@ //set to acceptable defaults for testing private int maxProcessingKB = 1 << 11; private Integer maxProcessingKBOrig; - private AtomicInteger maxReserveKB = new AtomicInteger(1 << 18); - private volatile int reserveBatchKB; + AtomicInteger maxReserveKB = new AtomicInteger(1 << 18); + AtomicInteger reserveBatchKB = new AtomicInteger(); private int maxActivePlans = DQPConfiguration.DEFAULT_MAX_ACTIVE_PLANS; //used as a hint to set the reserveBatchKB private boolean useWeakReferences = true; private boolean inlineLobs = true; @@ -295,7 +328,7 @@ private ReentrantLock lock = new ReentrantLock(true); private Condition batchesFreed = lock.newCondition(); - private AtomicInteger activeBatchKB = new AtomicInteger(); + AtomicInteger activeBatchKB = new AtomicInteger(); private AtomicLong readAttempts = new AtomicLong(); //TODO: consider the size estimate in the weighting function @@ -311,6 +344,7 @@ if (size() > maxSoftReferences) { BatchSoftReference bsr = eldest.getValue(); maxReserveKB.addAndGet(bsr.sizeEstimate); + reserveBatchKB.addAndGet(bsr.sizeEstimate); bsr.clear(); return true; } @@ -329,6 +363,12 @@ private AtomicLong writeCount = new AtomicLong(); private AtomicLong referenceHit = new AtomicLong(); + private static final Timer timer = new Timer("BufferManager Cleaner", true); //$NON-NLS-1$ + + public BufferManagerImpl() { + //timer.schedule(new Cleaner(this), 30000, 30000); + } + public long getBatchesAdded() { return batchAdded.get(); } @@ -470,12 +510,15 @@ public void setMaxReserveKB(int maxReserveBatchKB) { this.maxReserveKB.set(maxReserveBatchKB); + if (maxReserveBatchKB > -1) { + this.reserveBatchKB.set(maxReserveBatchKB); + } } @Override public void initialize() throws TeiidComponentException { int maxMemory = (int)Math.min(Runtime.getRuntime().maxMemory() / 1024, Integer.MAX_VALUE); - maxMemory -= 300 * 1024; //assume 300 megs of overhead for the AS/system stuff + maxMemory = Math.max(0, maxMemory - 300 * 1024); //assume 300 megs of overhead for the AS/system stuff if (getMaxReserveKB() < 0) { this.setMaxReserveKB(0); int one_gig = 1024 * 1024; @@ -485,7 +528,7 @@ } this.maxReserveKB.addAndGet(((int)Math.max(0, Math.min(one_gig, maxMemory) * .5))); } - this.reserveBatchKB = this.getMaxReserveKB(); + this.reserveBatchKB.set(this.getMaxReserveKB()); if (this.maxProcessingKBOrig == null) { //store the config value so that we can be reinitialized (this is not a clean approach) this.maxProcessingKBOrig = this.maxProcessingKB; @@ -513,7 +556,7 @@ } lock.lock(); try { - this.reserveBatchKB += count; + this.reserveBatchKB.addAndGet(count); batchesFreed.signalAll(); } finally { lock.unlock(); @@ -530,7 +573,7 @@ if (mode == BufferReserveMode.WAIT) { //don't wait for more than is available int waitCount = Math.min(count, this.getMaxReserveKB()); - while (waitCount > 0 && waitCount > this.reserveBatchKB) { + while (waitCount > 0 && waitCount > this.reserveBatchKB.get()) { try { batchesFreed.await(100, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { @@ -539,12 +582,12 @@ waitCount /= 2; } } - if (this.reserveBatchKB >= count || mode == BufferReserveMode.FORCE) { - this.reserveBatchKB -= count; + if (this.reserveBatchKB.get() >= count || mode == BufferReserveMode.FORCE) { + this.reserveBatchKB.addAndGet(-count); return count; } - int result = Math.max(0, this.reserveBatchKB); - this.reserveBatchKB -= result; + int result = Math.max(0, this.reserveBatchKB.get()); + this.reserveBatchKB.addAndGet(-result); return result; } finally { lock.unlock(); @@ -554,8 +597,9 @@ void persistBatchReferences() { int activeBatch = activeBatchKB.get(); - if (activeBatch <= reserveBatchKB) { - int memoryCount = activeBatch + getMaxReserveKB() - reserveBatchKB; + int reserveBatch = reserveBatchKB.get(); + if (activeBatch <= reserveBatch) { + int memoryCount = activeBatch + getMaxReserveKB() - reserveBatch; if (DataTypeManager.isValueCacheEnabled()) { if (memoryCount < getMaxReserveKB() / 8) { DataTypeManager.setValueCacheEnabled(false); @@ -565,9 +609,13 @@ } return; } - int maxToFree = Math.max(maxProcessingKB>>1, reserveBatchKB>>3); + int maxToFree = Math.max(maxProcessingKB>>1, reserveBatch>>3); + doEvictions(maxToFree, true); + } + + void doEvictions(int maxToFree, boolean checkActiveBatch) { int freed = 0; - while (freed <= maxToFree && activeBatchKB.get() > reserveBatchKB * .8) { + while (freed <= maxToFree && (!checkActiveBatch || activeBatchKB.get() > reserveBatchKB.get() * .8)) { CacheEntry ce = evictionQueue.firstEntry(true); if (ce == null) { break; @@ -623,6 +671,7 @@ BatchSoftReference ref = new BatchSoftReference(ce, SOFT_QUEUE, ce.getSizeEstimate()/2); softCache.put(ce.getId(), ref); maxReserveKB.addAndGet(- ce.getSizeEstimate()/2); + reserveBatchKB.addAndGet(- ce.getSizeEstimate()/2); } /** @@ -658,6 +707,7 @@ ce = bsr.get(); if (ce != null) { maxReserveKB.addAndGet(bsr.sizeEstimate); + reserveBatchKB.addAndGet(bsr.sizeEstimate); } } } else if (useWeakReferences) { @@ -727,6 +777,7 @@ } softCache.remove(ref.key); maxReserveKB.addAndGet(ref.sizeEstimate); + reserveBatchKB.addAndGet(ref.sizeEstimate); ref.clear(); } } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/LrfuEvictionQueue.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/LrfuEvictionQueue.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/LrfuEvictionQueue.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -47,10 +47,13 @@ protected AtomicLong clock; //combined recency/frequency lamda value between 0 and 1 lower -> LFU, higher -> LRU //TODO: adaptively adjust this value. more hits should move closer to lru - protected float crfLamda = .0002f; + protected double crfLamda; + protected double inverseCrfLamda = 1 - crfLamda; + protected int maxInterval; public LrfuEvictionQueue(AtomicLong clock) { this.clock = clock; + setCrfLamda(.0002); } public boolean remove(V value) { @@ -85,33 +88,47 @@ protected void recordAccess(V value, boolean initial) { assert Thread.holdsLock(value); CacheKey key = value.getKey(); - float lastAccess = key.getLastAccess(); - float currentClock = clock.get(); + int lastAccess = key.getLastAccess(); + long currentClock = clock.get(); if (initial && lastAccess == 0) { return; //we just want to timestamp this as created and not give it an ordering value } float orderingValue = key.getOrderingValue(); orderingValue = computeNextOrderingValue(currentClock, lastAccess, orderingValue); - value.setKey(new CacheKey(key.getId(), currentClock, orderingValue)); + value.setKey(new CacheKey(key.getId(), (int)currentClock, orderingValue)); } - float computeNextOrderingValue(float currentTime, - float lastAccess, float orderingValue) { + float computeNextOrderingValue(long currentTime, + int lastAccess, float orderingValue) { + long longLastAccess = lastAccess&0xffffffffl; + currentTime &= 0xffffffffl; + if (longLastAccess > currentTime) { + currentTime += (1l<<32); + } + long delta = currentTime - longLastAccess; orderingValue = (float) (//Frequency component - orderingValue*Math.pow(1-crfLamda, currentTime - lastAccess) + (delta>maxInterval?0:orderingValue*Math.pow(inverseCrfLamda, delta)) //recency component + Math.pow(currentTime, crfLamda)); return orderingValue; } - public float getCrfLamda() { + public double getCrfLamda() { return crfLamda; } - public void setCrfLamda(float crfLamda) { + public void setCrfLamda(double crfLamda) { this.crfLamda = crfLamda; + this.inverseCrfLamda = 1 - crfLamda; + int i = 0; + for (; i < 30; i++) { + if ((float)Math.pow(inverseCrfLamda, 1<<i) == 0) { + break; + } + } + this.maxInterval = i-1; } } 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-10-17 03:12:03 UTC (rev 3553) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -22,6 +22,7 @@ package org.teiid.common.buffer.impl; +import java.lang.ref.WeakReference; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; @@ -143,11 +144,12 @@ } @Override - public CacheEntry get(Long lock, Long id, Serializer<?> serializer) + public CacheEntry get(Long lock, Long oid, + WeakReference<? extends Serializer<?>> ref) throws TeiidComponentException { - Map<Long, CacheEntry> group = groups.get(serializer.getId()); + Map<Long, CacheEntry> group = groups.get(ref.get().getId()); if (group != null) { - return group.get(id); + return group.get(oid); } return null; } Added: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestLrfuEvictionQueue.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestLrfuEvictionQueue.java (rev 0) +++ trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestLrfuEvictionQueue.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -0,0 +1,44 @@ +/* + * 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.impl; + +import static org.junit.Assert.*; + +import java.util.concurrent.atomic.AtomicLong; + +import org.junit.Test; +import org.teiid.common.buffer.BaseCacheEntry; + +public class TestLrfuEvictionQueue { + + @Test public void testPrecision() { + LrfuEvictionQueue<?> q = new LrfuEvictionQueue<BaseCacheEntry>(new AtomicLong()); + float value = 0; + for (long i = Integer.MAX_VALUE; i < 10l + Integer.MAX_VALUE; i++) { + float valueNext = q.computeNextOrderingValue(i, (int)i-1, value); + assertTrue(valueNext > value); + value = valueNext; + } + } + +} Property changes on: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestLrfuEvictionQueue.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/test/java/org/teiid/query/processor/proc/TestProcedureProcessor.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/proc/TestProcedureProcessor.java 2011-10-17 03:12:03 UTC (rev 3553) +++ trunk/engine/src/test/java/org/teiid/query/processor/proc/TestProcedureProcessor.java 2011-10-17 18:49:11 UTC (rev 3554) @@ -2529,4 +2529,5 @@ } private static final boolean DEBUG = false; + }