5:37 p.m.
Author: shawkins
Date: 2011-10-20 11:37:18 -0400 (Thu, 20 Oct 2011)
New Revision: 3573
Modified:
trunk/engine/src/main/java/org/teiid/common/buffer/impl/BlockInputStream.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/ConcurrentBitSet.java
trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStorageManager.java
trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBufferFrontedFileStoreCache.java
trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStorageManager.java
Log:
TEIID-1750 fixing a bug with positional filestore writes and fixing size indexing.
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/BlockInputStream.java
===================================================================
---
trunk/engine/src/main/java/org/teiid/common/buffer/impl/BlockInputStream.java 2011-10-20
14:41:28 UTC (rev 3572)
+++
trunk/engine/src/main/java/org/teiid/common/buffer/impl/BlockInputStream.java 2011-10-20
15:37:18 UTC (rev 3573)
@@ -34,12 +34,10 @@
int blockIndex;
ByteBuffer buf;
boolean done;
- private final boolean threadSafe;
- BlockInputStream(BlockManager manager, int blockCount, boolean threadSafe) {
+ BlockInputStream(BlockManager manager, int blockCount) {
this.manager = manager;
this.maxBlock = blockCount;
- this.threadSafe = threadSafe;
}
@Override
@@ -58,9 +56,6 @@
return;
}
buf = manager.getBlock(blockIndex++);
- if (threadSafe) {
- buf = buf.duplicate();
- }
}
}
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-20
14:41:28 UTC (rev 3572)
+++
trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferFrontedFileStoreCache.java 2011-10-20
15:37:18 UTC (rev 3573)
@@ -405,12 +405,9 @@
List<BlockStore> stores = new ArrayList<BlockStore>();
int size = BLOCK_SIZE;
do {
- if ((size>>1) >= maxStorageObjectSize) {
- size>>=1; //adjust the last block size if needed
- }
- stores.add(new BlockStore(this.storageManager, size, 15,
BufferManagerImpl.CONCURRENCY_LEVEL>>2));
- size <<=2;
- } while (size>>2 < maxStorageObjectSize);
+ stores.add(new BlockStore(this.storageManager, size, 30,
BufferManagerImpl.CONCURRENCY_LEVEL));
+ size <<=1;
+ } while ((size>>1) < maxStorageObjectSize);
this.sizeBasedStores = stores.toArray(new BlockStore[stores.size()]);
}
@@ -503,7 +500,6 @@
}
} catch (Throwable e) {
if (e == PhysicalInfo.sizeChanged) {
- //System.out.println("size changed " + info.inode + " " +
info.block + " " + info);
//entries are mutable after adding, the original should be removed shortly so just
ignore
LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Object "+ entry.getId()
+" changed size since first persistence, keeping the original."); //$NON-NLS-1$
//$NON-NLS-2$
} else {
@@ -587,7 +583,7 @@
LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Getting object at
inode", info.inode, serializer.getId(), oid); //$NON-NLS-1$
}
BlockManager manager = getBlockManager(serializer.getId(), oid, info.inode);
- is = new BlockInputStream(manager, info.memoryBlockCount, info.evicting);
+ is = new BlockInputStream(manager, info.memoryBlockCount);
} else if (info.block != EMPTY_ADDRESS) {
memoryBufferEntries.recordAccess(info);
storageReads.incrementAndGet();
@@ -720,7 +716,7 @@
try {
if (demote && block == EMPTY_ADDRESS) {
storageWrites.getAndIncrement();
- BlockInputStream is = new BlockInputStream(bm, memoryBlockCount, false); //we know
this can always be single threaded
+ BlockInputStream is = new BlockInputStream(bm, memoryBlockCount);
BlockStore blockStore = sizeBasedStores[sizeIndex];
block = getAndSetNextClearBit(blockStore);
if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR,
MessageLevel.DETAIL)) {
@@ -930,9 +926,9 @@
return; //no changes
}
this.memoryBlockCount = newMemoryBlockCount;
- while (newMemoryBlockCount >= 1) {
+ while (newMemoryBlockCount > 1) {
this.sizeIndex++;
- newMemoryBlockCount>>=2;
+ newMemoryBlockCount = (newMemoryBlockCount>>1) +
((newMemoryBlockCount&0x01)==0?0:1);
}
}
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-20
14:41:28 UTC (rev 3572)
+++
trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2011-10-20
15:37:18 UTC (rev 3573)
@@ -323,7 +323,7 @@
}
}
- static final int CONCURRENCY_LEVEL = 32; //TODO: make this configurable
+ static final int CONCURRENCY_LEVEL = 32; //TODO: make this configurable since it is
roughly the same as max active plans
private static final int TARGET_BYTES_PER_ROW = 1 << 11; //2k bytes per row
private static ReferenceQueue<CacheEntry> SOFT_QUEUE = new
ReferenceQueue<CacheEntry>();
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/ConcurrentBitSet.java
===================================================================
---
trunk/engine/src/main/java/org/teiid/common/buffer/impl/ConcurrentBitSet.java 2011-10-20
14:41:28 UTC (rev 3572)
+++
trunk/engine/src/main/java/org/teiid/common/buffer/impl/ConcurrentBitSet.java 2011-10-20
15:37:18 UTC (rev 3573)
@@ -180,7 +180,7 @@
}
/**
- * Set to try to always allocate against the first available block in a segment.
+ * Set to always allocate against the first available block in a segment.
* @param compact
*/
public void setCompact(boolean compact) {
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStorageManager.java
===================================================================
---
trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStorageManager.java 2011-10-20
14:41:28 UTC (rev 3572)
+++
trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStorageManager.java 2011-10-20
15:37:18 UTC (rev 3573)
@@ -133,11 +133,10 @@
if (bytesUsed > 0) {
long used = usedBufferSpace.addAndGet(bytesUsed);
if (used > maxBufferSpace) {
- usedBufferSpace.addAndGet(-bytesUsed);
//TODO: trigger a compaction before this is thrown
throw new
IOException(QueryPlugin.Util.getString("FileStoreageManager.space_exhausted",
maxBufferSpace)); //$NON-NLS-1$
}
- fileAccess.setLength(bytesUsed);
+ fileAccess.setLength(newLength);
bytesUsed = 0;
}
fileAccess.seek(fileOffset);
Modified:
trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBufferFrontedFileStoreCache.java
===================================================================
---
trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBufferFrontedFileStoreCache.java 2011-10-20
14:41:28 UTC (rev 3572)
+++
trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBufferFrontedFileStoreCache.java 2011-10-20
15:37:18 UTC (rev 3573)
@@ -188,4 +188,26 @@
return fsc;
}
+ @Test public void testSizeIndex() throws Exception {
+ PhysicalInfo info = new PhysicalInfo(1l, 1l, -1, 0);
+ info.setSize(1<<13);
+ assertEquals(0, info.sizeIndex);
+
+ info = new PhysicalInfo(1l, 1l, -1, 0);
+ info.setSize(1 + (1<<13));
+ assertEquals(1, info.sizeIndex);
+
+ info = new PhysicalInfo(1l, 1l, -1, 0);
+ info.setSize(2 + (1<<15));
+ assertEquals(3, info.sizeIndex);
+ }
+
+ @Test(expected=Exception.class) public void testSizeChanged() throws Exception {
+ PhysicalInfo info = new PhysicalInfo(1l, 1l, -1, 0);
+ info.setSize(1<<13);
+ assertEquals(0, info.sizeIndex);
+
+ info.setSize(1 + (1<<13));
+ }
+
}
Modified:
trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStorageManager.java
===================================================================
---
trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStorageManager.java 2011-10-20
14:41:28 UTC (rev 3572)
+++
trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStorageManager.java 2011-10-20
15:37:18 UTC (rev 3573)
@@ -59,6 +59,25 @@
store.remove();
assertEquals(0, sm.getUsedBufferSpace());
}
+
+ @Test public void testPositionalWrite() throws Exception {
+ FileStorageManager sm = getStorageManager(null, null);
+ String tsID = "0"; //$NON-NLS-1$
+ FileStore store = sm.createFileStore(tsID);
+ byte[] expectedBytes = writeBytes(store, 2048);
+ assertEquals(4096, sm.getUsedBufferSpace());
+
+ writeBytes(store, 4096);
+ assertEquals(6144, sm.getUsedBufferSpace());
+
+ byte[] bytesRead = new byte[2048];
+ store.readFully(2048, bytesRead, 0, bytesRead.length);
+
+ assertArrayEquals(expectedBytes, bytesRead);
+
+ store.remove();
+ assertEquals(0, sm.getUsedBufferSpace());
+ }
@Test(expected=IOException.class) public void testMaxSpace() throws Exception {
FileStorageManager sm = getStorageManager(null, null);
@@ -81,15 +100,19 @@
static Random r = new Random();
- static void writeBytes(FileStore store)
+ static void writeBytes(FileStore store) throws IOException {
+ writeBytes(store, store.getLength());
+ }
+
+ static byte[] writeBytes(FileStore store, long start)
throws IOException {
byte[] bytes = new byte[2048];
r.nextBytes(bytes);
- long start = store.getLength();
- store.write(bytes, 0, bytes.length);
+ store.write(start, bytes, 0, bytes.length);
byte[] bytesRead = new byte[2048];
store.readFully(start, bytesRead, 0, bytesRead.length);
assertTrue(Arrays.equals(bytes, bytesRead));
+ return bytes;
}
@Test public void testWritingMultipleFiles() throws Exception {