-// Copyright (C) 2006-2015 CEA/DEN, EDF R&D
+// Copyright (C) 2006-2016 CEA/DEN, EDF R&D
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
const int ForEachLoop::NOT_RUNNING_BRANCH_ID=-1;
+const char ForEachLoop::INTERCEPTOR_STR[]="_interceptor";
+
InterceptorInputPort::InterceptorInputPort(const std::string& name, Node *node, TypeCode* type):AnyInputPort(name,node,type),
DataPort(name,node,type),Port(node),
_repr(0)
}
}
+ForEachLoopPassedData::ForEachLoopPassedData(const ForEachLoopPassedData& copy)
+: _passedIds(copy._passedIds),
+ _passedOutputs(copy._passedOutputs),
+ _nameOfOutputs(copy._nameOfOutputs),
+ _flagsIds(copy._flagsIds)
+{
+}
+
ForEachLoopPassedData::~ForEachLoopPassedData()
{
for(std::vector<SequenceAny *>::iterator it=_passedOutputs.begin();it!=_passedOutputs.end();it++)
_execNodes[id]->exUpdateState();
_nbOfEltConsumed++;
_initializingCounter--;
+ _currentIndex++;
if (_initializingCounter == 0)
_initNode->setState(DONE);
return YACS::NOEVENT;
int globalId(_execIds[id]);
if(_passedData)
globalId=_passedData->toAbsId(globalId);
+ sendEvent2("progress_ok",&globalId);
storeOutValsInSeqForOutOfScopeUse(globalId,id);
}
+ else
+ {
+ int globalId(_execIds[id]);
+ if(_passedData)
+ globalId=_passedData->toAbsId(globalId);
+ sendEvent2("progress_ko",&globalId);
+ }
//
if(_execCurrentId==getFinishedId())
{//No more elements of _dataPortToDispatch to treat
{
unsigned int id;
DynParaLoop::TypeOfNode ton(getIdentityOfNotifyerNode(node,id));
+ // TODO: deal with keepgoing without the dependency to Executor
if(ton!=WORK_NODE || !execInst->getKeepGoingProperty())
return DynParaLoop::updateStateOnFailedEventFrom(node,execInst);
else
}
}
+void ForEachLoop::InterceptorizeNameOfPort(std::string& portName)
+{
+ std::replace_if(portName.begin(), portName.end(), std::bind1st(std::equal_to<char>(), '.'), '_');
+ portName += INTERCEPTOR_STR;
+}
+
void ForEachLoop::buildDelegateOf(std::pair<OutPort *, OutPort *>& port, InPort *finalTarget, const std::list<ComposedNode *>& pointsOfView)
{
DynParaLoop::buildDelegateOf(port,finalTarget,pointsOfView);
}
else
{
- TypeCodeSeq *newTc=(TypeCodeSeq *)TypeCode::sequenceTc("","",port.first->edGetType());
+ TypeCode *tcTrad((YACS::ENGINE::TypeCode*)finalTarget->edGetType()->subContentType(getFEDeltaBetween(port.first,finalTarget)));
+ TypeCodeSeq *newTc=(TypeCodeSeq *)TypeCode::sequenceTc("","",tcTrad);
// The out going ports belong to the ForEachLoop, whereas
// the delegated port belongs to a node child of the ForEachLoop.
// The name of the delegated port contains dots (bloc.node.outport),
// whereas the name of the out going port shouldn't do.
- std::string outputPortName = getPortName(port.first);
- std::replace_if (outputPortName.begin(), outputPortName.end(),
- std::bind1st(std::equal_to<char>(), '.'), '_');
- outputPortName += "_interceptor";
- AnySplitOutputPort *newPort=new AnySplitOutputPort(outputPortName,this,newTc);
- InterceptorInputPort *intercptor=new InterceptorInputPort(outputPortName + "_in",this,port.first->edGetType());
+ std::string outputPortName(getPortName(port.first));
+ InterceptorizeNameOfPort(outputPortName);
+ AnySplitOutputPort *newPort(new AnySplitOutputPort(outputPortName,this,newTc));
+ InterceptorInputPort *intercptor(new InterceptorInputPort(outputPortName + "_in",this,tcTrad));
intercptor->setRepr(newPort);
newTc->decrRef();
newPort->addRepr(port.first,intercptor);
//AnyInputPort *interceptor=new AnyInputPort((*iter)->getName(),this,(*iter)->edGetType());
OutPort *portOut=getDynOutPortByAbsName(branchNb,getOutPortName(((*iter)->getRepr())));
DEBTRACE( portOut->getName() );
- AnyInputPort *interceptor=new AnyInputPort((*iter)->getName(),this,portOut->edGetType());
+ TypeCode *tc((TypeCode *)(*iter)->edGetType()->contentType());
+ AnyInputPort *interceptor=new AnyInputPort((*iter)->getName(),this,tc);
portOut->addInPort(interceptor);
_execOutGoingPorts[branchNb].push_back(interceptor);
}
*/
list<ProgressWeight> ForEachLoop::getProgressWeight() const
{
- list<ProgressWeight> ret;
- list<Node *> setOfNode=edGetDirectDescendants();
- int elemDone=getCurrentIndex();
- int elemTotal=getNbOfElementsToBeProcessed();
- for(list<Node *>::const_iterator iter=setOfNode.begin();iter!=setOfNode.end();iter++)
- {
- list<ProgressWeight> myCurrentSet=(*iter)->getProgressWeight();
- for(list<ProgressWeight>::iterator iter=myCurrentSet.begin();iter!=myCurrentSet.end();iter++)
- {
- (*iter).weightDone=((*iter).weightTotal) * elemDone;
- (*iter).weightTotal*=elemTotal;
- }
- ret.insert(ret.end(),myCurrentSet.begin(),myCurrentSet.end());
-
- }
- return ret;
+ list<ProgressWeight> ret;
+ list<Node *> setOfNode=edGetDirectDescendants();
+ int elemDone=getCurrentIndex();
+ int elemTotal=getNbOfElementsToBeProcessed();
+ for(list<Node *>::const_iterator iter=setOfNode.begin();iter!=setOfNode.end();iter++)
+ {
+ list<ProgressWeight> myCurrentSet=(*iter)->getProgressWeight();
+ for(list<ProgressWeight>::iterator iter=myCurrentSet.begin();iter!=myCurrentSet.end();iter++)
+ {
+ (*iter).weightDone=((*iter).weightTotal) * elemDone;
+ (*iter).weightTotal*=elemTotal;
+ }
+ ret.insert(ret.end(),myCurrentSet.begin(),myCurrentSet.end());
+ }
+ return ret;
}
int ForEachLoop::getNbOfElementsToBeProcessed() const
{
- return _splitterNode.getNumberOfElements();
+ int nbBranches = _nbOfBranches.getIntValue();
+ return _splitterNode.getNumberOfElements()
+ + (_initNode ? nbBranches:0)
+ + (_finalizeNode ? nbBranches:0) ;
}
/*!
return std::vector<unsigned int>();
if(_execOutGoingPorts.empty())
return std::vector<unsigned int>();
- std::size_t sz(_execVals.size()); outputs.resize(sz); nameOfOutputs.resize(sz);
+ std::size_t sz(_execVals.size());
+ outputs.resize(sz);
+ nameOfOutputs.resize(sz);
const std::vector<AnyInputPort *>& ports(_execOutGoingPorts[0]);
for(std::size_t i=0;i<sz;i++)
{
_passedData=new ForEachLoopPassedData(passedIds,passedOutputs,nameOfOutputs);
}
+int ForEachLoop::getFEDeltaBetween(OutPort *start, InPort *end)
+{
+ Node *ns(start->getNode()),*ne(end->getNode());
+ ComposedNode *co(getLowestCommonAncestor(ns,ne));
+ int ret(0);
+ Node *work(ns);
+ while(work!=co)
+ {
+ ForEachLoop *isFE(dynamic_cast<ForEachLoop *>(work));
+ if(isFE)
+ ret++;
+ work=work->getFather();
+ }
+ if(dynamic_cast<AnySplitOutputPort *>(start))
+ ret--;
+ return ret;
+}
+
+/*!
+ * This method is used to obtain the values already processed by the ForEachLoop.
+ * A new ForEachLoopPassedData object is returned. You have to delete it.
+ */
+ForEachLoopPassedData* ForEachLoop::getProcessedData()const
+{
+ std::vector<SequenceAny *> outputs;
+ std::vector<std::string> nameOfOutputs;
+ if(_execVals.empty() || _execOutGoingPorts.empty())
+ return new ForEachLoopPassedData(std::vector<unsigned int>(), outputs, nameOfOutputs);
+ std::size_t sz(_execVals.size());
+ outputs.resize(sz);
+ nameOfOutputs.resize(sz);
+ const std::vector<AnyInputPort *>& ports(_execOutGoingPorts[0]);
+ for(std::size_t i=0;i<sz;i++)
+ {
+ outputs[i]=_execVals[i]->removeUnsetItemsFromThis();
+ nameOfOutputs[i]=ports[i]->getName();
+ }
+ return new ForEachLoopPassedData(_execVals[0]->getSetItems(), outputs, nameOfOutputs);
+}
+
+void ForEachLoop::setProcessedData(ForEachLoopPassedData* processedData)
+{
+ if(_passedData)
+ delete _passedData;
+ _passedData = processedData;
+}
+
+/*!
+ * \param portName : "interceptorized" name of port.
+ */
+const YACS::ENGINE::TypeCode* ForEachLoop::getOutputPortType(const std::string& portName)const
+{
+ const YACS::ENGINE::TypeCode* ret=NULL;
+ vector<AnySplitOutputPort *>::const_iterator it;
+ for(it=_outGoingPorts.begin();it!=_outGoingPorts.end() && ret==NULL;it++)
+ {
+ std::string originalPortName(getPortName(*it));
+ //InterceptorizeNameOfPort(originalPortName);
+ DEBTRACE("ForEachLoop::getOutputPortType compare " << portName << " == " << originalPortName);
+ if(originalPortName == portName)
+ {
+ ret = (*it)->edGetType()->contentType();
+ }
+ }
+ return ret;
+}
