Salome HOME
Successful first launch of scheme on cluster.
[modules/yacs.git] / src / evalyfx / YACSEvalYFXPattern.cxx
1 // Copyright (C) 2012-2016  CEA/DEN, EDF R&D
2 //
3 // This library is free software; you can redistribute it and/or
4 // modify it under the terms of the GNU Lesser General Public
5 // License as published by the Free Software Foundation; either
6 // version 2.1 of the License, or (at your option) any later version.
7 //
8 // This library is distributed in the hope that it will be useful,
9 // but WITHOUT ANY WARRANTY; without even the implied warranty of
10 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
11 // Lesser General Public License for more details.
12 //
13 // You should have received a copy of the GNU Lesser General Public
14 // License along with this library; if not, write to the Free Software
15 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
16 //
17 // See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
18 //
19 // Author : Anthony Geay (EDF R&D)
20
21 #include "YACSEvalYFXPattern.hxx"
22 #include "YACSEvalResource.hxx"
23 #include "YACSEvalSeqAny.hxx"
24 #include "YACSEvalSession.hxx"
25 #include "YACSEvalObserver.hxx"
26 #include "YACSEvalSessionInternal.hxx"
27 #include "YACSEvalAutoPtr.hxx"
28
29 #include "ElementaryNode.hxx"
30 #include "RuntimeSALOME.hxx"
31 #include "Dispatcher.hxx"
32 #include "Executor.hxx"
33 #include "InputPort.hxx"
34 #include "LinkInfo.hxx"
35 #include "TypeCode.hxx"
36 #include "Proc.hxx"
37
38 #include "PythonPorts.hxx"
39 #include "ForEachLoop.hxx"
40 #include "PythonNode.hxx"
41 #include "InlineNode.hxx"
42 #include "ServiceNode.hxx"
43 #include "PyStdout.hxx"
44 #include "AutoGIL.hxx"
45
46 #include "ResourcesManager.hxx"
47
48 #include <map>
49 #include <limits>
50 #include <numeric>
51 #include <sstream>
52 #include <iterator>
53
54 ////
55 #include <stdlib.h>
56
57 const char YACSEvalYFXPattern::DFT_PROC_NAME[]="YFX";
58
59 const char YACSEvalYFXPattern::ST_OK[]="ALL_OK";
60
61 const char YACSEvalYFXPattern::ST_FAILED[]="SOME_SAMPLES_FAILED_AND_ALL_OF_THEM_FAILED_DETERMINISTICALLY";
62
63 const char YACSEvalYFXPattern::ST_ERROR[]="SOME_SAMPLES_FAILED_BUT_IMPOSSIBLE_TO_CONCLUDE_ON_THEM";
64
65 const std::size_t YACSEvalYFXPattern::MAX_LGTH_OF_INP_DUMP=10000;
66
67 const char YACSEvalYFXRunOnlyPattern::FIRST_FE_SUBNODE_NAME[]="Bloc";
68
69 const char YACSEvalYFXRunOnlyPattern::GATHER_NODE_NAME[]="__gather__";
70
71 class MyAutoThreadSaver
72 {
73 public:
74   MyAutoThreadSaver():_save(PyEval_SaveThread()) { }
75   ~MyAutoThreadSaver() { PyEval_RestoreThread(_save); }
76 private:
77   PyThreadState *_save;
78 };
79
80 std::vector< YACSEvalInputPort *> YACSEvalYFXPattern::getFreeInputPorts() const
81 {
82   std::size_t sz(_inputs.size());
83   std::vector< YACSEvalInputPort *> ret;
84   std::vector< YACSEvalInputPort >::const_iterator it(_inputs.begin());
85   for(std::size_t i=0;i<sz;i++,it++)
86     ret.push_back(const_cast<YACSEvalInputPort *>(&(*it)));
87   return ret;
88 }
89
90 std::vector< YACSEvalOutputPort *> YACSEvalYFXPattern::getFreeOutputPorts() const
91 {
92   std::size_t sz(_outputs.size());
93   std::vector< YACSEvalOutputPort *> ret;
94   std::vector< YACSEvalOutputPort >::const_iterator it(_outputs.begin());
95   for(std::size_t i=0;i<sz;i++,it++)
96     ret.push_back(const_cast<YACSEvalOutputPort *>(&(*it)));
97   return ret;
98 }
99
100 YACSEvalYFXPattern *YACSEvalYFXPattern::FindPatternFrom(YACSEvalYFX *boss, YACS::ENGINE::Proc *scheme, bool ownScheme)
101 {
102   if(!scheme)
103     throw YACS::Exception("YACSEvalYFXPattern::FindPatternFrom : input scheme must be not null !");
104   {
105       YACS::ENGINE::ComposedNode *zeRunNode(0);
106       bool isMatchingRunOnlyPattern(YACSEvalYFXRunOnlyPattern::IsMatching(scheme,zeRunNode));
107       if(isMatchingRunOnlyPattern)
108         return new YACSEvalYFXRunOnlyPattern(boss,scheme,ownScheme,zeRunNode);
109   }
110   throw YACS::Exception("YACSEvalYFXPattern::FindPatternFrom : no pattern found for the input scheme !");
111 }
112
113 bool YACSEvalYFXPattern::isAlreadyComputedResources() const
114 {
115   return _res!=0;
116 }
117
118 void YACSEvalYFXPattern::checkNonAlreadyComputedResources() const
119 {
120   if(isAlreadyComputedResources())
121     throw YACS::Exception("checkNonAlreadyComputedResources : instance of computed resources already computed !");
122 }
123
124 void YACSEvalYFXPattern::checkAlreadyComputedResources() const
125 {
126   if(!isAlreadyComputedResources())
127     throw YACS::Exception("checkAlreadyComputedResources : instance of computed resources not already computed !");
128 }
129
130 void YACSEvalYFXPattern::checkLocked() const
131 {
132   if(!isLocked())
133     throw YACS::Exception("YACSEvalYFXPattern::checkLocked : Pattern is not locked !");
134 }
135
136 void YACSEvalYFXPattern::checkNonLocked() const
137 {
138   if(isLocked())
139     throw YACS::Exception("YACSEvalYFXPattern::checkNonLocked : Pattern is locked !");
140 }
141
142 void YACSEvalYFXPattern::CheckNodeIsOK(YACS::ENGINE::ComposedNode *node)
143 {
144   /*YACS::ENGINE::LinkInfo info(YACS::ENGINE::LinkInfo::WARN_ONLY_DONT_STOP);
145   try
146   {
147       node->checkConsistency(info);
148   }
149   catch(YACS::Exception& e)
150   {
151   }
152   if(info.getNumberOfErrLinks(YACS::ENGINE::E_ALL)!=0)
153     throw YACS::Exception("YACSEvalYFXPattern::CheckNodeIsOK : found node is not OK !");
154   std::list<YACS::ENGINE::ElementaryNode *> allNodes(node->getRecursiveConstituents());
155   for(std::list<YACS::ENGINE::ElementaryNode *>::const_iterator it=allNodes.begin();it!=allNodes.end();it++)
156     {
157       YACS::ENGINE::ServiceNode *node0(dynamic_cast<YACS::ENGINE::ServiceNode *>(*it));
158       YACS::ENGINE::InlineNode *node1(dynamic_cast<YACS::ENGINE::InlineNode *>(*it));
159       if(node0)
160         {
161           YACS::ENGINE::Container *cont(node0->getContainer());
162           YACS::ENGINE::ComponentInstance *comp(node0->getComponent());
163           if(!cont || !comp)
164             {
165               std::ostringstream oss; oss << "YACSEvalYFXPattern::CheckNodeIsOK : ServiceNode called \"" << node0->getName() << "\" is not correctly defined !";
166               throw YACS::Exception(oss.str());
167             }
168         }
169       if(node1)
170         {
171           YACS::ENGINE::Container *cont(node1->getContainer());
172           if(!cont && node1->getExecutionMode()==YACS::ENGINE::InlineNode::REMOTE_STR)
173             {
174               std::ostringstream oss; oss << "YACSEvalYFXPattern::CheckNodeIsOK : InlineNode called \"" << node1->getName() << "\" is not correctly defined !";
175               throw YACS::Exception(oss.str());
176             }
177         }
178     }*/
179 }
180
181 void YACSEvalYFXPattern::registerObserver(YACSEvalObserver *observer)
182 {
183   if(_observer==observer)
184     return ;
185   if(_observer)
186     _observer->decrRef();
187   _observer=observer;
188   if(_observer)
189     _observer->incrRef();
190 }
191
192 YACSEvalYFXPattern::YACSEvalYFXPattern(YACSEvalYFX *boss, YACS::ENGINE::Proc *scheme, bool ownScheme):_boss(boss),_scheme(scheme),_ownScheme(ownScheme),_parallelizeStatus(true),_rm(new ResourcesManager_cpp),_res(0),_observer(0)
193 {
194 }
195
196 YACS::ENGINE::TypeCode *YACSEvalYFXPattern::createSeqTypeCodeFrom(YACS::ENGINE::Proc *scheme, const std::string& zeType)
197 {
198   std::ostringstream oss; oss << "list[" << zeType << "]";
199   YACS::ENGINE::TypeCode *tc(scheme->getTypeCode(zeType));
200   return scheme->createSequenceTc(oss.str(),oss.str(),tc);
201 }
202
203 void YACSEvalYFXPattern::setResources(YACSEvalListOfResources *res)
204 {
205   checkNonAlreadyComputedResources();
206   if(res!=_res)
207     delete _res;
208   _res=res;
209 }
210
211 void YACSEvalYFXPattern::resetResources()
212 {
213   delete _res;
214   _res=0;
215 }
216
217 YACSEvalSeqAny *YACSEvalYFXPattern::BuildValueInPort(YACS::ENGINE::InputPyPort *port)
218 {
219   if(!port)
220     throw YACS::Exception("YACSEvalYFXPattern::GetValueInPort : null input port !");
221   PyObject *obj(port->getPyObj());
222   YACS::ENGINE::TypeCode *tc(port->edGetType());
223   YACS::ENGINE::TypeCodeSeq *tcc(dynamic_cast<YACS::ENGINE::TypeCodeSeq *>(tc));
224   if(!tcc)
225     {
226       std::ostringstream oss; oss << "YACSEvalYFXPattern::GetValueInPort : internal error for tc of input \"" << port->getName() << "\"";
227       throw YACS::Exception(oss.str());
228     }
229   const YACS::ENGINE::TypeCode *tcct(tcc->contentType());
230   if(!PyList_Check(obj))
231     throw YACS::Exception("YACSEvalYFXPattern::GetValueInPort : internal error 2 !");
232   std::size_t sz(PyList_Size(obj));
233   if(tcct->kind()==YACS::ENGINE::Double)
234     {
235       std::vector<double> eltCpp(sz);
236       for(std::size_t i=0;i<sz;i++)
237         {
238           PyObject *elt(PyList_GetItem(obj,i));
239           eltCpp[i]=PyFloat_AsDouble(elt);
240         }
241       YACS::AutoCppPtr<YACSEvalSeqAnyDouble> elt(new YACSEvalSeqAnyDouble(eltCpp));
242       return elt.dettach();
243     }
244   else if(tcct->kind()==YACS::ENGINE::Int)
245     {
246       std::vector<int> eltCpp(sz);
247       for(std::size_t i=0;i<sz;i++)
248         {
249           PyObject *elt(PyList_GetItem(obj,i));
250           eltCpp[i]=PyInt_AsLong(elt);
251         }
252       YACS::AutoCppPtr<YACSEvalSeqAnyInt> elt(new YACSEvalSeqAnyInt(eltCpp));
253       return elt.dettach();
254     }
255   else
256     throw YACS::Exception("YACSEvalYFXPattern::GetValueInPort : not implemented yet for other than Double and Int !");
257 }
258
259 YACSEvalSeqAny *YACSEvalYFXPattern::BuildValueFromEngineFrmt(YACS::ENGINE::SequenceAny *data)
260 {
261   unsigned int sz(data->size());
262   std::vector<double> eltCpp(sz);
263   for(unsigned int ii=0;ii<sz;ii++)
264     {
265       YACS::ENGINE::AnyPtr elt((*data)[ii]);
266       YACS::ENGINE::Any *eltPtr((YACS::ENGINE::Any *)elt);
267       YACS::ENGINE::AtomAny *eltPtr2(dynamic_cast<YACS::ENGINE::AtomAny *>(eltPtr));
268       if(!eltPtr2)
269         {
270           std::ostringstream oss; oss << "YACSEvalYFXPattern::BuildValueFromEngineFrmt : error at pos #" << ii << " ! It is not an AtomAny !";
271           throw YACS::Exception(oss.str());
272         }
273       eltCpp[ii]=eltPtr2->getDoubleValue();
274     }
275   return new YACSEvalSeqAnyDouble(eltCpp);
276 }
277
278 void YACSEvalYFXPattern::cleanScheme()
279 {
280   if(_ownScheme)
281     delete _scheme;
282   _scheme=0;
283 }
284
285 YACSEvalYFXPattern::~YACSEvalYFXPattern()
286 {
287   if(_observer)
288     _observer->decrRef();
289   delete _rm;
290   delete _res;
291 }
292
293 /////////////////////
294
295 class YACSEvalYFXRunOnlyPatternInternalObserver : public YACS::ENGINE::Observer
296 {
297 public:
298   YACSEvalYFXRunOnlyPatternInternalObserver(YACSEvalYFXRunOnlyPattern *boss):_boss(boss) { if(!_boss) throw YACS::Exception("YACSEvalYFXRunOnlyPatternInternalObserver constructor : null boss not supported :)"); }
299   void notifyObserver(YACS::ENGINE::Node *object, const std::string& event);
300 private:
301   YACSEvalYFXRunOnlyPattern *_boss;
302 };
303
304 void YACSEvalYFXRunOnlyPatternInternalObserver::notifyObserver(YACS::ENGINE::Node *object, const std::string& event)
305 {
306   YACS::ENGINE::ForEachLoop *object2(dynamic_cast<YACS::ENGINE::ForEachLoop *>(object));
307   if(!object2)
308     return ;
309   YACSEvalObserver *obs(_boss->getObserver());
310   if(!obs)
311     return ;
312   if(event=="progress")
313     obs->notifyNewNumberOfPassedItems(_boss->getBoss(),object2->getCurrentIndex());
314 }
315
316 /////////////////////
317
318 YACSEvalYFXRunOnlyPattern::YACSEvalYFXRunOnlyPattern(YACSEvalYFX *boss, YACS::ENGINE::Proc *scheme, bool ownScheme, YACS::ENGINE::ComposedNode *runNode):YACSEvalYFXPattern(boss,scheme,ownScheme),_runNode(runNode),_generatedGraph(0),_FEInGeneratedGraph(0),_obs(new YACSEvalYFXRunOnlyPatternInternalObserver(this))
319 {
320   if(!_runNode)
321     throw YACS::Exception("YACSEvalYFXRunOnlyPattern : internal run node must be not null !");
322   buildInputPorts();
323   buildOutputPorts();
324 }
325
326 YACSEvalYFXRunOnlyPattern::~YACSEvalYFXRunOnlyPattern()
327 {
328   delete _obs;
329 }
330
331 void YACSEvalYFXRunOnlyPattern::setOutPortsOfInterestForEvaluation(const std::vector<YACSEvalOutputPort *>& outputsOfInterest)
332 {
333   checkNonLocked();
334   _outputsOfInterest=outputsOfInterest;
335 }
336
337 void YACSEvalYFXRunOnlyPattern::resetOutputsOfInterest()
338 {
339   checkLocked();
340   _outputsOfInterest.clear();
341 }
342
343 void YACSEvalYFXRunOnlyPattern::generateGraph()
344 {
345   static const char LISTPYOBJ_STR[]="list[pyobj]";
346   if(_outputsOfInterest.empty())
347     return ;
348   YACS::ENGINE::RuntimeSALOME::setRuntime();
349   YACS::ENGINE::RuntimeSALOME *r(YACS::ENGINE::getSALOMERuntime());
350   _generatedGraph=r->createProc(DFT_PROC_NAME);
351   YACS::ENGINE::TypeCode *pyobjTC(_generatedGraph->createInterfaceTc("python:obj:1.0","pyobj",std::list<YACS::ENGINE::TypeCodeObjref *>()));
352   std::ostringstream oss; oss << "Loop_" << _runNode->getName();
353   _generatedGraph->createType(YACSEvalAnyDouble::TYPE_REPR,"double");
354   _generatedGraph->createType(YACSEvalAnyInt::TYPE_REPR,"int");
355   //
356   YACS::ENGINE::InlineNode *n0(r->createScriptNode(YACS::ENGINE::PythonNode::KIND,"__initializer__"));
357   _generatedGraph->edAddChild(n0);
358   YACS::ENGINE::TypeCode *listPyobjTC(_generatedGraph->createSequenceTc(LISTPYOBJ_STR,LISTPYOBJ_STR,pyobjTC));
359   YACS::ENGINE::OutputPort *sender(n0->edAddOutputPort("sender",listPyobjTC));
360   std::ostringstream var0;
361   for(std::vector< YACSEvalInputPort >::const_iterator it=_inputs.begin();it!=_inputs.end();it++)
362     {
363       if((*it).isRandomVar())
364         {
365           var0 << (*it).getName() << ",";
366           YACS::ENGINE::TypeCode *tc(createSeqTypeCodeFrom(_generatedGraph,(*it).getTypeOfData()));
367           YACS::ENGINE::InputPort *inp(n0->edAddInputPort((*it).getName(),tc));
368           YACS::ENGINE::InputPyPort *inpc(dynamic_cast<YACS::ENGINE::InputPyPort *>(inp));
369           if(!inpc)
370             throw YACS::Exception("YACSEvalYFXRunOnlyPattern::generateGraph : internal error 1 !");
371           (*it).setUndergroundPortToBeSet(inpc);
372         }
373     }
374   std::ostringstream n0Script; n0Script << "sender=zip(" << var0.str() << ")\n";
375   n0->setScript(n0Script.str());
376   //
377   YACS::ENGINE::ForEachLoop *n1(r->createForEachLoop(oss.str(),pyobjTC));
378   _FEInGeneratedGraph=n1;
379   _generatedGraph->edAddChild(n1);
380   _generatedGraph->edAddCFLink(n0,n1);
381   _generatedGraph->edAddDFLink(sender,n1->edGetSeqOfSamplesPort());
382   YACS::ENGINE::InlineNode *n2(r->createScriptNode(YACS::ENGINE::PythonNode::KIND,GATHER_NODE_NAME));
383   _generatedGraph->edAddChild(n2);
384   _generatedGraph->edAddCFLink(n1,n2);
385   //
386   YACS::ENGINE::Bloc *n10(r->createBloc(FIRST_FE_SUBNODE_NAME));
387   n1->edAddChild(n10);
388   YACS::ENGINE::InlineNode *n100(r->createScriptNode(YACS::ENGINE::PythonNode::KIND,"__dispatch__"));
389   YACS::ENGINE::Node *n101(_runNode->cloneWithoutCompAndContDeepCpy(0,true));
390   n10->edAddChild(n100);
391   n10->edAddChild(n101);
392   YACS::ENGINE::InputPort *dispatchIn(n100->edAddInputPort("i0",pyobjTC));
393   n10->edAddCFLink(n100,n101);
394   n1->edAddDFLink(n1->edGetSamplePort(),dispatchIn);
395   std::ostringstream var1;
396   for(std::vector< YACSEvalInputPort >::const_iterator it=_inputs.begin();it!=_inputs.end();it++)
397     {
398       if((*it).isRandomVar())
399         {
400           var1 << (*it).getName() << ",";
401           YACS::ENGINE::OutputPort *myOut(n100->edAddOutputPort((*it).getName(),_generatedGraph->getTypeCode((*it).getTypeOfData())));
402           std::string tmpPortName(_runNode->getInPortName((*it).getUndergroundPtr()));
403           YACS::ENGINE::InputPort *myIn(n101->getInputPort(tmpPortName));
404           n10->edAddDFLink(myOut,myIn);
405         }
406     }
407   std::ostringstream n100Script;  n100Script << var1.str() << "=i0\n";
408   n100->setScript(n100Script.str());
409   for(std::vector< YACSEvalOutputPort * >::const_iterator it=_outputsOfInterest.begin();it!=_outputsOfInterest.end();it++)
410     {
411       YACS::ENGINE::TypeCode *tc(createSeqTypeCodeFrom(_generatedGraph,(*it)->getTypeOfData()));
412       YACS::ENGINE::InputPort *myIn(n2->edAddInputPort((*it)->getName(),tc));
413       std::string tmpPortName(_runNode->getOutPortName((*it)->getUndergroundPtr()));
414       YACS::ENGINE::OutputPort *myOut(n101->getOutputPort(tmpPortName));
415       _generatedGraph->edAddDFLink(myOut,myIn);
416     }
417   _generatedGraph->updateContainersAndComponents();
418 }
419
420 void YACSEvalYFXRunOnlyPattern::resetGeneratedGraph()
421 {
422   delete _generatedGraph;
423   _generatedGraph=0;
424   resetResources();
425 }
426
427 int YACSEvalYFXRunOnlyPattern::assignNbOfBranches()
428 {
429   checkAlreadyComputedResources();
430   if(!_generatedGraph)
431     throw YACS::Exception("YACSEvalYFXRunOnlyPattern::assignNbOfBranches : the generated graph has not been created !");
432   std::list<YACS::ENGINE::Node *> nodes(_generatedGraph->getChildren());
433   YACS::ENGINE::ForEachLoop *zeMainNode(0);
434   for(std::list<YACS::ENGINE::Node *>::const_iterator it=nodes.begin();it!=nodes.end();it++)
435     {
436       YACS::ENGINE::ForEachLoop *isZeMainNode(dynamic_cast<YACS::ENGINE::ForEachLoop *>(*it));
437       if(isZeMainNode)
438         {
439           if(!zeMainNode)
440             zeMainNode=isZeMainNode;
441           else
442             throw YACS::Exception("YACSEvalYFXRunOnlyPattern::assignNbOfBranches : internal error 1 !");
443         }
444     }
445   if(!zeMainNode)
446     throw YACS::Exception("YACSEvalYFXRunOnlyPattern::assignNbOfBranches : internal error 2 !");
447   unsigned int nbProcsDeclared(getResourcesInternal()->getNumberOfProcsDeclared());
448   nbProcsDeclared=std::max(nbProcsDeclared,4u);
449   int nbOfBranch=1;
450   if(getParallelizeStatus())
451     {
452       nbOfBranch=(nbProcsDeclared/getResourcesInternal()->getMaxLevelOfParallelism());
453       nbOfBranch=std::max(nbOfBranch,1);
454     }
455   YACS::ENGINE::InputPort *zeInputToSet(zeMainNode->edGetNbOfBranchesPort());
456   YACS::ENGINE::AnyInputPort *zeInputToSetC(dynamic_cast<YACS::ENGINE::AnyInputPort *>(zeInputToSet));
457   if(!zeInputToSetC)
458     throw YACS::Exception("YACSEvalYFXRunOnlyPattern::assignNbOfBranches : internal error 3 !");
459   YACS::ENGINE::Any *a(YACS::ENGINE::AtomAny::New(nbOfBranch));
460   zeInputToSetC->put(a);
461   zeInputToSetC->exSaveInit();
462   a->decrRef();
463   return nbOfBranch;
464 }
465
466 void YACSEvalYFXRunOnlyPattern::assignRandomVarsInputs()
467 {
468   std::size_t sz(std::numeric_limits<std::size_t>::max());
469   for(std::vector< YACSEvalInputPort >::const_iterator it=_inputs.begin();it!=_inputs.end();it++)
470     if((*it).isRandomVar())
471       {
472         std::size_t locSize((*it).initializeUndergroundWithSeq());
473         if(sz==std::numeric_limits<std::size_t>::max())
474           sz=locSize;
475         else
476           if(sz!=locSize)
477             throw YACS::Exception("YACSEvalYFXRunOnlyPattern::assignRandomVarsInputs : length of sequences in random vars must be the same !");
478       }
479 }
480
481 bool YACSEvalYFXRunOnlyPattern::isLocked() const
482 {
483   return _generatedGraph!=0;
484 }
485
486 YACSEvalListOfResources *YACSEvalYFXRunOnlyPattern::giveResources()
487 {
488   checkLocked();
489   if(!isAlreadyComputedResources())
490     {
491       YACS::ENGINE::DeploymentTree dt(_runNode->getDeploymentTree());
492       _runNode->removeRecursivelyRedundantCL();
493       YACSEvalListOfResources *res(new YACSEvalListOfResources(_runNode->getMaxLevelOfParallelism(),getCatalogInAppli(),dt));
494       setResources(res);
495     }
496   return getResourcesInternal();
497 }
498
499 YACS::ENGINE::Proc *YACSEvalYFXRunOnlyPattern::getUndergroundGeneratedGraph() const
500 {
501   return _generatedGraph;
502 }
503
504 std::string YACSEvalYFXRunOnlyPattern::getErrorDetailsInCaseOfFailure() const
505 {
506   std::string st(getStatusOfRunStr());//test if a run has occurred.
507   if(st==ST_OK)
508     throw YACS::Exception("YACSEvalYFXRunOnlyPattern::getErrorDetailsInCaseOfFailure : The execution of scheme has been carried out to the end without any problem !");
509   // All the problem can only comes from foreach -> scan it
510   YACS::ENGINE::ForEachLoop *fe(findTopForEach());
511   YACS::ENGINE::NodeStateNameMap nsm;
512   unsigned nbB(fe->getNumberOfBranchesCreatedDyn());
513   std::ostringstream oss;
514   for(unsigned j=0;j<nbB;j++)
515     {
516       YACS::ENGINE::Node *nn(fe->getChildByNameExec(FIRST_FE_SUBNODE_NAME,j));
517       YACS::ENGINE::Bloc *nnc(dynamic_cast<YACS::ENGINE::Bloc *>(nn));
518       if(!nnc)
519         throw YACS::Exception("YACSEvalYFXRunOnlyPattern::getErrorDetailsInCaseOfFailure : internal error 1 ! The direct son of main foreach is expected to be a Bloc !");
520       if(nnc->getState()==YACS::DONE)
521         continue;
522       std::list< YACS::ENGINE::ElementaryNode *> fec(nnc->getRecursiveConstituents());
523       for(std::list< YACS::ENGINE::ElementaryNode *>::reverse_iterator it1=fec.rbegin();it1!=fec.rend();it1++)
524         {
525           YACS::StatesForNode st0((*it1)->getState());
526           if(st0!=YACS::DONE)
527             {
528               oss << "NODE = " << nnc->getChildName(*it1) << std::endl;
529               oss << "STATUS = " << nsm[st0] << std::endl;
530               oss << "BRANCH ID = " << j << std::endl;
531               std::list<YACS::ENGINE::InputPort *> inps((*it1)->getSetOfInputPort());
532               for(std::list<YACS::ENGINE::InputPort *>::const_iterator it2=inps.begin();it2!=inps.end();it2++)
533                 {
534                   std::string d((*it2)->getHumanRepr());
535                   if(d.size()>10000)
536                     d=d.substr(0,MAX_LGTH_OF_INP_DUMP);
537                   oss << "INPUT \"" << (*it2)->getName() << "\" = " << d << std::endl;
538                 }
539               oss << "DETAILS = " << std::endl;
540               oss << (*it1)->getErrorDetails();
541             }
542         }
543     }
544   return oss.str();
545 }
546
547 std::string YACSEvalYFXRunOnlyPattern::getStatusOfRunStr() const
548 {
549   YACS::StatesForNode st(_generatedGraph->getState());
550   switch(st)
551     {
552     case YACS::READY:
553     case YACS::TOLOAD:
554     case YACS::LOADED:
555     case YACS::TOACTIVATE:
556     case YACS::ACTIVATED:
557     case YACS::SUSPENDED:
558     case YACS::PAUSE:
559     case YACS::DISABLED:
560     case YACS::DESACTIVATED:
561       {
562         std::ostringstream oss; oss << "YACSEvalYFXRunOnlyPattern::getStatusOfRunStr : Unexpected state \"" << YACS::ENGINE::Node::getStateName(st) << "\" ! Did you invoke run ?";
563         throw YACS::Exception(oss.str());
564       }
565     case YACS::LOADFAILED:
566     case YACS::EXECFAILED:
567     case YACS::ERROR:
568     case YACS::INTERNALERR:
569       return std::string(ST_ERROR);
570     case YACS::FAILED:
571       return std::string(ST_FAILED);
572     case YACS::DONE:
573       return std::string(ST_OK);
574     default:
575       {
576         std::ostringstream oss; oss << "YACSEvalYFXRunOnlyPattern::getStatusOfRunStr : unrecognized and managed state \"" << YACS::ENGINE::Node::getStateName(st) << "\" !";
577         throw YACS::Exception(oss.str());
578       }
579     }
580 }
581
582 std::vector<YACSEvalSeqAny *> YACSEvalYFXRunOnlyPattern::getResults() const
583 {
584   if(_generatedGraph->getState()!=YACS::DONE)
585     throw YACS::Exception("YACSEvalYFXRunOnlyPattern::getResults : the execution did not finished correctly ! getResults should not be called !");
586   std::vector<YACSEvalSeqAny *> ret(_outputsOfInterest.size());
587   YACS::ENGINE::Node *node(_generatedGraph->getChildByName(GATHER_NODE_NAME));
588   YACS::ENGINE::PythonNode *nodeC(dynamic_cast<YACS::ENGINE::PythonNode *>(node));
589   if(!nodeC)
590     throw YACS::Exception("YACSEvalYFXRunOnlyPattern::getResults : internal error !");
591   std::size_t ii(0);
592   for(std::vector< YACSEvalOutputPort * >::const_iterator it=_outputsOfInterest.begin();it!=_outputsOfInterest.end();it++,ii++)
593     {
594       YACS::ENGINE::InPort *input(nodeC->getInPort((*it)->getName()));
595       YACS::ENGINE::InputPyPort *inputC(dynamic_cast<YACS::ENGINE::InputPyPort *>(input));
596       if(!inputC)
597         {
598           std::ostringstream oss; oss << "YACSEvalYFXRunOnlyPattern::getResults : internal error for input \"" << (*it)->getName() << "\"";
599           throw YACS::Exception(oss.str());
600         }
601       ret[ii]=BuildValueInPort(inputC);
602     }
603   return ret;
604 }
605
606 /*!
607  * This method works if run succeeded (true return) and also if graph has failed. Graph failed means soft error of evaluation due to error in evaluation (example 1/0 or a normal throw from one node)
608  * If a more serious error occured (SIGSEGV of a server or internal error in YACS engine, cluster error, loose of connection...) this method will throw an exception to warn the caller that the results may be 
609  */
610 std::vector<YACSEvalSeqAny *> YACSEvalYFXRunOnlyPattern::getResultsInCaseOfFailure(std::vector<unsigned int>& passedIds) const
611 {
612   YACS::StatesForNode st(_generatedGraph->getState());
613   if(st==YACS::DONE)
614     {
615       passedIds.clear();
616       std::vector<YACSEvalSeqAny *> ret(getResults());
617       if(!ret.empty())
618         {
619           if(!ret[0])
620             throw YACS::Exception("YACSEvalYFXRunOnlyPattern::getResultsInCaseOfFailure : internal error ! The returned vector has a null pointer at pos #0 !");
621           std::size_t sz(ret[0]->size());
622           passedIds.resize(sz);
623           for(std::size_t i=0;i<sz;i++)
624             passedIds[i]=i;
625         }
626       return ret;
627     }
628   getStatusOfRunStr();// To check that the status is recognized.
629   std::list<YACS::ENGINE::Node *> lns(_generatedGraph->edGetDirectDescendants());
630   YACS::ENGINE::ForEachLoop *fe(findTopForEach());
631   //
632   YACS::ENGINE::Executor exe;
633   std::vector<YACS::ENGINE::SequenceAny *> outputs;
634   std::vector<std::string> nameOfOutputs;
635   passedIds=fe->getPassedResults(&exe,outputs,nameOfOutputs);//<- the key invokation is here.
636   std::size_t sz(passedIds.size()),ii(0);
637   std::vector<YACSEvalSeqAny *> ret(_outputsOfInterest.size());
638   for(std::vector<YACSEvalOutputPort *>::const_iterator it1=_outputsOfInterest.begin();it1!=_outputsOfInterest.end();it1++,ii++)
639     {
640       YACS::ENGINE::OutputPort *p((*it1)->getUndergroundPtr());
641       std::string st(_runNode->getOutPortName(p));
642       std::ostringstream oss; oss << FIRST_FE_SUBNODE_NAME << '.' << _runNode->getName() << '.' << st;
643       st=oss.str();
644       YACS::ENGINE::ForEachLoop::InterceptorizeNameOfPort(st);
645       std::vector<std::string>::iterator it2(std::find(nameOfOutputs.begin(),nameOfOutputs.end(),st));
646       if(it2==nameOfOutputs.end())
647         {
648           std::ostringstream oss; oss << "YACSEvalYFXRunOnlyPattern::getResultsInCaseOfFailure : internal error 3 ! Unable to locate interceptor with name " << st << " ! Possibilities are : ";
649           std::copy(nameOfOutputs.begin(),nameOfOutputs.end(),std::ostream_iterator<std::string>(oss," "));
650           oss << " !";
651           throw YACS::Exception(oss.str());
652         }
653       std::size_t pos(std::distance(nameOfOutputs.begin(),it2));
654       ret[ii]=BuildValueFromEngineFrmt(outputs[pos]);
655     }
656   return ret;
657 }
658
659 void YACSEvalYFXRunOnlyPattern::emitStart() const
660 {
661   YACSEvalObserver *obs(getObserver());
662   if(!obs)
663     return ;
664   obs->notifyNumberOfSamplesToEval(getBoss(),_FEInGeneratedGraph->getNbOfElementsToBeProcessed());
665 }
666
667 bool YACSEvalYFXRunOnlyPattern::go(bool stopASAP, YACSEvalSession *session) const
668 {
669   emitStart();
670   if(getResourcesInternal()->isInteractive())
671     {
672       YACS::ENGINE::Executor exe;
673       exe.setKeepGoingProperty(!stopASAP);
674       {
675         MyAutoThreadSaver locker;
676         exe.RunW(getUndergroundGeneratedGraph());
677       }
678       return getUndergroundGeneratedGraph()->getState()==YACS::DONE;
679     }
680   else
681     {
682       const char EFXGenFileName[]="EFXGenFileName";
683       const char EFXGenContent[]="import getpass,datetime,os\nn=datetime.datetime.now()\nreturn os.path.join(os.path.sep,\"tmp\",\"EvalYFX_%s_%s_%s.xml\"%(getpass.getuser(),n.strftime(\"%d%b%y\"),n.strftime(\"%H%M%S\")))";
684       const char EFXGenContent2[]="import getpass,datetime\nn=datetime.datetime.now()\nreturn \"EvalYFX_%s_%s_%s\"%(getpass.getuser(),n.strftime(\"%d%b%y\"),n.strftime(\"%H%M%S\"))";
685       //
686       YACS::ENGINE::AutoPyRef func(YACS::ENGINE::evalPy(EFXGenFileName,EFXGenContent));
687       YACS::ENGINE::AutoPyRef val(YACS::ENGINE::evalFuncPyWithNoParams(func));
688       std::string locSchemaFile(PyString_AsString(val));
689       getUndergroundGeneratedGraph()->saveSchema(locSchemaFile);
690       func=YACS::ENGINE::evalPy(EFXGenFileName,EFXGenContent2);
691       val=YACS::ENGINE::evalFuncPyWithNoParams(func);
692       std::string jobName(PyString_AsString(val));
693       YACSEvalListOfResources *rss(getResourcesInternal());
694       const YACSEvalParamsForCluster& cli(rss->getAddParamsForCluster());
695       std::vector<std::string> machines(rss->getAllChosenMachines());
696       if(machines.size()!=1)
697         throw YACS::Exception("YACSEvalYFXRunOnlyPattern::go : internal error ! In batch mode and not exactly one machine !");
698       Engines::SalomeLauncher_var sl(session->getInternal()->goFetchingSalomeLauncherInNS());
699       Engines::ResourceParameters rr;
700       rr.name=CORBA::string_dup(machines[0].c_str());
701       rr.hostname=CORBA::string_dup("");
702       rr.can_launch_batch_jobs=true;
703       rr.can_run_containers=true;
704       rr.OS=CORBA::string_dup("Linux");
705       rr.componentList.length(0);
706       rr.nb_proc=rss->getNumberOfProcsDeclared();// <- important
707       rr.mem_mb=1024;
708       rr.cpu_clock=1000;
709       rr.nb_node=1;// useless only nb_proc used.
710       rr.nb_proc_per_node=1;// useless only nb_proc used.
711       rr.policy=CORBA::string_dup("cycl");
712       rr.resList.length(0);
713       Engines::JobParameters jp;
714       jp.job_name=CORBA::string_dup(jobName.c_str());
715       jp.job_type=CORBA::string_dup("yacs_file");
716       jp.job_file=CORBA::string_dup(locSchemaFile.c_str());
717       jp.env_file=CORBA::string_dup("");
718       jp.in_files.length();
719       jp.out_files.length();
720       jp.work_directory=CORBA::string_dup(cli.getRemoteWorkingDir().c_str());
721       jp.local_directory=CORBA::string_dup(cli.getLocalWorkingDir().c_str());
722       jp.result_directory=CORBA::string_dup(cli.getLocalWorkingDir().c_str());
723       jp.maximum_duration=CORBA::string_dup(cli.getMaxDuration().c_str());
724       jp.resource_required=rr;
725       jp.queue=CORBA::string_dup("");
726       jp.exclusive=false;
727       jp.mem_per_cpu=rr.mem_mb;
728       jp.wckey=CORBA::string_dup(cli.getWCKey().c_str());
729       jp.extra_params=CORBA::string_dup("");
730       jp.specific_parameters.length(0);
731       jp.launcher_file=CORBA::string_dup("");
732       jp.launcher_args=CORBA::string_dup("");
733       CORBA::Long jobid(sl->createJob(jp));
734       sl->launchJob(jobid);
735       std::cerr << "*** " << jobName << " -> " << jobid << std::endl;
736       return false;
737     }
738 }
739
740 bool YACSEvalYFXRunOnlyPattern::IsMatching(YACS::ENGINE::Proc *scheme, YACS::ENGINE::ComposedNode *& runNode)
741 {
742   std::list<YACS::ENGINE::Node *> nodes(scheme->getChildren());
743   if(nodes.empty())
744     return false;
745   bool areAllElementary(true);
746   for(std::list<YACS::ENGINE::Node *>::const_iterator it=nodes.begin();it!=nodes.end() && areAllElementary;it++)
747     if(!dynamic_cast<YACS::ENGINE::ElementaryNode *>(*it))
748       areAllElementary=false;
749   if(areAllElementary)
750     {
751       if(scheme)
752         CheckNodeIsOK(scheme);
753       runNode=scheme;
754       return true;
755     }
756   if(nodes.size()!=1)
757     return false;
758   YACS::ENGINE::ComposedNode *candidate(dynamic_cast<YACS::ENGINE::ComposedNode *>(nodes.front()));
759   runNode=candidate;
760   if(candidate)
761     CheckNodeIsOK(candidate);
762   return candidate!=0;
763 }
764
765 void YACSEvalYFXRunOnlyPattern::buildInputPorts()
766 {
767   _inputs.clear();
768   std::list< YACS::ENGINE::InputPort *> allInputPorts(_runNode->getSetOfInputPort());
769   std::vector<std::string> allNames;
770   for(std::list< YACS::ENGINE::InputPort *>::const_iterator it=allInputPorts.begin();it!=allInputPorts.end();it++)
771     {
772       YACS::ENGINE::InputPort *elt(*it);
773       if(!elt)
774         throw YACS::Exception("YACSEvalYFXRunOnlyPattern::buildInputPorts : presence of null input !");
775       std::set<YACS::ENGINE::OutPort *> bls(elt->edSetOutPort());
776       if(bls.empty())
777         {
778           if(YACSEvalPort::IsInputPortPublishable(elt))
779             {
780               std::string inpName(elt->getName());
781               if(inpName.empty())
782                 throw YACS::Exception("YACSEvalYFXRunOnlyPattern::buildInputPorts : an input has empty name ! Should not !");
783               _inputs.push_back(YACSEvalInputPort(elt));
784               if(std::find(allNames.begin(),allNames.end(),inpName)!=allNames.end())
785                 {
786                   std::ostringstream oss; oss << "YACSEvalYFXRunOnlyPattern::buildInputPorts : input name \"" << inpName << "\" appears more than once !";
787                   throw YACS::Exception(oss.str());
788                 }
789               allNames.push_back(inpName);
790             }
791         }
792     }
793 }
794
795 void YACSEvalYFXRunOnlyPattern::buildOutputPorts()
796 {
797   _outputs.clear();
798   std::list< YACS::ENGINE::OutputPort *> allOutputPorts(_runNode->getSetOfOutputPort());
799   std::vector<std::string> allNames;
800   for(std::list< YACS::ENGINE::OutputPort *>::const_iterator it=allOutputPorts.begin();it!=allOutputPorts.end();it++)
801     {
802       YACS::ENGINE::OutputPort *elt(*it);
803       if(YACSEvalPort::IsOutputPortPublishable(elt))
804         {
805           if(!elt)
806             throw YACS::Exception("YACSEvalYFXRunOnlyPattern::buildOutputPorts : presence of null output !");
807           std::string outpName(elt->getName());
808           if(outpName.empty())
809             throw YACS::Exception("YACSEvalYFXRunOnlyPattern::buildOutputPorts : an output has empty name ! Should not !");
810           if(std::find(allNames.begin(),allNames.end(),outpName)!=allNames.end())
811             {
812               std::ostringstream oss; oss << "YACSEvalYFXRunOnlyPattern::buildOutputPorts : output name \"" << outpName << "\" appears more than once !";
813               throw YACS::Exception(oss.str());
814             }
815           _outputs.push_back(YACSEvalOutputPort(*it));
816         }
817     }
818 }
819
820 YACS::ENGINE::ForEachLoop *YACSEvalYFXRunOnlyPattern::findTopForEach() const
821 {
822   std::list<YACS::ENGINE::Node *> lns(_generatedGraph->edGetDirectDescendants());
823   YACS::ENGINE::ForEachLoop *fe(0);
824   for(std::list<YACS::ENGINE::Node *>::const_iterator it=lns.begin();it!=lns.end();it++)
825     {
826       fe=dynamic_cast<YACS::ENGINE::ForEachLoop *>(*it);
827       if(fe)
828         break;
829     }
830   if(!fe)
831     throw YACS::Exception("YACSEvalYFXRunOnlyPattern::findTopForEach : internal error 2 ! ForEach is not accessible !");
832   return fe;
833 }