uimaj-core/src/test/java/org/apache/uima/analysis_engine/impl/MultiprocessingAnalysisEngine_implTest.java - uima-uimaj - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 package org.apache.uima.analysis_engine.impl;

 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;

 import java.util.HashMap;
 import java.util.Map;
 import java.util.Random;

 import org.apache.uima.Constants;
 import org.apache.uima.UIMAException;
 import org.apache.uima.UIMAFramework;
 import org.apache.uima.UIMA_IllegalStateException;
 import org.apache.uima.ae.multiplier.SimpleCasGenerator;
 import org.apache.uima.analysis_engine.AnalysisEngine;
 import org.apache.uima.analysis_engine.AnalysisEngineDescription;
 import org.apache.uima.analysis_engine.ResultSpecification;
 import org.apache.uima.analysis_engine.metadata.AnalysisEngineMetaData;
 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.Misc;
 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;
 import org.apache.uima.resource.metadata.NameValuePair;
 import org.apache.uima.resource.metadata.TypeSystemDescription;
 import org.apache.uima.resource.metadata.impl.Capability_impl;
 import org.apache.uima.resource.metadata.impl.ConfigurationParameter_impl;
 import org.apache.uima.resource.metadata.impl.NameValuePair_impl;
 import org.apache.uima.resource.metadata.impl.TypeSystemDescription_impl;
 import org.apache.uima.test.junit_extension.JUnitExtension;
 import org.apache.uima.util.XMLInputSource;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;

 public class MultiprocessingAnalysisEngine_implTest {

   private final static boolean doSleeps = true; // running w/ dosleeps = false should show 100% cpu

   private AnalysisEngineDescription mSimpleDesc;

   private AnalysisEngineDescription mAggDesc;

   public volatile TypeSystem mLastTypeSystem;

   @BeforeEach
   public void setUp() throws Exception {
     try {
       mSimpleDesc = new AnalysisEngineDescription_impl();
       mSimpleDesc.setFrameworkImplementation(Constants.JAVA_FRAMEWORK_NAME);
       mSimpleDesc.setPrimitive(true);
       mSimpleDesc
               .setAnnotatorImplementationName("org.apache.uima.analysis_engine.impl.TestAnnotator");
       mSimpleDesc.getMetaData().setName("Simple Test");
       TypeSystemDescription typeSys = new TypeSystemDescription_impl();
       typeSys.addType("foo.Bar", "test", "uima.tcas.Annotation");
       typeSys.addType("NamedEntity", "test", "uima.tcas.Annotation");
       typeSys.addType("DocumentStructure", "test", "uima.tcas.Annotation");
       mSimpleDesc.getAnalysisEngineMetaData().setTypeSystem(typeSys);
       Capability cap = new Capability_impl();
       cap.addOutputType("NamedEntity", true);
       cap.addOutputType("DocumentStructure", true);
       Capability[] caps = new Capability[] { cap };
       mSimpleDesc.getAnalysisEngineMetaData().setCapabilities(caps);

       mAggDesc = new AnalysisEngineDescription_impl();
       mAggDesc.setPrimitive(false);
       mAggDesc.getMetaData().setName("Simple Test Aggregate");
       mAggDesc.getDelegateAnalysisEngineSpecifiersWithImports().put("Test", mSimpleDesc);
       FixedFlow_impl flow = new FixedFlow_impl();
       flow.setFixedFlow(new String[] { "Test" });
       mAggDesc.getAnalysisEngineMetaData().setFlowConstraints(flow);
       mAggDesc.getAnalysisEngineMetaData().setCapabilities(caps);
     } catch (Exception e) {
       JUnitExtension.handleException(e);
     }
   }

   @Test
   public void testInitialize() throws Exception {
     try {
       // initialize MultiprocesingTextAnalysisEngine
       MultiprocessingAnalysisEngine_impl mtae = new MultiprocessingAnalysisEngine_impl();
       boolean result = mtae.initialize(mSimpleDesc, null);
       assertThat(result).isTrue();

       // initialize again - should fail
       Exception ex = null;
       try {
         mtae.initialize(mSimpleDesc, null);
       } catch (UIMA_IllegalStateException e) {
         ex = e;
       }
       assertThat(ex).isNotNull();

       // initialize a new TAE with parameters
       Map<String, Object> map = new HashMap<>();
       map.put(AnalysisEngine.PARAM_NUM_SIMULTANEOUS_REQUESTS, 5);
       map.put(AnalysisEngine.PARAM_TIMEOUT_PERIOD, 60000);
       MultiprocessingAnalysisEngine_impl mtae2 = new MultiprocessingAnalysisEngine_impl();
       result = mtae2.initialize(mSimpleDesc, map);
       assertThat(result).isTrue();
       // check parameter values
       assertThat(mtae2.getPool().getSize()).isEqualTo(5);
       assertThat(mtae2.getTimeout()).isEqualTo(60000);
     } catch (Exception e) {
       JUnitExtension.handleException(e);
     }
   }

   @Test
   public void testGetAnalysisEngineMetaData() throws Exception {
     try {
       MultiprocessingAnalysisEngine_impl mtae = new MultiprocessingAnalysisEngine_impl();
       boolean result = mtae.initialize(mSimpleDesc, null);
       assertThat(result).isTrue();

       AnalysisEngineMetaData md = mtae.getAnalysisEngineMetaData();
       assertThat(md).isNotNull();
       assertThat(md.getName()).isEqualTo("Simple Test");
     } catch (Exception e) {
       JUnitExtension.handleException(e);
     }
   }

   @Test
   public void testNewCAS() throws Exception {
     try {
       MultiprocessingAnalysisEngine_impl mtae = new MultiprocessingAnalysisEngine_impl();
       boolean result = mtae.initialize(mSimpleDesc, null);
       assertThat(result).isTrue();

       CAS cas1 = mtae.newCAS();
       // should have the type foo.Bar
       assertNotNull(cas1.getTypeSystem().getType("foo.Bar"));

       // should be able to get as many as we want and they should all be different
       CAS cas2 = mtae.newCAS();
       assertThat(cas2).isNotNull();
       assertThat(cas1 != cas2).isTrue();
       CAS cas3 = mtae.newCAS();
       assertThat(cas3).isNotNull();
       assertThat(cas1 != cas3).isTrue();
       assertThat(cas2 != cas3).isTrue();
       CAS cas4 = mtae.newCAS();
       assertThat(cas4).isNotNull();
       assertThat(cas1 != cas4).isTrue();
       assertThat(cas2 != cas4).isTrue();
       assertThat(cas3 != cas4).isTrue();

       // try aggregate
       MultiprocessingAnalysisEngine_impl mtae2 = new MultiprocessingAnalysisEngine_impl();
       result = mtae2.initialize(mAggDesc, null);
       assertThat(result).isTrue();

       CAS cas5 = mtae2.newCAS();
       // should have the type foo.Bar
       assertNotNull(cas5.getTypeSystem().getType("foo.Bar"));
     } catch (Exception e) {
       JUnitExtension.handleException(e);
     }
   }

   @Test
   public void testProcess() throws Exception {
     try {
       // test simple primitive MultiprocessingTextAnalysisEngine
       // (using TestAnnotator class)
       _testProcess(mSimpleDesc, 0);

       // test simple aggregate MultiprocessingTextAnalysisEngine
       // (again using TestAnnotator class)

       _testProcess(mAggDesc, 0);

       /**
        * 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<>();
       params.put(AnalysisEngine.PARAM_NUM_SIMULTANEOUS_REQUESTS, 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 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;
             } else {
               fail(failure.getMessage());
             }
           }
         }
       }

       // Check TestAnnotator fields only at the very end of processing,
       // we can't test from the threads themselves since the state of
       // these fields is nondeterministic during the multithreaded processing.
       assertEquals("testing...", TestAnnotator.getLastDocument());
       ResultSpecification lastResultSpec = TestAnnotator.getLastResultSpec();
       ResultSpecification resultSpec = new ResultSpecification_impl(lastResultSpec.getTypeSystem());
       resultSpec.addResultType("NamedEntity", true);
       assertEquals(resultSpec, lastResultSpec);
       MultiThreadUtils.terminateThreads(threads);
     } catch (Exception e) {
       JUnitExtension.handleException(e);
     }
   }

   @Test
   public void testProcessManyCM() throws Exception {
     // get Resource Specifier from XML file
     XMLInputSource in = new XMLInputSource("src/test/resources/ExampleTae/SimpleCasGenerator.xml");
     ResourceSpecifier specifier = UIMAFramework.getXMLParser().parseResourceSpecifier(in);
     for (int i = 0; i < 10; i++) {
       processMany(specifier);
     }
   }

   @Test
   public void testProcessManyAgg() throws Exception {
     // get Resource Specifier from XML file
     XMLInputSource in = new XMLInputSource("src/test/resources/ExampleTae/SimpleTestAggregate.xml");
     ResourceSpecifier specifier = UIMAFramework.getXMLParser().parseResourceSpecifier(in);
     for (int i = 0; i < 10; i++) {
       processMany(specifier);
     }
   }

   // rename to run this in a loop
   public void tstLoopProcessManyAgg() throws Exception {
     XMLInputSource in = new XMLInputSource("src/test/resources/ExampleTae/SimpleTestAggregate.xml");
     ResourceSpecifier specifier = UIMAFramework.getXMLParser().parseResourceSpecifier(in);
     Misc.timeLoops("ProcessManyAgg", 1000, () -> processMany(specifier));
   }

   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 {

       // multiple threads!
       MultiprocessingAnalysisEngine_impl ae = new MultiprocessingAnalysisEngine_impl();
       Map<String, Object> params = new HashMap<>();
       params.put(AnalysisEngine.PARAM_NUM_SIMULTANEOUS_REQUESTS, NUM_INSTANCES);

       ae.initialize(specifier, params);

       ProcessThreadMany[] threads = new ProcessThreadMany[NUM_THREADS];
       for (int i = 0; i < NUM_THREADS; i++) {
         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);

       for (ProcessThreadMany ptm : threads) {

         Throwable failure = ptm.getFailure();
         if (failure != null) {
           if (failure instanceof Exception) {
             throw (Exception) failure;
           } else {
             fail(failure.getMessage());
           }
         }
       }

       // Check TestAnnotator fields only at the very end of processing,
       // we can't test from the threads themselves since the state of
       // these fields is nondeterministic during the multithreaded processing.
       assertEquals("testing...", SimpleCasGenerator.getLastDocument());
       ResultSpecification lastResultSpec = SimpleCasGenerator.getLastResultSpec();
       ResultSpecification resultSpec = new ResultSpecification_impl(lastResultSpec.getTypeSystem());
       resultSpec.addResultType("NamedEntity", true);
       assertEquals(resultSpec, lastResultSpec);
       MultiThreadUtils.terminateThreads(threads);
     } catch (Exception e) {
       JUnitExtension.handleException(e);
     }
   }

   @Test
   public void testReconfigure() throws Exception {
     try {
       // create simple primitive TextAnalysisEngine descriptor (using TestAnnotator class)
       AnalysisEngineDescription primitiveDesc = new AnalysisEngineDescription_impl();
       primitiveDesc.setPrimitive(true);
       primitiveDesc
               .setAnnotatorImplementationName("org.apache.uima.analysis_engine.impl.TestAnnotator");
       primitiveDesc.getMetaData().setName("Reconfigure Test 1");
       ConfigurationParameter p1 = new ConfigurationParameter_impl();
       p1.setName("StringParam");
       p1.setDescription("parameter with String data type");
       p1.setType(ConfigurationParameter.TYPE_STRING);
       primitiveDesc.getMetaData().getConfigurationParameterDeclarations()
               .setConfigurationParameters(new ConfigurationParameter[] { p1 });
       primitiveDesc.getMetaData().getConfigurationParameterSettings().setParameterSettings(
               new NameValuePair[] { new NameValuePair_impl("StringParam", "Test1") });

       // instantiate MultiprocessingTextAnalysisEngine
       MultiprocessingAnalysisEngine_impl tae = new MultiprocessingAnalysisEngine_impl();
       tae.initialize(primitiveDesc, null);

       // check value of string param (TestAnnotator saves it in a static field)
       assertEquals("Test1", TestAnnotator.stringParamValue);

       // reconfigure
       tae.setConfigParameterValue("StringParam", "Test2");
       tae.reconfigure();

       // test again
       assertEquals("Test2", TestAnnotator.stringParamValue);

       // test aggregate TAE
       AnalysisEngineDescription aggDesc = new AnalysisEngineDescription_impl();
       aggDesc.setFrameworkImplementation(Constants.JAVA_FRAMEWORK_NAME);
       aggDesc.setPrimitive(false);
       aggDesc.getDelegateAnalysisEngineSpecifiersWithImports().put("Test", primitiveDesc);
       aggDesc.getMetaData().setName("Reconfigure Test 2");
       FixedFlow_impl flow = new FixedFlow_impl();
       flow.setFixedFlow(new String[] { "Test" });
       aggDesc.getAnalysisEngineMetaData().setFlowConstraints(flow);
       ConfigurationParameter p2 = new ConfigurationParameter_impl();
       p2.setName("StringParam");
       p2.setDescription("parameter with String data type");
       p2.setType(ConfigurationParameter.TYPE_STRING);
       p2.setOverrides(new String[] { "Test/StringParam" });
       aggDesc.getMetaData().getConfigurationParameterDeclarations()
               .setConfigurationParameters(new ConfigurationParameter[] { p2 });
       aggDesc.getMetaData().getConfigurationParameterSettings().setParameterSettings(
               new NameValuePair[] { new NameValuePair_impl("StringParam", "Test3") });
       // instantiate TextAnalysisEngine
       MultiprocessingAnalysisEngine_impl aggTae = new MultiprocessingAnalysisEngine_impl();
       aggTae.initialize(aggDesc, null);

       assertEquals("Test3", TestAnnotator.stringParamValue);

       // reconfigure
       aggTae.setConfigParameterValue("StringParam", "Test4");
       aggTae.reconfigure();

       // test again
       assertEquals("Test4", TestAnnotator.stringParamValue);

       // reconfigure WITHOUT setting that parameter
       aggTae.reconfigure();
       // test again
       assertEquals("Test4", TestAnnotator.stringParamValue);
     } catch (Exception e) {
       JUnitExtension.handleException(e);
     }
   }

   /**
    * Auxilliary method used by testProcess()
    *
    * @param aTaeDesc
    *          description of TextAnalysisEngine to test
    * @param i
    *          thread identifier for multithreaded testing
    */
   protected void _testProcess(AnalysisEngineDescription aTaeDesc, int i) throws UIMAException {
     // create and initialize MultiprocessingTextAnalysisEngine
     MultiprocessingAnalysisEngine_impl tae = new MultiprocessingAnalysisEngine_impl();
     tae.initialize(aTaeDesc, null);

     // 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.

     // process(CAS)
     CAS tcas = tae.newCAS();
     tcas.setDocumentText("new test");
     tae.process(tcas);
     assertEquals("new test", TestAnnotator.lastDocument);
     tcas.reset();

     // process(CAS,ResultSpecification)
     ResultSpecification resultSpec = new ResultSpecification_impl(tcas.getTypeSystem());
     resultSpec.addResultType("NamedEntity", true);

     tcas.setDocumentText("testing...");
     tae.process(tcas, resultSpec);
     assertEquals("testing...", TestAnnotator.lastDocument);
     assertEquals(resultSpec, TestAnnotator.lastResultSpec);
     tcas.reset();
   }

   class ProcessThread extends ThreadM {

     Throwable mFailure = null;

     AnalysisEngine mAE;

     ProcessThread(AnalysisEngine aAE) {
       mAE = aAE;
     }

     @Override
     public void run() {
       Random r = new Random();
       while (true) {

         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;
     }

     public void resetFailure() {
       mFailure = null;
     }
   }

   class ProcessThreadMany extends ProcessThread {

     ProcessThreadMany(AnalysisEngine aAE) {
       super(aAE);
     }

     @Override
     public void run() {

       while (true) {

         if (!MultiThreadUtils.wait4go(this)) {
           break;
         }

         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;
         }
       }
     }

     @Override
     public synchronized Throwable getFailure() {
       return mFailure;
     }

   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	package org.apache.uima.analysis_engine.impl;

	import static org.assertj.core.api.Assertions.assertThat;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertNotNull;
	import static org.junit.Assert.fail;

	import java.util.HashMap;
	import java.util.Map;
	import java.util.Random;

	import org.apache.uima.Constants;
	import org.apache.uima.UIMAException;
	import org.apache.uima.UIMAFramework;
	import org.apache.uima.UIMA_IllegalStateException;
	import org.apache.uima.ae.multiplier.SimpleCasGenerator;
	import org.apache.uima.analysis_engine.AnalysisEngine;
	import org.apache.uima.analysis_engine.AnalysisEngineDescription;
	import org.apache.uima.analysis_engine.ResultSpecification;
	import org.apache.uima.analysis_engine.metadata.AnalysisEngineMetaData;
	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.Misc;
	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;
	import org.apache.uima.resource.metadata.NameValuePair;
	import org.apache.uima.resource.metadata.TypeSystemDescription;
	import org.apache.uima.resource.metadata.impl.Capability_impl;
	import org.apache.uima.resource.metadata.impl.ConfigurationParameter_impl;
	import org.apache.uima.resource.metadata.impl.NameValuePair_impl;
	import org.apache.uima.resource.metadata.impl.TypeSystemDescription_impl;
	import org.apache.uima.test.junit_extension.JUnitExtension;
	import org.apache.uima.util.XMLInputSource;
	import org.junit.jupiter.api.BeforeEach;
	import org.junit.jupiter.api.Test;

	public class MultiprocessingAnalysisEngine_implTest {

	private final static boolean doSleeps = true; // running w/ dosleeps = false should show 100% cpu

	private AnalysisEngineDescription mSimpleDesc;

	private AnalysisEngineDescription mAggDesc;

	public volatile TypeSystem mLastTypeSystem;

	@BeforeEach
	public void setUp() throws Exception {
	try {
	mSimpleDesc = new AnalysisEngineDescription_impl();
	mSimpleDesc.setFrameworkImplementation(Constants.JAVA_FRAMEWORK_NAME);
	mSimpleDesc.setPrimitive(true);
	mSimpleDesc
	.setAnnotatorImplementationName("org.apache.uima.analysis_engine.impl.TestAnnotator");
	mSimpleDesc.getMetaData().setName("Simple Test");
	TypeSystemDescription typeSys = new TypeSystemDescription_impl();
	typeSys.addType("foo.Bar", "test", "uima.tcas.Annotation");
	typeSys.addType("NamedEntity", "test", "uima.tcas.Annotation");
	typeSys.addType("DocumentStructure", "test", "uima.tcas.Annotation");
	mSimpleDesc.getAnalysisEngineMetaData().setTypeSystem(typeSys);
	Capability cap = new Capability_impl();
	cap.addOutputType("NamedEntity", true);
	cap.addOutputType("DocumentStructure", true);
	Capability[] caps = new Capability[] { cap };
	mSimpleDesc.getAnalysisEngineMetaData().setCapabilities(caps);

	mAggDesc = new AnalysisEngineDescription_impl();
	mAggDesc.setPrimitive(false);
	mAggDesc.getMetaData().setName("Simple Test Aggregate");
	mAggDesc.getDelegateAnalysisEngineSpecifiersWithImports().put("Test", mSimpleDesc);
	FixedFlow_impl flow = new FixedFlow_impl();
	flow.setFixedFlow(new String[] { "Test" });
	mAggDesc.getAnalysisEngineMetaData().setFlowConstraints(flow);
	mAggDesc.getAnalysisEngineMetaData().setCapabilities(caps);
	} catch (Exception e) {
	JUnitExtension.handleException(e);
	}
	}

	@Test
	public void testInitialize() throws Exception {
	try {
	// initialize MultiprocesingTextAnalysisEngine
	MultiprocessingAnalysisEngine_impl mtae = new MultiprocessingAnalysisEngine_impl();
	boolean result = mtae.initialize(mSimpleDesc, null);
	assertThat(result).isTrue();

	// initialize again - should fail
	Exception ex = null;
	try {
	mtae.initialize(mSimpleDesc, null);
	} catch (UIMA_IllegalStateException e) {
	ex = e;
	}
	assertThat(ex).isNotNull();

	// initialize a new TAE with parameters
	Map<String, Object> map = new HashMap<>();
	map.put(AnalysisEngine.PARAM_NUM_SIMULTANEOUS_REQUESTS, 5);
	map.put(AnalysisEngine.PARAM_TIMEOUT_PERIOD, 60000);
	MultiprocessingAnalysisEngine_impl mtae2 = new MultiprocessingAnalysisEngine_impl();
	result = mtae2.initialize(mSimpleDesc, map);
	assertThat(result).isTrue();
	// check parameter values
	assertThat(mtae2.getPool().getSize()).isEqualTo(5);
	assertThat(mtae2.getTimeout()).isEqualTo(60000);
	} catch (Exception e) {
	JUnitExtension.handleException(e);
	}
	}

	@Test
	public void testGetAnalysisEngineMetaData() throws Exception {
	try {
	MultiprocessingAnalysisEngine_impl mtae = new MultiprocessingAnalysisEngine_impl();
	boolean result = mtae.initialize(mSimpleDesc, null);
	assertThat(result).isTrue();

	AnalysisEngineMetaData md = mtae.getAnalysisEngineMetaData();
	assertThat(md).isNotNull();
	assertThat(md.getName()).isEqualTo("Simple Test");
	} catch (Exception e) {
	JUnitExtension.handleException(e);
	}
	}

	@Test
	public void testNewCAS() throws Exception {
	try {
	MultiprocessingAnalysisEngine_impl mtae = new MultiprocessingAnalysisEngine_impl();
	boolean result = mtae.initialize(mSimpleDesc, null);
	assertThat(result).isTrue();

	CAS cas1 = mtae.newCAS();
	// should have the type foo.Bar
	assertNotNull(cas1.getTypeSystem().getType("foo.Bar"));

	// should be able to get as many as we want and they should all be different
	CAS cas2 = mtae.newCAS();
	assertThat(cas2).isNotNull();
	assertThat(cas1 != cas2).isTrue();
	CAS cas3 = mtae.newCAS();
	assertThat(cas3).isNotNull();
	assertThat(cas1 != cas3).isTrue();
	assertThat(cas2 != cas3).isTrue();
	CAS cas4 = mtae.newCAS();
	assertThat(cas4).isNotNull();
	assertThat(cas1 != cas4).isTrue();
	assertThat(cas2 != cas4).isTrue();
	assertThat(cas3 != cas4).isTrue();

	// try aggregate
	MultiprocessingAnalysisEngine_impl mtae2 = new MultiprocessingAnalysisEngine_impl();
	result = mtae2.initialize(mAggDesc, null);
	assertThat(result).isTrue();

	CAS cas5 = mtae2.newCAS();
	// should have the type foo.Bar
	assertNotNull(cas5.getTypeSystem().getType("foo.Bar"));
	} catch (Exception e) {
	JUnitExtension.handleException(e);
	}
	}

	@Test
	public void testProcess() throws Exception {
	try {
	// test simple primitive MultiprocessingTextAnalysisEngine
	// (using TestAnnotator class)
	_testProcess(mSimpleDesc, 0);

	// test simple aggregate MultiprocessingTextAnalysisEngine
	// (again using TestAnnotator class)

	_testProcess(mAggDesc, 0);

	/**
	* 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<>();
	params.put(AnalysisEngine.PARAM_NUM_SIMULTANEOUS_REQUESTS, 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 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;
	} else {
	fail(failure.getMessage());
	}
	}
	}
	}

	// Check TestAnnotator fields only at the very end of processing,
	// we can't test from the threads themselves since the state of
	// these fields is nondeterministic during the multithreaded processing.
	assertEquals("testing...", TestAnnotator.getLastDocument());
	ResultSpecification lastResultSpec = TestAnnotator.getLastResultSpec();
	ResultSpecification resultSpec = new ResultSpecification_impl(lastResultSpec.getTypeSystem());
	resultSpec.addResultType("NamedEntity", true);
	assertEquals(resultSpec, lastResultSpec);
	MultiThreadUtils.terminateThreads(threads);
	} catch (Exception e) {
	JUnitExtension.handleException(e);
	}
	}

	@Test
	public void testProcessManyCM() throws Exception {
	// get Resource Specifier from XML file
	XMLInputSource in = new XMLInputSource("src/test/resources/ExampleTae/SimpleCasGenerator.xml");
	ResourceSpecifier specifier = UIMAFramework.getXMLParser().parseResourceSpecifier(in);
	for (int i = 0; i < 10; i++) {
	processMany(specifier);
	}
	}

	@Test
	public void testProcessManyAgg() throws Exception {
	// get Resource Specifier from XML file
	XMLInputSource in = new XMLInputSource("src/test/resources/ExampleTae/SimpleTestAggregate.xml");
	ResourceSpecifier specifier = UIMAFramework.getXMLParser().parseResourceSpecifier(in);
	for (int i = 0; i < 10; i++) {
	processMany(specifier);
	}
	}

	// rename to run this in a loop
	public void tstLoopProcessManyAgg() throws Exception {
	XMLInputSource in = new XMLInputSource("src/test/resources/ExampleTae/SimpleTestAggregate.xml");
	ResourceSpecifier specifier = UIMAFramework.getXMLParser().parseResourceSpecifier(in);
	Misc.timeLoops("ProcessManyAgg", 1000, () -> processMany(specifier));
	}

	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 {

	// multiple threads!
	MultiprocessingAnalysisEngine_impl ae = new MultiprocessingAnalysisEngine_impl();
	Map<String, Object> params = new HashMap<>();
	params.put(AnalysisEngine.PARAM_NUM_SIMULTANEOUS_REQUESTS, NUM_INSTANCES);

	ae.initialize(specifier, params);

	ProcessThreadMany[] threads = new ProcessThreadMany[NUM_THREADS];
	for (int i = 0; i < NUM_THREADS; i++) {
	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);

	for (ProcessThreadMany ptm : threads) {

	Throwable failure = ptm.getFailure();
	if (failure != null) {
	if (failure instanceof Exception) {
	throw (Exception) failure;
	} else {
	fail(failure.getMessage());
	}
	}
	}

	// Check TestAnnotator fields only at the very end of processing,
	// we can't test from the threads themselves since the state of
	// these fields is nondeterministic during the multithreaded processing.
	assertEquals("testing...", SimpleCasGenerator.getLastDocument());
	ResultSpecification lastResultSpec = SimpleCasGenerator.getLastResultSpec();
	ResultSpecification resultSpec = new ResultSpecification_impl(lastResultSpec.getTypeSystem());
	resultSpec.addResultType("NamedEntity", true);
	assertEquals(resultSpec, lastResultSpec);
	MultiThreadUtils.terminateThreads(threads);
	} catch (Exception e) {
	JUnitExtension.handleException(e);
	}
	}

	@Test
	public void testReconfigure() throws Exception {
	try {
	// create simple primitive TextAnalysisEngine descriptor (using TestAnnotator class)
	AnalysisEngineDescription primitiveDesc = new AnalysisEngineDescription_impl();
	primitiveDesc.setPrimitive(true);
	primitiveDesc
	.setAnnotatorImplementationName("org.apache.uima.analysis_engine.impl.TestAnnotator");
	primitiveDesc.getMetaData().setName("Reconfigure Test 1");
	ConfigurationParameter p1 = new ConfigurationParameter_impl();
	p1.setName("StringParam");
	p1.setDescription("parameter with String data type");
	p1.setType(ConfigurationParameter.TYPE_STRING);
	primitiveDesc.getMetaData().getConfigurationParameterDeclarations()
	.setConfigurationParameters(new ConfigurationParameter[] { p1 });
	primitiveDesc.getMetaData().getConfigurationParameterSettings().setParameterSettings(
	new NameValuePair[] { new NameValuePair_impl("StringParam", "Test1") });

	// instantiate MultiprocessingTextAnalysisEngine
	MultiprocessingAnalysisEngine_impl tae = new MultiprocessingAnalysisEngine_impl();
	tae.initialize(primitiveDesc, null);

	// check value of string param (TestAnnotator saves it in a static field)
	assertEquals("Test1", TestAnnotator.stringParamValue);

	// reconfigure
	tae.setConfigParameterValue("StringParam", "Test2");
	tae.reconfigure();

	// test again
	assertEquals("Test2", TestAnnotator.stringParamValue);

	// test aggregate TAE
	AnalysisEngineDescription aggDesc = new AnalysisEngineDescription_impl();
	aggDesc.setFrameworkImplementation(Constants.JAVA_FRAMEWORK_NAME);
	aggDesc.setPrimitive(false);
	aggDesc.getDelegateAnalysisEngineSpecifiersWithImports().put("Test", primitiveDesc);
	aggDesc.getMetaData().setName("Reconfigure Test 2");
	FixedFlow_impl flow = new FixedFlow_impl();
	flow.setFixedFlow(new String[] { "Test" });
	aggDesc.getAnalysisEngineMetaData().setFlowConstraints(flow);
	ConfigurationParameter p2 = new ConfigurationParameter_impl();
	p2.setName("StringParam");
	p2.setDescription("parameter with String data type");
	p2.setType(ConfigurationParameter.TYPE_STRING);
	p2.setOverrides(new String[] { "Test/StringParam" });
	aggDesc.getMetaData().getConfigurationParameterDeclarations()
	.setConfigurationParameters(new ConfigurationParameter[] { p2 });
	aggDesc.getMetaData().getConfigurationParameterSettings().setParameterSettings(
	new NameValuePair[] { new NameValuePair_impl("StringParam", "Test3") });
	// instantiate TextAnalysisEngine
	MultiprocessingAnalysisEngine_impl aggTae = new MultiprocessingAnalysisEngine_impl();
	aggTae.initialize(aggDesc, null);

	assertEquals("Test3", TestAnnotator.stringParamValue);

	// reconfigure
	aggTae.setConfigParameterValue("StringParam", "Test4");
	aggTae.reconfigure();

	// test again
	assertEquals("Test4", TestAnnotator.stringParamValue);

	// reconfigure WITHOUT setting that parameter
	aggTae.reconfigure();
	// test again
	assertEquals("Test4", TestAnnotator.stringParamValue);
	} catch (Exception e) {
	JUnitExtension.handleException(e);
	}
	}

	/**
	* Auxilliary method used by testProcess()
	*
	* @param aTaeDesc
	* description of TextAnalysisEngine to test
	* @param i
	* thread identifier for multithreaded testing
	*/
	protected void _testProcess(AnalysisEngineDescription aTaeDesc, int i) throws UIMAException {
	// create and initialize MultiprocessingTextAnalysisEngine
	MultiprocessingAnalysisEngine_impl tae = new MultiprocessingAnalysisEngine_impl();
	tae.initialize(aTaeDesc, null);

	// 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.

	// process(CAS)
	CAS tcas = tae.newCAS();
	tcas.setDocumentText("new test");
	tae.process(tcas);
	assertEquals("new test", TestAnnotator.lastDocument);
	tcas.reset();

	// process(CAS,ResultSpecification)
	ResultSpecification resultSpec = new ResultSpecification_impl(tcas.getTypeSystem());
	resultSpec.addResultType("NamedEntity", true);

	tcas.setDocumentText("testing...");
	tae.process(tcas, resultSpec);
	assertEquals("testing...", TestAnnotator.lastDocument);
	assertEquals(resultSpec, TestAnnotator.lastResultSpec);
	tcas.reset();
	}

	class ProcessThread extends ThreadM {

	Throwable mFailure = null;

	AnalysisEngine mAE;

	ProcessThread(AnalysisEngine aAE) {
	mAE = aAE;
	}

	@Override
	public void run() {
	Random r = new Random();
	while (true) {

	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;
	}

	public void resetFailure() {
	mFailure = null;
	}
	}

	class ProcessThreadMany extends ProcessThread {

	ProcessThreadMany(AnalysisEngine aAE) {
	super(aAE);
	}

	@Override
	public void run() {

	while (true) {

	if (!MultiThreadUtils.wait4go(this)) {
	break;
	}

	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;
	}
	}
	}

	@Override
	public synchronized Throwable getFailure() {
	return mFailure;
	}

	}

	}