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 2015/05/08 16:03:09 UTC
svn commit: r1678355 - in
/uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima:
analysis_engine/impl/ cas/impl/ cas/test/ internal/util/ jcas/impl/
Author: schor
Date: Fri May 8 14:03:08 2015
New Revision: 1678355
URL: http://svn.apache.org/r1678355
Log:
[UIMA-4383] better multi-threaded testing by first creating/starting threads which then "hold" at a start point, then quickly doing notifies to get them going.
Modified:
uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/analysis_engine/impl/MultiprocessingAnalysisEngine_implTest.java
uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/impl/XmiCasDeserializerTest.java
uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/test/IteratorTest.java
uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/AnalysisEnginePoolTest.java
uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/MultiThreadUtils.java
uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/jcas/impl/JCasHashMapTest.java
Modified: uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/analysis_engine/impl/MultiprocessingAnalysisEngine_implTest.java
URL: http://svn.apache.org/viewvc/uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/analysis_engine/impl/MultiprocessingAnalysisEngine_implTest.java?rev=1678355&r1=1678354&r2=1678355&view=diff
==============================================================================
--- uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/analysis_engine/impl/MultiprocessingAnalysisEngine_implTest.java (original)
+++ uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/analysis_engine/impl/MultiprocessingAnalysisEngine_implTest.java Fri May 8 14:03:08 2015
@@ -38,6 +38,8 @@ import org.apache.uima.analysis_engine.m
import org.apache.uima.analysis_engine.metadata.impl.FixedFlow_impl;
import org.apache.uima.cas.CAS;
import org.apache.uima.cas.TypeSystem;
+import org.apache.uima.internal.util.MultiThreadUtils;
+import org.apache.uima.internal.util.MultiThreadUtils.ThreadM;
import org.apache.uima.resource.ResourceSpecifier;
import org.apache.uima.resource.metadata.Capability;
import org.apache.uima.resource.metadata.ConfigurationParameter;
@@ -52,7 +54,9 @@ import org.apache.uima.util.XMLInputSour
public class MultiprocessingAnalysisEngine_implTest extends TestCase {
-
+
+ private final static boolean doSleeps = true; // running w/ dosleeps = false should show 100% cpu
+
private AnalysisEngineDescription mSimpleDesc;
private AnalysisEngineDescription mAggDesc;
@@ -198,35 +202,53 @@ public class MultiprocessingAnalysisEngi
_testProcess(mAggDesc, 0);
- // multiple threads!
+ /**
+ * Multi threading ae processing tests
+ *
+ * The goal is to run multiple AE's together at the same time, and repeat this.
+ *
+ * The threads the AE's run on are created and started, once, ahead of time, since that takes a while.
+ * When they start their run() method executes a wait on a waitnotify object; and they sit until
+ * notified.
+ *
+ * At the beginning, all threads are notified quickly, and start running their individual test(s).
+ *
+ * When the test is finished, the thread goes back into the wait4go state.
+ *
+ */
MultiprocessingAnalysisEngine_impl ae = new MultiprocessingAnalysisEngine_impl();
Map<String, Object> params = new HashMap<String, Object>();
- params.put(AnalysisEngine.PARAM_NUM_SIMULTANEOUS_REQUESTS, Integer.valueOf(3));
+ params.put(AnalysisEngine.PARAM_NUM_SIMULTANEOUS_REQUESTS, Integer.valueOf(8));
ae.initialize(mAggDesc, params);
final int NUM_THREADS = Math.min(50, Runtime.getRuntime().availableProcessors() * 5);
ProcessThread[] threads = new ProcessThread[NUM_THREADS];
- Random random = new Random();
- for (int repetitions = 0; repetitions < 4; repetitions++) {
- for (int i = 0; i < NUM_THREADS; i++) {
- threads[i] = new ProcessThread(ae);
- threads[i].start();
- Thread.sleep(0, random.nextInt(100000)); // delay between 0 and 100 microseconds
- }
-
- // wait for threads to finish and check if they got exceptions
- for (int i = 0; i < NUM_THREADS; i++) {
- try {
- threads[i].join();
- } catch (InterruptedException ie) {
- System.err.println("got unexpected Interrupted exception " + ie);
- }
- if (threads[i].isAlive()) {
- System.err.println("timeout waiting for thread to complete " + i);
- fail("timeout waiting for thread to complete " + i);
- }
+ Random random = new Random();
- Throwable failure = threads[i].getFailure();
+ for (int i = 0; i < NUM_THREADS; i++) {
+ threads[i] = new ProcessThread(ae);
+ threads[i].start();
+ }
+
+ for (int repetitions = 0; repetitions < 4; repetitions++) {
+
+ MultiThreadUtils.kickOffThreads(threads);
+
+ // wait for all threads to finish and check if they got exceptions
+ MultiThreadUtils.waitForAllReady(threads);
+
+ for (ProcessThread pt : threads) {
+// try {
+// threads[i].join();
+// } catch (InterruptedException ie) {
+// System.err.println("got unexpected Interrupted exception " + ie);
+// }
+// if (threads[i].isAlive()) {
+// System.err.println("timeout waiting for thread to complete " + i);
+// fail("timeout waiting for thread to complete " + i);
+// }
+
+ Throwable failure = pt.getFailure();
if (failure != null) {
if (failure instanceof Exception) {
throw (Exception) failure;
@@ -245,7 +267,7 @@ public class MultiprocessingAnalysisEngi
ResultSpecification resultSpec = new ResultSpecification_impl(lastResultSpec.getTypeSystem());
resultSpec.addResultType("NamedEntity", true);
assertEquals(resultSpec, lastResultSpec);
-
+ MultiThreadUtils.terminateThreads(threads);
} catch (Exception e) {
JUnitExtension.handleException(e);
}
@@ -273,7 +295,7 @@ public class MultiprocessingAnalysisEngi
final int NUM_THREADS = Math.min(50, Runtime.getRuntime().availableProcessors() * 5);
final int NUM_INSTANCES = (int)(NUM_THREADS * .7);
-
+
public void processMany(ResourceSpecifier specifier) throws Exception {
try {
@@ -289,20 +311,15 @@ public class MultiprocessingAnalysisEngi
threads[i] = new ProcessThreadMany(ae);
threads[i].start();
}
+
+ MultiThreadUtils.kickOffThreads(threads);
+
+ // wait for all threads to finish and check if they got exceptions
+ MultiThreadUtils.waitForAllReady(threads);
- // wait for threads to finish and check if they got exceptions
- for (int i = 0; i < NUM_THREADS; i++) {
- try {
- threads[i].join(); // increase timeout - Jenkins sometimes is slow
- } catch (InterruptedException ie) {
- System.err.println("got unexpected Interrupted exception " + ie);
- }
- if (threads[i].isAlive()) {
- System.err.println("timeout waiting for thread to complete " + i);
- fail("timeout waiting for thread to complete " + i);
- }
+ for (ProcessThreadMany ptm : threads) {
- Throwable failure = threads[i].getFailure();
+ Throwable failure = ptm.getFailure();
if (failure != null) {
if (failure instanceof Exception) {
throw (Exception) failure;
@@ -320,7 +337,7 @@ public class MultiprocessingAnalysisEngi
ResultSpecification resultSpec = new ResultSpecification_impl(lastResultSpec.getTypeSystem());
resultSpec.addResultType("NamedEntity", true);
assertEquals(resultSpec, lastResultSpec);
-
+ MultiThreadUtils.terminateThreads(threads);
} catch (Exception e) {
JUnitExtension.handleException(e);
}
@@ -431,104 +448,125 @@ public class MultiprocessingAnalysisEngi
assertEquals(resultSpec, TestAnnotator.lastResultSpec);
tcas.reset();
}
-
- class ProcessThread extends Thread {
+
+ class ProcessThread extends ThreadM {
+
+ Throwable mFailure = null;
+
+ AnalysisEngine mAE;
+
ProcessThread(AnalysisEngine aAE) {
mAE = aAE;
}
public void run() {
Random r = new Random();
- try {
+ while (true) {
- // Test each form of the process method. When TestAnnotator executes, it
- // stores in static fields the document text and the ResultSpecification.
- // We use thse to make sure the information propogates correctly to the
- // annotator. (However, we can't check these until after the threads are
- // finished, as their state is nondeterministic during multithreaded
- // processing.)
-
- // process(CAS)
- CAS tcas = mAE.newCAS();
- mLastTypeSystem = tcas.getTypeSystem();
- tcas.setDocumentText("new test");
- mAE.process(tcas);
- Thread.sleep(0, r.nextInt(10000)); // 0 to 10 microseconds
- tcas.reset();
-
- // process(CAS,ResultSpecification)
- ResultSpecification resultSpec = new ResultSpecification_impl(tcas.getTypeSystem());
- resultSpec.addResultType("NamedEntity", true);
-
- tcas.setDocumentText("testing...");
- Thread.sleep(0, r.nextInt(100000)); // 0 to 100 microseconds
- mAE.process(tcas, resultSpec);
- Thread.sleep(0, r.nextInt(1000)); // 0 to 1 microseconds
- tcas.reset();
-
- } catch (Throwable t) {
- t.printStackTrace();
- //can't cause unit test to fail by throwing exception from thread.
- //record the failure and the main thread will check for it later.
- mFailure = t;
+ if (!MultiThreadUtils.wait4go(this)) { // wait for go signal after all threads are setup.
+ break; // time to terminate
+ }
+
+ try {
+
+ // Test each form of the process method. When TestAnnotator executes, it
+ // stores in static fields the document text and the ResultSpecification.
+ // We use thse to make sure the information propogates correctly to the
+ // annotator. (However, we can't check these until after the threads are
+ // finished, as their state is nondeterministic during multithreaded
+ // processing.)
+
+ // process(CAS)
+ CAS tcas = mAE.newCAS();
+// for (int i = 0; i < 1000; i++) { // uncomment to debug
+ mLastTypeSystem = tcas.getTypeSystem();
+ tcas.setDocumentText("new test");
+ mAE.process(tcas);
+ // System.out.println("Debug finished processing a cas");
+ if (doSleeps)
+ Thread.sleep(0, r.nextInt(1000)); // 0 to 1 microseconds
+ tcas.reset();
+
+ // process(CAS,ResultSpecification)
+ ResultSpecification resultSpec = new ResultSpecification_impl(tcas.getTypeSystem());
+ resultSpec.addResultType("NamedEntity", true);
+
+ tcas.setDocumentText("testing...");
+ if (doSleeps)
+ Thread.sleep(0, r.nextInt(1000)); // 0 to 1 microseconds
+ mAE.process(tcas, resultSpec);
+ if (doSleeps)
+ Thread.sleep(0, r.nextInt(1000)); // 0 to 1 microseconds
+ tcas.reset();
+// }
+ } catch (Throwable t) {
+ t.printStackTrace();
+ //can't cause unit test to fail by throwing exception from thread.
+ //record the failure and the main thread will check for it later.
+ mFailure = t;
+ }
}
}
public synchronized Throwable getFailure() {
return mFailure;
}
-
- Throwable mFailure = null;
- AnalysisEngine mAE;
+ public void resetFailure() {
+ mFailure = null;
+ }
}
- class ProcessThreadMany extends Thread {
-
- Throwable mFailure = null;
-
- AnalysisEngine mAE;
+ class ProcessThreadMany extends ProcessThread {
ProcessThreadMany(AnalysisEngine aAE) {
- mAE = aAE;
+ super(aAE);
}
public void run() {
- try {
-
- Random r = new Random();
-
- // Test each form of the process method. When TestAnnotator executes, it
- // stores in static fields the document text and the ResultSpecification.
- // We use thse to make sure the information propagates correctly to the
- // annotator. (However, we can't check these until after the threads are
- // finished, as their state is nondeterministic during multithreaded
- // processing.)
+
+ while (true) {
+
+ if (!MultiThreadUtils.wait4go(this)) {
+ break;
+ }
- // process(CAS)
- for (int i = 0; i < 5; i++) {
- CAS tcas = mAE.newCAS();
- mLastTypeSystem = tcas.getTypeSystem();
- tcas.setDocumentText("new test");
- mAE.process(tcas);
- Thread.sleep(0, r.nextInt(100000)); // between 0 and 100 microseconds
- tcas.reset();
-
- // process(CAS,ResultSpecification)
- ResultSpecification resultSpec = new ResultSpecification_impl(tcas.getTypeSystem());
- resultSpec.addResultType("NamedEntity", true);
-
- tcas.setDocumentText("testing...");
- Thread.sleep(0, r.nextInt(1000)); // probably yields, sleep for 1 microsecond
- mAE.process(tcas, resultSpec);
- Thread.sleep(0, r.nextInt(10000)); // between 0 and 10 microseconds
- tcas.reset();
+ try {
+
+ Random r = new Random();
+
+ // Test each form of the process method. When TestAnnotator executes, it
+ // stores in static fields the document text and the ResultSpecification.
+ // We use thse to make sure the information propagates correctly to the
+ // annotator. (However, we can't check these until after the threads are
+ // finished, as their state is nondeterministic during multithreaded
+ // processing.)
+
+ // process(CAS)
+ for (int i = 0; i < 5; i++) {
+ CAS tcas = mAE.newCAS();
+ mLastTypeSystem = tcas.getTypeSystem();
+ tcas.setDocumentText("new test");
+ mAE.process(tcas);
+ Thread.sleep(0, r.nextInt(1000)); // between 0 and 1 microseconds
+ tcas.reset();
+
+ // process(CAS,ResultSpecification)
+ ResultSpecification resultSpec = new ResultSpecification_impl(tcas.getTypeSystem());
+ resultSpec.addResultType("NamedEntity", true);
+
+ tcas.setDocumentText("testing...");
+ Thread.sleep(0, r.nextInt(1000)); // between 0 and 1 microseconds
+ mAE.process(tcas, resultSpec);
+ Thread.sleep(0, r.nextInt(1000)); // between 0 and 1 microseconds
+ tcas.reset();
+ }
+ } catch (Throwable t) {
+ t.printStackTrace();
+ //can't cause unit test to fail by throwing exception from thread.
+ //record the failure and the main thread will check for it later.
+ mFailure = t;
}
- } catch (Throwable t) {
- t.printStackTrace();
- //can't cause unit test to fail by throwing exception from thread.
- //record the failure and the main thread will check for it later.
- mFailure = t;
}
}
Modified: uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/impl/XmiCasDeserializerTest.java
URL: http://svn.apache.org/viewvc/uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/impl/XmiCasDeserializerTest.java?rev=1678355&r1=1678354&r2=1678355&view=diff
==============================================================================
--- uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/impl/XmiCasDeserializerTest.java (original)
+++ uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/impl/XmiCasDeserializerTest.java Fri May 8 14:03:08 2015
@@ -57,6 +57,8 @@ import org.apache.uima.cas.impl.XmiSeria
import org.apache.uima.cas.impl.XmiSerializationSharedData.XmiArrayElement;
import org.apache.uima.cas.text.AnnotationFS;
import org.apache.uima.cas_data.impl.CasComparer;
+import org.apache.uima.internal.util.MultiThreadUtils;
+import org.apache.uima.internal.util.MultiThreadUtils.ThreadM;
import org.apache.uima.internal.util.XmlAttribute;
import org.apache.uima.internal.util.XmlElementNameAndContents;
import org.apache.uima.resource.metadata.FsIndexDescription;
@@ -259,16 +261,20 @@ public class XmiCasDeserializerTest exte
public void run() {
try {
- serialize(cas, null);
+ while (true) {
+ if (!MultiThreadUtils.wait4go((ThreadM) Thread.currentThread())) {
+ break;
+ }
+
+ serialize(cas, null);
+ }
// serialize(cas, null);
// serialize(cas, null);
// serialize(cas, null);
} catch (IOException e) {
-
- e.printStackTrace();
+ throw new RuntimeException(e);
} catch (SAXException e) {
-
- e.printStackTrace();
+ throw new RuntimeException(e);
}
}
}
@@ -301,23 +307,29 @@ public class XmiCasDeserializerTest exte
CasCopier.copyCas(cas, cases[i], true);
}
- // start n threads, serializing as XMI
+ // start n threads, serializing as XMI
+ MultiThreadUtils.ThreadM [] threads = new MultiThreadUtils.ThreadM[MAX_THREADS];
+ for (int i = 0; i < MAX_THREADS; i++) {
+ threads[i] = new MultiThreadUtils.ThreadM(new DoSerialize(cases[i]));
+ threads[i].start();
+ }
+ MultiThreadUtils.waitForAllReady(threads);
for (int i = 0; i < threadsToUse.length; i++) {
- Thread [] threads = new Thread[MAX_THREADS];
+ MultiThreadUtils.ThreadM[] sliceOfThreads = new MultiThreadUtils.ThreadM[threadsToUse[i]];
+ System.arraycopy(threads, 0, sliceOfThreads, 0, threadsToUse[i]);
+
long startTime = System.currentTimeMillis();
- for (int ti = 0; ti < threadsToUse[i]; ti++) {
- threads[ti] = new Thread(new DoSerialize(cases[ti]));
-
- threads[ti].start();
- }
- for (int ti = 0; ti < threadsToUse[i]; ti++) {
- threads[ti].join();
- //System.out.print(" "+ ti);
- }
+
+ MultiThreadUtils.kickOffThreads(sliceOfThreads);
+
+ MultiThreadUtils.waitForAllReady(sliceOfThreads);
+
System.out.println("\nNumber of threads serializing: " + threadsToUse[i] +
" Normalized millisecs (should be close to the same): " + (System.currentTimeMillis() - startTime) / threadsToUse[i]);
}
+
+ MultiThreadUtils.terminateThreads(threads);
}
public void testDeltaCasIndexExistingFsInView() throws Exception {
Modified: uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/test/IteratorTest.java
URL: http://svn.apache.org/viewvc/uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/test/IteratorTest.java?rev=1678355&r1=1678354&r2=1678355&view=diff
==============================================================================
--- uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/test/IteratorTest.java (original)
+++ uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/cas/test/IteratorTest.java Fri May 8 14:03:08 2015
@@ -50,6 +50,7 @@ import org.apache.uima.cas.impl.LowLevel
import org.apache.uima.cas.impl.TypeImpl;
import org.apache.uima.cas.text.AnnotationFS;
import org.apache.uima.internal.util.IntVector;
+import org.apache.uima.internal.util.MultiThreadUtils;
import org.apache.uima.jcas.JCas;
import org.apache.uima.jcas.tcas.Annotation;
import org.apache.uima.resource.ResourceInitializationException;
@@ -400,44 +401,52 @@ public class IteratorTest extends TestCa
int numberOfCores = Math.min(50, Runtime.getRuntime().availableProcessors() * 5);
System.out.println("test multicore iterator with " + numberOfCores + " threads");
- Thread[] threads = new Thread[numberOfCores];
+ MultiThreadUtils.ThreadM[] threads = new MultiThreadUtils.ThreadM[numberOfCores];
final Throwable[] tthrowable = new Throwable[1]; // trick to get a return value in a parameter
tthrowable[0] = null;
- for (int r = 0; r < 10; r++) {
- for (int i = 0; i < numberOfCores; i++) {
- final int finalI = i;
- threads[i] = new Thread(new Runnable() {
-
- public void run() {
- try {
+ for (int i = 0; i < numberOfCores; i++) {
+ final int finalI = i;
+ threads[i] = new MultiThreadUtils.ThreadM() {
+
+ public void run() {
+ try {
+ while (true) {
+ if (!MultiThreadUtils.wait4go(this)) {
+ break;
+ }
setIteratorWithoutMods(setIndex, finalI);
sortedIteratorWithoutMods(sortedIndex);
bagIteratorWithoutMods(bagIndex);
- } catch (Throwable e) {
- tthrowable[0] = e;
- e.printStackTrace();
- throw new RuntimeException(e);
}
- }} );
- threads[i].start();
- }
+ } catch (Throwable e) {
+ tthrowable[0] = e;
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }};
+ threads[i].start();
+ }
+
+
+ for (int r = 0; r < 10; r++) {
+
+ MultiThreadUtils.kickOffThreads(threads);
+
+ MultiThreadUtils.waitForAllReady(threads);
+
for (int i = 0; i < numberOfCores; i++) {
- try {
if (tthrowable[0] != null) {
assertTrue(false);
}
- threads[i].join();
if (tthrowable[0] != null) {
assertTrue(false);
}
- } catch (InterruptedException e) {
- e.printStackTrace();
- assertTrue(false);
- }
}
}
+
+ MultiThreadUtils.terminateThreads(threads);
}
public void testIterator() {
Modified: uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/AnalysisEnginePoolTest.java
URL: http://svn.apache.org/viewvc/uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/AnalysisEnginePoolTest.java?rev=1678355&r1=1678354&r2=1678355&view=diff
==============================================================================
--- uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/AnalysisEnginePoolTest.java (original)
+++ uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/AnalysisEnginePoolTest.java Fri May 8 14:03:08 2015
@@ -123,10 +123,12 @@ public class AnalysisEnginePoolTest exte
threads[i] = new ProcessThread(pool, i);
threads[i].start();
}
+
+ MultiThreadUtils.kickOffThreads(threads);
+
+ MultiThreadUtils.waitForAllReady(threads);
- // wait for threads to finish and check if they got exceptions
for (int i = 0; i < NUM_THREADS; i++) {
- threads[i].join();
Throwable failure = threads[i].getFailure();
if (failure != null) {
if (failure instanceof Exception) {
@@ -145,6 +147,7 @@ public class AnalysisEnginePoolTest exte
resultSpec.addResultType("NamedEntity", true);
assertEquals(resultSpec, TestAnnotator.getLastResultSpec());
+ MultiThreadUtils.terminateThreads(threads);
} catch (Exception e) {
JUnitExtension.handleException(e);
}
@@ -228,22 +231,27 @@ public class AnalysisEnginePoolTest exte
}
}
- class ProcessThread extends Thread {
+ class ProcessThread extends MultiThreadUtils.ThreadM {
ProcessThread(AnalysisEnginePool aPool, int aId) {
mPool = aPool;
mId = aId;
}
public void run() {
- try {
- // System.out.println("thread started");
- _testProcess(mPool, mId);
- // System.out.println("thread finished");
- } catch (Throwable t) {
- t.printStackTrace();
- //can't cause unit test to fail by throwing exception from thread.
- //record the failure and the main thread will check for it later.
- mFailure = t;
+ while (true) {
+ try {
+ if (!MultiThreadUtils.wait4go(this)) {
+ break;
+ }
+ // System.out.println("thread started");
+ _testProcess(mPool, mId);
+ // System.out.println("thread finished");
+ } catch (Throwable t) {
+ t.printStackTrace();
+ //can't cause unit test to fail by throwing exception from thread.
+ //record the failure and the main thread will check for it later.
+ mFailure = t;
+ }
}
}
Modified: uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/MultiThreadUtils.java
URL: http://svn.apache.org/viewvc/uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/MultiThreadUtils.java?rev=1678355&r1=1678354&r2=1678355&view=diff
==============================================================================
--- uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/MultiThreadUtils.java (original)
+++ uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/internal/util/MultiThreadUtils.java Fri May 8 14:03:08 2015
@@ -270,4 +270,115 @@ public class MultiThreadUtils extends Te
}
}
}
+
+ /*
+ * Thread management utilities
+ */
+
+ public final static int THREAD_ABOUT_TO_WAIT = 1;
+ public final static int THREAD_RUNNING = 2;
+ private final static int THREAD_TERMINATE = 3;
+ private final static int THREAD_ABOUT_TO_RUN = 4;
+
+ public static class ThreadM extends Thread {
+ public ThreadM(Runnable runnable) {
+ super(runnable);
+ }
+
+ public ThreadM() {
+ super();
+ }
+
+ public volatile int state = 0;
+ }
+
+ public static void waitForAllReady(ThreadM[] threads) {
+ for (ThreadM thread : threads) {
+ wait4wait(thread);
+ }
+ }
+
+// public static void waitForAllTerminate(ThreadM[] threads) {
+// for (Thread t : threads) {
+// while (true) {
+// if (t.getState() == Thread.State.TERMINATED) {
+// break;
+// }
+// sleep(10000);
+// }
+// }
+// }
+
+ public static void sleep(int nanoSeconds) {
+ try {
+ Thread.sleep(0, nanoSeconds);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void wait4wait(ThreadM t) {
+ while (true) {
+ if (t.state == THREAD_ABOUT_TO_WAIT && t.getState() == Thread.State.WAITING) {
+ return;
+ }
+ sleep(1000);
+ }
+ }
+ /**
+ *
+ * @param t the thread
+ * @return true if running, false if time to terminate
+ */
+ public static boolean wait4go(ThreadM t) {
+ synchronized(t) {
+ try {
+ t.state = THREAD_ABOUT_TO_WAIT;
+ t.wait();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ if (t.state == MultiThreadUtils.THREAD_TERMINATE) {
+ return false;
+ }
+ t.state = MultiThreadUtils.THREAD_RUNNING;
+ return true;
+ }
+
+ public static void kickOffThreads(ThreadM[] threads) {
+ waitForAllReady(threads);
+ // after all threads in wait state, rapidly release them
+ for (ThreadM t : threads) {
+ synchronized (t) {
+ t.state = THREAD_ABOUT_TO_RUN;
+ t.notify();
+// System.out.println("Debug notifying thread " + thread.getName());
+ }
+ }
+ // insure all threads started
+ for (ThreadM t : threads) {
+ wait4start(t);
+ }
+ }
+
+ private static void wait4start(ThreadM t) {
+ while (true) {
+ if (t.state != THREAD_ABOUT_TO_RUN || t.getState() == Thread.State.RUNNABLE) {
+ return;
+ }
+ sleep(1000);
+ }
+ }
+
+ public static void terminateThreads(ThreadM[] threads) {
+ for (ThreadM t : threads) {
+ wait4wait(t);
+ t.state = THREAD_TERMINATE;
+ synchronized (t) {
+ t.notify();
+ }
+ }
+ }
+
}
Modified: uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/jcas/impl/JCasHashMapTest.java
URL: http://svn.apache.org/viewvc/uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/jcas/impl/JCasHashMapTest.java?rev=1678355&r1=1678354&r2=1678355&view=diff
==============================================================================
--- uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/jcas/impl/JCasHashMapTest.java (original)
+++ uima/uimaj/trunk/uimaj-core/src/test/java/org/apache/uima/jcas/impl/JCasHashMapTest.java Fri May 8 14:03:08 2015
@@ -19,6 +19,7 @@
package org.apache.uima.jcas.impl;
+import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -203,45 +204,49 @@ public class JCasHashMapTest extends Tes
Thread thisThread = Thread.currentThread();
final int subThreadPriority = thisThread.getPriority();
- thisThread.setPriority(subThreadPriority - 1);
+ thisThread.setPriority(subThreadPriority - 1);
+ final MultiThreadUtils.ThreadM[] threads = new MultiThreadUtils.ThreadM[numberOfThreads];
+ final JCasHashMap m = new JCasHashMap(200, true); // true = do use cache
+ final Random r = new Random(); // used to sleep from 0 to 4 milliseconds
+ final int hashKey = 15;
+ final TOP fs = new TOP(hashKey, FAKE_TOP_TYPE_INSTANCE);
+ final FeatureStructureImpl[] found = new FeatureStructureImpl[numberOfThreads];
+
+ for (int i = 0; i < numberOfThreads; i++) {
+ final int finalI = i;
+ threads[i] = new MultiThreadUtils.ThreadM() {
+ public void run() {
+ while (true) {
+ if (!MultiThreadUtils.wait4go(this)) {
+ break;
+ }
+ MultiThreadUtils.sleep(r.nextInt(500000)); // 0-500 microseconds
+ found[finalI] = m.getReserve(hashKey);
+ }
+ }
+ };
+ threads[i].setPriority(subThreadPriority);
+ threads[i].start();
+ }
for (int loopCount = 0; loopCount < 10; loopCount ++) {
System.out.println(" JCasHashMap collide loop count is " + loopCount);
- final Random r = new Random(); // used to sleep from 0 to 4 milliseconds
- final int hashKey = 15;
- final TOP fs = new TOP(hashKey, FAKE_TOP_TYPE_INSTANCE);
// create threads and start them
- // note: this needs to be fixed, following the discovery that creating a thread takes much longer than expected;
- // use the Executor framework instead and create the threads ahead of time. See or use MultiThreadUtils
for (int th = 2; th <= numberOfThreads; th *= 2) {
JCasHashMap.setDEFAULT_CONCURRENCY_LEVEL(th);
- final JCasHashMap m = new JCasHashMap(200, true); // true = do use cache
-
- final Thread[] threads = new Thread[numberOfThreads];
- final FeatureStructureImpl[] found = new FeatureStructureImpl[numberOfThreads];
+ Arrays.fill(found, null);
+ m.clear();
- for (int i = 0; i < numberOfThreads; i++) {
- final int finalI = i;
- threads[i] = new Thread(new Runnable() {
- public void run() {
- try {
- Thread.sleep(r.nextInt(5));
- } catch (InterruptedException e) {
- }
- found[finalI] = m.getReserve(hashKey);
- }
- });
- threads[i].setPriority(subThreadPriority);
- threads[i].start();
- }
+ MultiThreadUtils.kickOffThreads(threads);
+
Thread.sleep(20);
// verify that one thread finished, others are waiting, because of the reserve.
// this assumes that all the threads got to run.
int numberWaiting = 0;
int threadFinished = -1;
for (int i = 0; i < numberOfThreads; i++) {
- if (threads[i].isAlive()) {
+ if (threads[i].state == MultiThreadUtils.THREAD_RUNNING) {
numberWaiting ++;
} else {
threadFinished = i;
@@ -251,28 +256,30 @@ public class JCasHashMapTest extends Tes
assertEquals(numberOfThreads - 1, numberWaiting); // expected 7 but was 8
m.put(fs);
found[threadFinished] = fs;
+
+ MultiThreadUtils.waitForAllReady(threads);
- // loop a few times to give enough time for the other threads to finish.
- long startOfWait = System.currentTimeMillis();
- while (System.currentTimeMillis() - startOfWait < 30000) { // wait up to 30 seconds in case of machine stall
-
- // Attempt to insure we let the threads under test run in preference to this one
- Thread.sleep(20); // imprecise. Intent is to allow other thread that was waiting, to run
- // before this thread resumes. Depends on thread priorities, but
- // multiple threads could be running at the same time.
-
- numberWaiting = 0;
- for (int i = 0; i < numberOfThreads; i++) {
- if (threads[i].isAlive()) {
- numberWaiting ++;
- }
- }
- if (numberWaiting == 0) {
- break;
- }
- }
+// // loop a few times to give enough time for the other threads to finish.
+// long startOfWait = System.currentTimeMillis();
+// while (System.currentTimeMillis() - startOfWait < 30000) { // wait up to 30 seconds in case of machine stall
+//
+// // Attempt to insure we let the threads under test run in preference to this one
+// Thread.sleep(20); // imprecise. Intent is to allow other thread that was waiting, to run
+// // before this thread resumes. Depends on thread priorities, but
+// // multiple threads could be running at the same time.
+//
+// numberWaiting = 0;
+// for (int i = 0; i < numberOfThreads; i++) {
+// if (threads[i].state == MultiThreadUtils.THREAD_RUNNING) {
+// numberWaiting ++;
+// }
+// }
+// if (numberWaiting == 0) {
+// break;
+// }
+// }
- assertEquals(0, numberWaiting); // if not 0 by now, something is likely wrong, or machine stalled more than 30 seconds
+// assertEquals(0, numberWaiting); // if not 0 by now, something is likely wrong, or machine stalled more than 30 seconds
// System.out.format("JCasHashMapTest collide, found = %s%n", intList(found));
for (FeatureStructureImpl f : found) {
if (f != fs) {
@@ -282,6 +289,8 @@ public class JCasHashMapTest extends Tes
}
}
}
+
+ MultiThreadUtils.terminateThreads(threads);
}