You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by sc...@apache.org on 2008/01/07 22:13:07 UTC
svn commit: r609779 [1/2] - in
/incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm:
container/CPEFactory.java engine/BoundedWorkQueue.java
engine/CPMEngine.java engine/ProcessingUnit.java
Author: schor
Date: Mon Jan 7 13:13:06 2008
New Revision: 609779
URL: http://svn.apache.org/viewvc?rev=609779&view=rev
Log:
[UIMA-686] fix stop logic to avoid deadlocks, and 100% cpu use spin loop waiting for resource. Also- fix logging in ProcessingUnit to make the code more readable, and shrink it by about a 1/3.
Modified:
incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/container/CPEFactory.java
incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/BoundedWorkQueue.java
incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/CPMEngine.java
incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/ProcessingUnit.java
Modified: incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/container/CPEFactory.java
URL: http://svn.apache.org/viewvc/incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/container/CPEFactory.java?rev=609779&r1=609778&r2=609779&view=diff
==============================================================================
--- incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/container/CPEFactory.java (original)
+++ incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/container/CPEFactory.java Mon Jan 7 13:13:06 2008
@@ -426,7 +426,7 @@
}
// Provide CollectionReader with the number of documents to process
((ConfigurableResource_ImplBase) colreader).setConfigParameterValue("processSize",
- new Integer(new Long(numDocs2Process).intValue()));
+ new Integer((int)numDocs2Process));
CpeConfiguration cpeType = getCpeDescriptor().getCpeConfiguration();
if (cpeType != null && cpeType.getStartingEntityId() != null
&& cpeType.getStartingEntityId().trim().length() > 0) {
Modified: incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/BoundedWorkQueue.java
URL: http://svn.apache.org/viewvc/incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/BoundedWorkQueue.java?rev=609779&r1=609778&r2=609779&view=diff
==============================================================================
--- incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/BoundedWorkQueue.java (original)
+++ incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/BoundedWorkQueue.java Mon Jan 7 13:13:06 2008
@@ -237,11 +237,26 @@
*/
public synchronized Object dequeue(long aTimeout) {
Object resource = dequeue();
- if (resource == null && cpm.isRunning()) {
+ // next 5 lines commented out - was old method of waiting. -- Jan 2008 MIS
+ // changes include waiting a little (WAIT_TIMEOUT) if !cpm.isRunning, to prevent
+ // 100% CPU utilization while waiting for existing processes to finish
+ // also, System.currentTimeMillis only called once in revised version.
+// if (resource == null && cpm.isRunning()) {
+// try {
+// // add 1 millisecond to expire time to account for "rounding" issues
+// long timeExpire = (0 == aTimeout)? Long.MAX_VALUE : (System.currentTimeMillis() + aTimeout + 1);
+// long timeLeft = timeExpire - System.currentTimeMillis();
+ if (resource == null) {
try {
// add 1 millisecond to expire time to account for "rounding" issues
- long timeExpire = (0 == aTimeout)? Long.MAX_VALUE : (System.currentTimeMillis() + aTimeout + 1);
- long timeLeft = timeExpire - System.currentTimeMillis();
+ long timeNow = System.currentTimeMillis();
+ long timeExpire =
+ (! cpm.isRunning()) ?
+ timeNow + WAIT_TIMEOUT : // a value to avoid 100% cpu
+ ((0 == aTimeout) ?
+ Long.MAX_VALUE :
+ timeNow + aTimeout + 1);
+ long timeLeft = timeExpire - timeNow;
while (timeLeft > 0) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
@@ -269,7 +284,6 @@
}
resource = dequeue();
}
-
return resource;
}
Modified: incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/CPMEngine.java
URL: http://svn.apache.org/viewvc/incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/CPMEngine.java?rev=609779&r1=609778&r2=609779&view=diff
==============================================================================
--- incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/CPMEngine.java (original)
+++ incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/CPMEngine.java Mon Jan 7 13:13:06 2008
@@ -570,9 +570,9 @@
UIMAFramework.getLogger(this.getClass()).logrb(Level.INFO, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_terminate_pipelines__INFO",
new Object[] { Thread.currentThread().getName(), String.valueOf(killed) });
- synchronized (workQueue) {
- workQueue.notifyAll();
- }
+// synchronized (workQueue) { // redundant - enqueue call above does this
+// workQueue.notifyAll();
+// }
}
}
@@ -729,7 +729,7 @@
// Already stopped
return;
}
- try {
+// try {
// Change global status
isRunning = false;
// terminate this thread if the thread has been previously suspended
@@ -743,6 +743,19 @@
// stop we should try to clean things up as best as we can. First empty process queue or work
// queue, dump result of analysis into output queue and let the consumers process that.
// When all queues are empty we are done.
+
+ // The logic below (now commented out) has a race condition -
+ // The workQueue / outputQueue can become (temporarily) empty, but then
+ // can be filled with the eof token
+ // But this code proceeds to stop all the CAS processors,
+ // which results in a hang because the pool isn't empty and the process thread waits for
+ // an available cas processor forever.
+
+ // Fix is to not kill the cas processors. Just let them finish normally. The artifact producer
+ // will stop sending new CASes and send through an eof token, which causes normal shutdown to
+ // occur for all the threads.
+
+ /*
if (workQueue != null) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(
@@ -797,6 +810,7 @@
new Object[] { Thread.currentThread().getName() });
}
}
+
for (int i = 0; processingUnits != null && i < processingUnits.length
&& processingUnits[i] != null; i++) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
@@ -806,6 +820,7 @@
}
processingUnits[i].stopCasProcessors(false);
}
+
} catch (Exception e) {
if (UIMAFramework.getLogger().isLoggable(Level.FINER)) {
e.printStackTrace();
@@ -814,6 +829,7 @@
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_exception__FINER",
new Object[] { Thread.currentThread().getName(), e.getMessage() });
}
+ */
}
/**
@@ -2206,9 +2222,9 @@
"UIMA_CPM_done_placed_eof_in_queue__FINEST",
new Object[] { Thread.currentThread().getName(), outputQueue.getName() });
}
- synchronized (outputQueue) {
- outputQueue.notifyAll();
- }
+// synchronized (outputQueue) { // redundant, the above enqueue does this
+// outputQueue.notifyAll();
+// }
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
@@ -2471,9 +2487,9 @@
"UIMA_CPM_done_placed_eof_in_queue__FINEST",
new Object[] { Thread.currentThread().getName(), outputQueue.getName() });
}
- synchronized (outputQueue) {
- outputQueue.notifyAll();
- }
+// synchronized (outputQueue) { // redundant - the above enqueue does this
+// outputQueue.notifyAll();
+// }
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
@@ -2568,9 +2584,9 @@
"UIMA_CPM_done_placed_eof_in_queue__FINEST",
new Object[] { Thread.currentThread().getName(), workQueue.getName() });
}
- synchronized (workQueue) {
- workQueue.notifyAll();
- }
+// synchronized (workQueue) { // redundant - the above enqueue does this
+// workQueue.notifyAll();
+// }
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
@@ -2818,9 +2834,9 @@
throw new ResourceInitializationException(e);
} finally {
casPool.releaseCas(cas);
- synchronized (casPool) {
- casPool.notifyAll();
- }
+// synchronized (casPool) { // redundant, the above releaseCas call does this
+// casPool.notifyAll();
+// }
}
}
@@ -2949,9 +2965,9 @@
if (aCASList[i] != null) {
// aCASList[i].reset();
casPool.releaseCas(aCASList[i]);
- synchronized (casPool) {
- casPool.notifyAll();
- }
+// synchronized (casPool) { // redundant - the above releaseCas call does this
+// casPool.notifyAll();
+// }
} else {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
@@ -3010,8 +3026,15 @@
eofToken[0] = new EOFToken();
outputQueue.enqueue(eofToken);
- synchronized (outputQueue) {
- outputQueue.notifyAll();
+// synchronized (outputQueue) { // redundant - the above enqueue call does this
+// outputQueue.notifyAll();
+// }
+
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
+ "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
+ "UIMA_CPM_done_placed_eof_in_queue__FINEST",
+ new Object[] { Thread.currentThread().getName(), outputQueue.getName() });
}
}
@@ -3056,9 +3079,9 @@
if (this.isRunning() == false) {
readerState = 1009;
casPool.releaseCas(casList[0]);
- synchronized (casPool) {
- casPool.notifyAll();
- }
+// synchronized (casPool) { // redundant - the above releaseCas call does this
+// casPool.notifyAll();
+// }
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
"process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,