Support for Aggregate CAS Multiplier

src/cas/cas.cpp

@@ -179,6 +179,7 @@ namespace uima {
       iv_indexRepository(NULL),
       iv_filterBuilder(NULL),
       iv_componentInfo(NULL),
+      iv_owner(NULL),
       iv_utDocumentType(uima::lowlevel::TypeSystem::INVALID_TYPE),
       iv_utDocumentLangAsIntFeat(uima::lowlevel::TypeSystem::INVALID_FEATURE),
       iv_utDocumentLangAsStrFeat(uima::lowlevel::TypeSystem::INVALID_FEATURE),
@@ -221,6 +222,7 @@ namespace uima {
     iv_typeSystem = inCas->iv_typeSystem;
     iv_heap = inCas->iv_heap;
     iv_componentInfo = inCas->iv_componentInfo;
+    iv_owner = inCas->iv_owner;
     iv_utDocumentLangAsIntFeat = uima::lowlevel::TypeSystem::INVALID_FEATURE;
     iv_utDocumentLangAsStrFeat = uima::lowlevel::TypeSystem::INVALID_FEATURE;
     refreshCachedTypes();
@@ -262,6 +264,7 @@ namespace uima {
       iv_sofaCount(0),
       initialSofaCreated(false),
       iv_initialView(NULL),
+      iv_owner(NULL),
       iv_indexRepository(NULL),
       iv_filterBuilder(NULL),
       iv_componentInfo(NULL),
@@ -676,6 +679,10 @@ namespace uima {
                       );
   }
 
+  void CAS::setOwner(AnnotatorContext *owner) {
+    iv_baseCas->iv_owner = owner;
+  }
+
   // deprecated version
   void CAS::setDocumentText(UChar const * cpDocument, size_t uiLength, bool bCopyToCAS ) {
     if (cpDocument == NULL) {
@@ -809,6 +816,21 @@ namespace uima {
     }
   }
 
+  void CAS::release() {
+    if (iv_baseCas->iv_owner) {
+      iv_baseCas->iv_owner->releaseCAS(*this);
+    } else {
+      ErrorMessage msg(UIMA_MSG_ID_EXC_INVALID_CAS_RELEASE);
+      msg.addParam("This CAS does not have any owner");
+      UIMA_EXC_THROW_NEW(CASException,
+                         UIMA_ERR_CAS_RELEASE,
+                         msg,
+                         ErrorMessage(UIMA_MSG_ID_EXCON_UNKNOWN_CONTEXT),
+                         ErrorInfo::recoverable
+                        );
+    }
+  }
+
   ANIndex CAS::getAnnotationIndex(Type const & crType) {
     if (isbaseCas) {
       assertWithMsg(false, "Annotation Index does not exist in Base CAS!");

src/cas/uima/cas.hpp

@@ -191,6 +191,8 @@ namespace uima {
     void bumpSofaCount();	
     void invalidBaseCasMethod();
 
+    /** Set the owner of the base CAS */
+    void setOwner(AnnotatorContext* owner);
     void registerView(SofaFS);
     void updateDocumentAnnotation( );
     void copyDocumentString(UnicodeStringRef);
@@ -231,6 +233,7 @@ namespace uima {
     bool initialSofaCreated;
     bool isDeletingViews;  //set this flag to true when destroying CAS
     AnnotatorContext *iv_componentInfo;
+    AnnotatorContext *iv_owner;
 
     uima::lowlevel::TyFSType     iv_utDocumentType;
     uima::lowlevel::TyFSFeature  iv_utDocumentLangAsIntFeat;
@@ -772,6 +775,13 @@ namespace uima {
     icu::UnicodeString getAnnotationIndexID() const {
       return CAS::INDEXID_ANNOTATION;
     }
+
+    /**
+     * When called this CAS will release itself by calling releaseCas on the AnnotatorContext that owns it.
+     * NOTE: This only works for CASes that have an owner, ie. belong to a CASPool.
+     */
+    void release();
+
     /** @} */
 
     /** @defgroup PreDefTypes Predefined Types

src/framework/Makefile.am

@@ -78,7 +78,9 @@ libuima_la_SOURCES += consoleui.cpp
 libuima_la_SOURCES += cp2ucnvrt.cpp					   
 libuima_la_SOURCES += dottypesystemwriter.cpp				   
 libuima_la_SOURCES += engine.cpp					   
-libuima_la_SOURCES += exceptions.cpp					   
+libuima_la_SOURCES += exceptions.cpp
+libuima_la_SOURCES += flow_controller.cpp
+libuima_la_SOURCES += flow.cpp
 libuima_la_SOURCES += ftools.cpp					   
 libuima_la_SOURCES += internal_aggregate_engine.cpp			   
 libuima_la_SOURCES += internal_capability_container.cpp		   

src/framework/annotator_context.cpp

@@ -165,7 +165,11 @@ namespace uima {
   }
 
   TyErrorId AnnotatorContext::defineCASPool(size_t numInstances) {
-    iv_pCasPool = new CASPool(getTaeSpecifier(),numInstances);
+    if (iv_pParentAnC) {            // If this is a delegate
+      iv_pCasPool = new CASPool(this, iv_pParentAnC->getTaeSpecifier(),numInstances);
+    } else {
+      iv_pCasPool = new CASPool(this, getTaeSpecifier(), numInstances);
+    }
     if (iv_pCasPool == NULL) {
       return UIMA_ERR_USER_ANNOTATOR_OUT_OF_MEMORY;
     }
@@ -500,7 +504,7 @@ namespace uima {
   NameValuePair const * AnnotatorContext::findNameValuePair(const icu::UnicodeString & paramName,
       const icu::UnicodeString & ancKey) const {
     /*       return findNameValuePair(getGroupNameWhenNotSpec(), paramName, iv_pTaeSpecifier->getSearchStrategy()); */
-    NameValuePair const * pValueLocal = pValueLocal = iv_pTaeSpecifier->getNameValuePair(paramName, ancKey);
+    NameValuePair const * pValueLocal = iv_pTaeSpecifier->getNameValuePair(paramName, ancKey);
 
     // the request was invalid we got an exception we leave to others to catch
 

src/framework/annotator_mgr.cpp

@@ -103,6 +103,8 @@ namespace uima {
         launchDeInit();
       }
       assert( iv_vecEntries.empty() );
+      if (iv_pFlowController)
+        delete iv_pFlowController;
 
     }
 
@@ -152,12 +154,21 @@ namespace uima {
       assert(iv_vecEntries.empty());
 
       AnnotatorContext & rANC = iv_pEngine->getAnnotatorContext();
-
       AnalysisEngineDescription const & crTAESpecifier = rANC.getTaeSpecifier(); // this method must be added
-
+      const AnalysisEngineMetaData* pEngineMetadata = crTAESpecifier.getAnalysisEngineMetaData();
       assert( ! crTAESpecifier.isPrimitive() );
-      //BSIvector < icu::UnicodeString > const & crVecEngineNames = crTAESpecifier.getAnalysisEngineMetaData()->getFixedFlow()->getNodes();
-      vector < icu::UnicodeString > const & crVecEngineNames = crTAESpecifier.getAnalysisEngineMetaData()->getFlowConstraints()->getNodes();
+
+      // FIXME: This shouldn't have been necessary since FlowContrainst::getFlowContraintsType
+      // should have been const in the first place
+      auto flowContraints = CONST_CAST(FlowConstraints *, pEngineMetadata->getFlowConstraints());
+      if (flowContraints->getFlowConstraintsType() == FlowConstraints::FIXED) {
+        iv_pFlowController = new FixedFlowController;
+        iv_pFlowController->initialize(rANC);
+      }
+
+      if (const OperationalProperties* operationalProps = pEngineMetadata->getOperationalProperties())
+        iv_bOutputNewCases = operationalProps->getOutputsNewCASes();
+      vector < icu::UnicodeString > const & crVecEngineNames = flowContraints->getNodes();
 
       // for all engines in the flow
       size_t ui;
@@ -431,11 +442,7 @@ namespace uima {
       return bHasTOF;
     }
 
-
-
-
-
-    TyErrorId AnnotatorManager::launchProcessDocument(CAS & cas, ResultSpecification const & crResultSpec) {
+    TyErrorId AnnotatorManager::processCapabilityLanguageFlow(CAS &cas, ResultSpecification const &crResultSpec) {
       /*
       This works as follows:
       The passes result spec is copied and for each delegate AE, it is determined
@@ -486,27 +493,13 @@ namespace uima {
         bool requiresTCas=true;
 
         if (cas.isBackwardCompatibleCas()) {
-	  tcas = &cas;
-		}
-          //this populates the tofsToBeRemoved vector so always call it
-          callEngine = shouldEngineBeCalled(*pCapContainer,
-                                            resSpec,
-                                            cas.getDocumentAnnotation().getLanguage(),
-                                            tofsToBeRemoved);
-          //check the FlowConstraintType specified in the aggregate engine
-          //if CapabilityLanguageFlow whether engine is called is
-          //determined by shouldEngineBeCalled()
-          AnnotatorContext & rANC = iv_pEngine->getAnnotatorContext();
-          AnalysisEngineDescription const & crTAESpecifier = rANC.getTaeSpecifier();
-          FlowConstraints const * pFlow = crTAESpecifier.getAnalysisEngineMetaData()->getFlowConstraints();
-          FlowConstraints * flow = CONST_CAST(FlowConstraints *, pFlow);
-          FlowConstraints::EnFlowType flowType = flow->getFlowConstraintsType();
-
-          //if FixedFlow specified all engines are always called so reset callEngine is true
-          if (flowType == FlowConstraints::FIXED) {
-            callEngine=true;
-          }
-
+          tcas = &cas;
+        }
+        //this populates the tofsToBeRemoved vector so always call it
+        callEngine = shouldEngineBeCalled(*pCapContainer,
+                                          resSpec,
+                                          cas.getDocumentAnnotation().getLanguage(),
+                                          tofsToBeRemoved);
 
         if ( callEngine ) {
 
@@ -583,7 +576,186 @@ namespace uima {
       }
 
       UIMA_ANNOTATOR_TIMING(iv_clTimerLaunchProcess.stop());
-      return(utRetVal);
+      return utRetVal;
+    }
+
+    CAS *AnnotatorManager::processUntilNextOutputCas() {
+      unique_ptr<Flow> flow{};
+      while (true) {
+        CAS *currentCas = nullptr;
+        Step nextStep;
+        flow = nullptr;
+
+        // get a cas from the stack
+        if (casIterStack.empty())
+          return nullptr;
+
+        StackFrame &frame = casIterStack.top();
+        try {
+          if (frame.casMultiplier && frame.casMultiplier->hasNext()) {
+            currentCas = &frame.casMultiplier->next();
+            // compute flow for newly produced CAS
+            flow = frame.originalFlow->newCasProduced(*currentCas, frame.lastEngineKey);
+          }
+        } catch (Exception &exception) {
+          if (!frame.originalFlow->continueOnFailure(frame.lastEngineKey /* ,exception */))
+            throw;
+        }
+
+        if (!currentCas) {
+          // if there is no more output CASes from the stack, take the original CAS that was processed by
+          // the CAS Multiplier and continue with its flow
+          currentCas = frame.originalCas;
+          flow = std::move(frame.originalFlow);
+          currentCas->setCurrentComponentInfo(nullptr);
+          casIterStack.pop();
+        }
+
+        activeCASes.insert(currentCas);
+
+        if (nextStep.getType() == Step::StepType::UNSPECIFIED) {
+          nextStep = flow->next(); // get the next step for the current flow
+        }
+
+        while (nextStep.getType() != Step::StepType::FINALSTEP) {
+          if (nextStep.getType() == Step::StepType::SIMPLESTEP) {
+            // find the AE specified by the step
+            const icu::UnicodeString &nextAEKey = nextStep.getSimpleStep()->getEngineName();
+            auto it = std::find_if(iv_vecEntries.begin(), iv_vecEntries.end(),
+                                   [&, nextAEKey](const EngineEntry &entry) {
+                                     return entry.iv_pEngine->getAnnotatorContext().iv_AnCKey == nextAEKey;
+                                   });
+
+            if (it != iv_vecEntries.end()) {
+              AnalysisEngine *nextAE = it->iv_pEngine;
+              CAS *outputCas = nullptr;
+
+              // call process one the AE and see if it has produced a new CAS
+              try {
+                CASIterator casIter = nextAE->processAndOutputNewCASes(*currentCas);
+                if (casIter.hasNext())
+                  outputCas = &casIter.next();
+              } catch (Exception &e) {
+                if (!flow->continueOnFailure(nextAEKey))
+                  throw;
+              }
+
+              if (outputCas) {
+                // new CAS is output so put the current components on the stack so we can process
+                // the other output CASes and original CASes later
+                std::unique_ptr<Flow> nextFlow = flow->newCasProduced(*outputCas, nextAEKey);
+                casIterStack.push({nextAE, currentCas, std::move(flow), nextAEKey});
+                flow = std::move(nextFlow);
+                currentCas = outputCas;
+                activeCASes.insert(currentCas);
+              } else {
+                // No new CASes are output, this CAS is done being processed by the current engine.
+                currentCas->setCurrentComponentInfo(nullptr);
+              }
+            } else {
+              UIMA_EXC_THROW_NEW(EngineProcessingException,
+                                 UIMA_ERR_USER_ANNOTATOR_COULD_NOT_PROCESS,
+                                 UIMA_MSG_ID_EXCON_PROCESSING_CAS,
+                                 ErrorMessage(UIMA_MSG_ID_LITERAL_STRING, "Unknown Delegate Key " + nextAEKey),
+                                 ErrorInfo::unrecoverable);
+            }
+          } else if (nextStep.getType() == Step::StepType::PARALLELSTEP) {
+            // TODO: ParallelStep not supported yet
+            UIMA_EXC_THROW_NEW(NotYetImplementedException,
+                               UIMA_ERR_NOT_YET_IMPLEMENTED,
+                               UIMA_MSG_ID_EXC_NOT_YET_IMPLEMENTED,
+                               ErrorMessage(UIMA_MSG_ID_LITERAL_STRING, "Parallel Step not supported yet"),
+                               ErrorInfo::unrecoverable
+            );
+          } else {
+            UIMA_EXC_THROW_NEW(EngineProcessingException,
+                               UIMA_ERR_USER_ANNOTATOR_COULD_NOT_PROCESS,
+                               UIMA_MSG_ID_EXCON_PROCESSING_CAS,
+                               ErrorMessage(UIMA_MSG_ID_LITERAL_STRING, "Unknown Step Type"),
+                               ErrorInfo::unrecoverable);
+          }
+
+          nextStep = flow->next();
+        }
+
+        const FinalStep *finalStep = nextStep.getFinalStep();
+        activeCASes.erase(currentCas);
+        if (currentCas == inputCas) {
+          if (finalStep->getForceDropCAS()) {
+            // Not allowed to drop the input CAS so something must have gone wrong
+            UIMA_EXC_THROW_NEW(EngineProcessingException,
+                               UIMA_ERR_USER_ANNOTATOR_COULD_NOT_PROCESS,
+                               UIMA_MSG_ID_EXCON_PROCESSING_CAS,
+                               ErrorMessage(UIMA_MSG_ID_LITERAL_STRING, "Illegal CAS drop"),
+                               ErrorInfo::unrecoverable);
+          }
+          return nullptr;
+        }
+
+        if (iv_bOutputNewCases && !finalStep->getForceDropCAS())
+          return currentCas;
+        currentCas->release();
+      }
+    }
+
+    bool AnnotatorManager::hasNext() {
+      if (!nextCas)
+        nextCas = processUntilNextOutputCas();
+      return nextCas != nullptr;
+    }
+
+    CAS & AnnotatorManager::next() {
+      CAS* result = nextCas;
+      if (!result)
+        result = processUntilNextOutputCas();
+      if (!result) {
+        UIMA_EXC_THROW_NEW(Exception,
+                   UIMA_ERR_USER_ANNOTATOR_COULD_NOT_PROCESS,
+                   UIMA_MSG_ID_EXCON_PROCESSING_CAS,
+                   ErrorMessage(UIMA_MSG_ID_LITERAL_STRING, "There is not next() available."),
+                   ErrorInfo::unrecoverable);
+      }
+      nextCas = nullptr;
+      return *result;
+    }
+
+    void AnnotatorManager::release() {
+      while (!casIterStack.empty()) {
+        StackFrame& frame = casIterStack.top();
+        frame.originalFlow->aborted();
+        casIterStack.pop();
+      }
+      for (CAS *cas : activeCASes) {
+        if (cas != inputCas)
+          cas->release();
+      }
+
+      activeCASes.clear();
+    }
+
+
+    TyErrorId AnnotatorManager::launchProcessDocument(CAS &cas, ResultSpecification const &crResultSpec) {
+      //if engine uses Capability Language Flow
+      // TODO: Turn this logic and processCapabilityLanguageFlow into a separate CapabilityLanguageFlowController class that inherits from FlowController
+      AnnotatorContext &rANC = iv_pEngine->getAnnotatorContext();
+      AnalysisEngineDescription const &crTAESpecifier = rANC.getTaeSpecifier();
+      FlowConstraints const *pFlow = crTAESpecifier.getAnalysisEngineMetaData()->getFlowConstraints();
+      FlowConstraints *flow = CONST_CAST(FlowConstraints *, pFlow);
+
+      // Process according to capability language specifications
+      if (flow->getFlowConstraintsType() == FlowConstraints::CAPABILITYLANGUAGE)
+        return processCapabilityLanguageFlow(cas, crResultSpec);
+
+      inputCas = &cas;
+
+      casIterStack.push({nullptr, inputCas, iv_pFlowController->computeFlow(*inputCas), {}});
+      try {
+        nextCas = processUntilNextOutputCas();
+      } catch (...) {
+        release();
+        throw;
+      }
+      return UIMA_ERR_NONE;
     }
 
 #ifdef UIMA_DEBUG_ANNOTATOR_TIMING