9:22 p.m.
Author: jolee
Date: 2013-02-27 15:22:37 -0500 (Wed, 27 Feb 2013)
New Revision: 4555
Modified:
branches/7.7.x/common-core/src/main/java/org/teiid/core/types/DataTypeManager.java
branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java
branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/SizeUtility.java
branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java
branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java
branches/7.7.x/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java
Log:
TEIID-2410: issues with output buffer blocking
Modified:
branches/7.7.x/common-core/src/main/java/org/teiid/core/types/DataTypeManager.java
===================================================================
---
branches/7.7.x/common-core/src/main/java/org/teiid/core/types/DataTypeManager.java 2013-02-27
17:16:02 UTC (rev 4554)
+++
branches/7.7.x/common-core/src/main/java/org/teiid/core/types/DataTypeManager.java 2013-02-27
20:22:37 UTC (rev 4555)
@@ -174,8 +174,16 @@
};
public static final int MAX_STRING_LENGTH = 4000;
- public static final int MAX_LOB_MEMORY_BYTES = 1 << 13;
+ public static final int MAX_LOB_MEMORY_BYTES = Math.max(nextPowOf2(2*MAX_STRING_LENGTH),
1<<13);
+ public static int nextPowOf2(int val) {
+ int result = 1;
+ while (result < val) {
+ result <<= 1;
+ }
+ return result;
+ }
+
public static final class DataTypeAliases {
public static final String VARCHAR = "varchar"; //$NON-NLS-1$
public static final String TINYINT = "tinyint"; //$NON-NLS-1$
Modified:
branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java
===================================================================
---
branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2013-02-27
17:16:02 UTC (rev 4554)
+++
branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2013-02-27
20:22:37 UTC (rev 4555)
@@ -114,7 +114,7 @@
if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR,
MessageLevel.TRACE)) {
LogManager.logTrace(LogConstants.CTX_BUFFER_MGR, "Asynch eviction run",
impl.reserveBatchBytes.get(), impl.maxReserveBytes.get(), impl.activeBatchBytes.get());
//$NON-NLS-1$
}
- impl.doEvictions(0, false);
+ impl.doEvictions(0, !agingOut);
if (!agingOut) {
try {
Thread.sleep(100); //we don't want to evict too fast, because the processing
threads are more than capable of evicting
@@ -691,13 +691,13 @@
}
return;
}
- long maxToFree = Math.max(maxProcessingBytes>>1, reserveBatch>>3);
+ long maxToFree = Math.min(maxProcessingBytes, (activeBatch - reserveBatch)<<1);
doEvictions(maxToFree, true);
}
void doEvictions(long maxToFree, boolean checkActiveBatch) {
int freed = 0;
- while (freed <= maxToFree && (!checkActiveBatch || activeBatchBytes.get()
> reserveBatchBytes.get() * .8)) {
+ while (freed <= maxToFree && (!checkActiveBatch || (maxToFree == 0
&& activeBatchBytes.get() > reserveBatchBytes.get() * .7) || (maxToFree > 0
&& activeBatchBytes.get() > reserveBatchBytes.get() * .8))) {
CacheEntry ce = evictionQueue.firstEntry(true);
if (ce == null) {
break;
@@ -715,9 +715,14 @@
} finally {
synchronized (ce) {
if (evicted && memoryEntries.remove(ce.getId()) != null) {
- freed += ce.getSizeEstimate();
+ if (maxToFree > 0) {
+ freed += ce.getSizeEstimate();
+ }
activeBatchBytes.addAndGet(-ce.getSizeEstimate());
evictionQueue.remove(ce); //ensures that an intervening get will still be cleaned
+ if (!checkActiveBatch) {
+ break;
+ }
}
}
}
Modified:
branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/SizeUtility.java
===================================================================
---
branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/SizeUtility.java 2013-02-27
17:16:02 UTC (rev 4554)
+++
branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/SizeUtility.java 2013-02-27
20:22:37 UTC (rev 4555)
@@ -47,7 +47,7 @@
private static Map<Class<?>, int[]> SIZE_ESTIMATES = new
HashMap<Class<?>, int[]>(128);
private static Set<Class<?>> VARIABLE_SIZE_TYPES = new
HashSet<Class<?>>();
static {
- SIZE_ESTIMATES.put(DataTypeManager.DefaultDataClasses.STRING, new int[] {100, 256});
+ SIZE_ESTIMATES.put(DataTypeManager.DefaultDataClasses.STRING, new int[] {100,
Math.max(100, DataTypeManager.nextPowOf2(DataTypeManager.MAX_STRING_LENGTH/16))});
SIZE_ESTIMATES.put(DataTypeManager.DefaultDataClasses.DATE, new int[] {20, 28});
SIZE_ESTIMATES.put(DataTypeManager.DefaultDataClasses.TIME, new int[] {20, 28});
SIZE_ESTIMATES.put(DataTypeManager.DefaultDataClasses.TIMESTAMP, new int[] {20, 28});
Modified: branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java
===================================================================
---
branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2013-02-27
17:16:02 UTC (rev 4554)
+++
branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2013-02-27
20:22:37 UTC (rev 4555)
@@ -431,15 +431,18 @@
}
}
- public boolean hasWaitingPlans(RequestWorkItem item) {
- synchronized (waitingPlans) {
- if (!waitingPlans.isEmpty()) {
- return true;
- }
- this.bufferFullPlans.add(item);
- }
- return false;
- }
+ public boolean blockOnOutputBuffer(RequestWorkItem item) {
+ synchronized (waitingPlans) {
+ if (!waitingPlans.isEmpty()) {
+ return false;
+ }
+ if (item.useCallingThread || item.getDqpWorkContext().getSession().isEmbedded())
{
+ return false;
+ }
+ this.bufferFullPlans.add(item);
+ }
+ return true;
+ }
void removeRequest(final RequestWorkItem workItem) {
finishProcessing(workItem);
Modified:
branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java
===================================================================
---
branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2013-02-27
17:16:02 UTC (rev 4554)
+++
branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2013-02-27
20:22:37 UTC (rev 4555)
@@ -550,11 +550,12 @@
super.flushBatchDirect(batch, add);
if (!add && !processor.hasFinalBuffer()) {
resultsBuffer.setRowCount(batch.getEndRow());
- } else if (!processor.hasFinalBuffer() //restrict the buffer size for forward only
results
+ } else if (isForwardOnly() && add
+ && !processor.hasFinalBuffer() //restrict the buffer size for forward only
results
&& !batch.getTerminationFlag()
&& transactionState != TransactionState.ACTIVE
- && this.getTupleBuffer().getManagedRowCount() >=
OUTPUT_BUFFER_MAX_BATCHES * this.getTupleBuffer().getBatchSize()) {
- if (!dqpCore.hasWaitingPlans(RequestWorkItem.this)) {
+ && resultsBuffer.getManagedRowCount() >= OUTPUT_BUFFER_MAX_BATCHES *
resultsBuffer.getBatchSize()) {
+ if (dqpCore.blockOnOutputBuffer(RequestWorkItem.this)) {
//requestMore will trigger more processing
throw BlockedException.block(requestID, "Blocking due to full results
TupleBuffer", //$NON-NLS-1$
this.getTupleBuffer().getId(), "rows",
this.getTupleBuffer().getManagedRowCount(), "batch size",
this.getTupleBuffer().getBatchSize()); //$NON-NLS-1$ //$NON-NLS-2$
Modified:
branches/7.7.x/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java
===================================================================
---
branches/7.7.x/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java 2013-02-27
17:16:02 UTC (rev 4554)
+++
branches/7.7.x/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java 2013-02-27
20:22:37 UTC (rev 4555)
@@ -345,8 +345,30 @@
break;
}
}
+
+ //insensitive should not block
+ reqMsg.setCursorType(ResultSet.TYPE_SCROLL_INSENSITIVE);
+
+ message = core.executeRequest(reqMsg.getExecutionId(), reqMsg);
+ rm = message.get(500000, TimeUnit.MILLISECONDS);
+ assertNull(rm.getException());
+
+ assertEquals(rowsPerBatch, rm.getResultsList().size());
+ item =
core.getRequestWorkItem(DQPWorkContext.getWorkContext().getRequestID(reqMsg.getExecutionId()));
+
+ message = core.processCursorRequest(reqMsg.getExecutionId(), 9, rowsPerBatch);
+ rm = message.get(500000, TimeUnit.MILLISECONDS);
+ assertNull(rm.getException());
+ assertEquals(rowsPerBatch, rm.getResultsList().size());
+ //ensure that we are idle
+ for (int i = 0; i < 10 && item.getThreadState() != ThreadState.IDLE; i++)
{
+ Thread.sleep(100);
+ }
+ assertEquals(ThreadState.IDLE, item.getThreadState());
+ assertEquals(item.resultsBuffer.getManagedRowCount(), 400); //should have the full
results
}
+
@Test public void testBufferReuse() throws Exception {
//the sql should return 100 rows
String sql = "SELECT A.IntKey FROM BQT1.SmallA as A, BQT1.SmallA as B ORDER
BY A.IntKey"; //$NON-NLS-1$