8:45 p.m.
Author: shawkins
Date: 2011-06-29 14:45:44 -0400 (Wed, 29 Jun 2011)
New Revision: 3295
Modified:
branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/AbstractWorkItem.java
branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java
branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java
branches/7.4.x/engine/src/test/java/org/teiid/dqp/internal/process/TestWorkItemState.java
Log:
TEIID-1614 fix for possible hang in the more_work state
Modified:
branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/AbstractWorkItem.java
===================================================================
---
branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/AbstractWorkItem.java 2011-06-29
17:50:11 UTC (rev 3294)
+++
branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/AbstractWorkItem.java 2011-06-29
18:45:44 UTC (rev 3295)
@@ -124,15 +124,14 @@
this.threadState = ThreadState.MORE_WORK;
break;
case MORE_WORK:
+ if (this.callingThread != null && !this.isProcessing) {
+ useCallingThread();
+ }
break;
case IDLE:
this.threadState = ThreadState.MORE_WORK;
if (this.callingThread != null) {
- if (this.callingThread == Thread.currentThread()) {
- run(); //restart with the calling thread
- } else {
- this.notifyAll(); //notify the waiting caller
- }
+ useCallingThread();
} else {
resumeProcessing();
}
@@ -144,6 +143,14 @@
LogManager.logDetail(org.teiid.logging.LogConstants.CTX_DQP, new Object[] {this,
"ignoring more work, since the work item is done"}); //$NON-NLS-1$
}
}
+
+ private void useCallingThread() {
+ if (this.callingThread == Thread.currentThread()) {
+ run(); //restart with the calling thread
+ } else {
+ this.notifyAll(); //notify the waiting caller
+ }
+ }
private void logTrace(String msg) {
LogManager.logTrace(org.teiid.logging.LogConstants.CTX_DQP, new Object[] {this, msg,
this.threadState});
Modified: branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java
===================================================================
---
branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2011-06-29
17:50:11 UTC (rev 3294)
+++
branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2011-06-29
18:45:44 UTC (rev 3295)
@@ -93,9 +93,11 @@
private long creationTime = System.currentTimeMillis();
private DQPWorkContext workContext = DQPWorkContext.getWorkContext();
private List<CompletionListener<T>> completionListeners = new
LinkedList<CompletionListener<T>>();
+ private String parentName;
public FutureWork(final Callable<T> processor, int priority) {
super(processor);
+ this.parentName = Thread.currentThread().getName();
this.priority = priority;
}
@@ -105,6 +107,12 @@
}
@Override
+ public void run() {
+ LogManager.logDetail("Running task for parent thread", parentName);
//$NON-NLS-1$
+ super.run();
+ }
+
+ @Override
public int getPriority() {
return priority;
}
@@ -721,6 +729,8 @@
processorDataManager.setMetadataRepository(metadataRepository);
dataTierMgr = new TempTableDataManager(processorDataManager, this.bufferManager,
this.processWorkerPool, this.rsCache, this.matTables, this.cacheFactory);
dataTierMgr.setEventDistributor(eventDistributor);
+
+ LogManager.logDetail(LogConstants.CTX_DQP, "DQPCore started
maxThreads", this.config.getMaxThreads(), "maxActivePlans",
this.maxActivePlans, "source concurrency", this.userRequestSourceConcurrency);
//$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$
}
public void setBufferService(BufferService service) {
Modified:
branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java
===================================================================
---
branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java 2011-06-29
17:50:11 UTC (rev 3294)
+++
branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java 2011-06-29
18:45:44 UTC (rev 3295)
@@ -37,6 +37,8 @@
import org.teiid.common.buffer.BufferManager;
import org.teiid.core.TeiidComponentException;
import org.teiid.language.SQLConstants;
+import org.teiid.logging.LogConstants;
+import org.teiid.logging.LogManager;
import org.teiid.query.QueryPlugin;
import org.teiid.query.metadata.QueryMetadataInterface;
import org.teiid.query.metadata.TempMetadataID;
@@ -96,6 +98,7 @@
public synchronized MatState setState(MatState state, Boolean valid, Long timestamp) {
MatState oldState = this.state;
+ LogManager.logDetail(LogConstants.CTX_MATVIEWS, this, "setting matState to",
state, valid, timestamp, "old values", oldState, this.valid); //$NON-NLS-1$
//$NON-NLS-2$
if (valid != null) {
this.valid = valid;
}
Modified:
branches/7.4.x/engine/src/test/java/org/teiid/dqp/internal/process/TestWorkItemState.java
===================================================================
---
branches/7.4.x/engine/src/test/java/org/teiid/dqp/internal/process/TestWorkItemState.java 2011-06-29
17:50:11 UTC (rev 3294)
+++
branches/7.4.x/engine/src/test/java/org/teiid/dqp/internal/process/TestWorkItemState.java 2011-06-29
18:45:44 UTC (rev 3295)
@@ -181,12 +181,37 @@
for (int i = 0; i < 10 && item.getThreadState() != ThreadState.IDLE; i++) {
Thread.sleep(100);
}
- if (item.getThreadState() != ThreadState.IDLE) {
- fail();
- }
+ assertEquals(ThreadState.IDLE, item.getThreadState());
item.moreWork();
//if we don't return from this call, that means that this thread has been hijacked
-
//we should instead use t.
}
-
+
+ @Test public void testUsingCallingThreadMoreWork() throws Exception {
+ final int[] processCount = new int[1];
+ final TestWorkItem item = new TestWorkItem(false, false, Thread.currentThread()) {
+ @Override
+ protected boolean shouldPause() {
+ return false;
+ }
+
+ @Override
+ protected void process() {
+ super.process();
+ processCount[0]++;
+ }
+ };
+ item.run();
+ assertEquals(ThreadState.IDLE, item.getThreadState());
+ Thread t = new Thread() {
+ @Override
+ public void run() {
+ item.moreWork();
+ }
+ };
+ t.start();
+ t.join();
+ item.moreWork();
+ assertEquals(2, processCount[0]);
+ }
}