From 6a818e9acc870a5fe80a9c1ebdf3ea636d099134 Mon Sep 17 00:00:00 2001
From: Anthony Geay <anthony.geay@edf.fr>
Date: Fri, 16 Jan 2015 18:10:25 +0100
Subject: [PATCH] Retrieve state of ForEachLoop node even in case of failure.

---
 src/engine/Any.cxx         | 32 +++++++++++++++++++++++++++++++-
 src/engine/Any.hxx         |  4 ++++
 src/engine/Executor.hxx    |  1 +
 src/engine/ForEachLoop.cxx | 35 +++++++++++++++++++++++++++++++++++
 src/engine/ForEachLoop.hxx |  5 +++++
 src/engine_swig/pilot.i    | 37 +++++++++++++++++++++++++++++++++++++
 6 files changed, 113 insertions(+), 1 deletion(-)

diff --git a/src/engine/Any.cxx b/src/engine/Any.cxx
index 0ca4b843a..e062e0111 100644
--- a/src/engine/Any.cxx
+++ b/src/engine/Any.cxx
@@ -28,6 +28,9 @@
 using namespace YACS::ENGINE;
 using namespace std;
 
+// forbidden value int=-269488145 double=-1.54947e+231 bool=239
+const char SeqAlloc::DFT_CHAR_VAR=-17;//0xEF
+
 StringOnHeap::StringOnHeap(const char *val):_dealloc(0),_str(strdup(val))
 {
 }
@@ -382,7 +385,7 @@ void SeqAlloc::initCoarseMemory(char *mem, unsigned int size, Deallocator deallo
         memcpy(_start,mem,sizeInByte);
       else
         {
-          for(unsigned int i=0;i<sizeInByte;i++) _start[i]=0;
+          for(unsigned int i=0;i<sizeInByte;i++) _start[i]=DFT_CHAR_VAR;// see getSetItems
         }
     }
   _finish=_start+sizeInByte;
@@ -439,6 +442,20 @@ unsigned int SeqAlloc::size() const
   return (_finish-_start)/_sizeOf1Elm;
 }
 
+std::vector<unsigned int> SeqAlloc::getSetItems() const
+{
+  std::vector<unsigned int> ret;
+  unsigned int sz(size());
+  for(unsigned int i=0;i<sz;i++)
+    {
+      const char *pt(_start+i*_sizeOf1Elm);
+      for(unsigned j=0;j<_sizeOf1Elm && *pt==DFT_CHAR_VAR;j++,pt++); //see initCoarseMemory
+      if(pt!=_start+(i+1)*_sizeOf1Elm)
+        ret.push_back(i);
+    }
+  return ret;
+}
+
 void SequenceAny::clear()
 {
   for (char *cur=_alloc._start;cur!=_alloc._finish;cur+=_alloc._sizeOf1Elm)
@@ -539,6 +556,19 @@ Any *SequenceAny::clone() const
   return new SequenceAny(*this);
 }
 
+SequenceAny *SequenceAny::removeUnsetItemsFromThis() const
+{
+  std::vector<unsigned int> its(getSetItems());
+  std::size_t sz(its.size());
+  SequenceAny *ret(SequenceAny::New(getType()->contentType(),sz));
+  for(std::size_t i=0;i<sz;i++)
+    {
+      AnyPtr obj((*this)[its[i]]);
+      ret->setEltAtRank(i,obj);
+    }
+  return ret;
+}
+
 SequenceAny *SequenceAny::New(const TypeCode *typeOfContent)
 {
   if(typeOfContent->kind() == Objref)
diff --git a/src/engine/Any.hxx b/src/engine/Any.hxx
index cc9a51c9d..3f4560f1b 100644
--- a/src/engine/Any.hxx
+++ b/src/engine/Any.hxx
@@ -162,6 +162,8 @@ namespace YACS
       void destroy(char *pt, const TypeCode *tc);
       void deallocate(char *pt);
       unsigned int size() const;
+      std::vector<unsigned int> getSetItems() const;
+      static const char DFT_CHAR_VAR;
     };
     
     class YACSLIBENGINE_EXPORT ComposedAny : public Any
@@ -201,6 +203,8 @@ namespace YACS
       static SequenceAny *New(const TypeCode *typeOfContent, unsigned lgth);
       template<class T>
       static SequenceAny *New(T *val, unsigned int lgth, Deallocator deAlloc);
+      std::vector<unsigned int> getSetItems() const { return _alloc.getSetItems(); }
+      SequenceAny *removeUnsetItemsFromThis() const;
     protected:
       void putMyReprAtPlace(char *data) const;
       static void putReprAtPlace(char *data, const char *src, const TypeCode *type, bool deepCpy);
diff --git a/src/engine/Executor.hxx b/src/engine/Executor.hxx
index bf606ea6e..fedd17196 100644
--- a/src/engine/Executor.hxx
+++ b/src/engine/Executor.hxx
@@ -114,6 +114,7 @@ namespace YACS
       void waitPause();
       static int _maxThreads;
       static size_t _threadStackSize;
+      YACS::BASES::Mutex& getTheMutexForSchedulerUpdate() { return _mutexForSchedulerUpdate; }
     protected:
       bool checkBreakPoints();
       void waitResume();
diff --git a/src/engine/ForEachLoop.cxx b/src/engine/ForEachLoop.cxx
index 1dc70d97e..325d05801 100644
--- a/src/engine/ForEachLoop.cxx
+++ b/src/engine/ForEachLoop.cxx
@@ -21,6 +21,9 @@
 #include "TypeCode.hxx"
 #include "Visitor.hxx"
 #include "ComposedNode.hxx"
+#include "Executor.hxx"
+#include "AutoLocker.hxx"
+
 #include <iostream>
 #include <sstream>
 
@@ -852,3 +855,35 @@ std::string ForEachLoop::getProgress() const
     aProgress << "0";
   return aProgress.str();
 }
+
+/*!
+ * This method allows to retrieve the state of \a this during execution or after. This method works even if this is \b NOT complete, or during execution or after a failure in \a this.
+ * The typical usage of this method is to retrieve the results of items that passed successfully to avoid to lose all of them if only one fails.
+ * This method has one input \a execut and 3 outputs.
+ *
+ * \param [in] execut - The single input is for threadsafety reasons because this method can be called safely during the execution of \a this.
+ * \param [out] outputs - For each output ports in \a this linked with nodes sharing the same father than \a this the passed results are stored.
+ *                        All of the items in \a outputs have the same size.
+ * \param [out] nameOfOutputs - The array with same size than \a outputs, that tells for each item in outputs the output port it refers to.
+ * \return the list of ids among \c this->edGetSeqOfSamplesPort() that run successfully. The length of this returned array will be the length of all
+ *         SequenceAny objects contained in \a outputs.
+ *
+ * \sa edGetSeqOfSamplesPort
+ */
+std::vector<unsigned int> ForEachLoop::getPassedResults(Executor *execut, std::vector<SequenceAny *>& outputs, std::vector<std::string>& nameOfOutputs) const
+{
+  YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&(execut->getTheMutexForSchedulerUpdate()));
+  if(_execVals.empty())
+    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);
+  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 _execVals[0]->getSetItems();
+}
+
diff --git a/src/engine/ForEachLoop.hxx b/src/engine/ForEachLoop.hxx
index 8181cd8aa..f390dcd36 100644
--- a/src/engine/ForEachLoop.hxx
+++ b/src/engine/ForEachLoop.hxx
@@ -123,6 +123,8 @@ namespace YACS
       static const char NAME[];
     };
 
+    class Executor;
+
     class YACSLIBENGINE_EXPORT ForEachLoop : public DynParaLoop
     {
       friend class SplitterNode;
@@ -169,6 +171,9 @@ namespace YACS
       virtual std::string typeName() {return "YACS__ENGINE__ForEachLoop";}
       virtual void resetState(int level);
       std::string getProgress() const;
+#ifndef SWIG
+      std::vector<unsigned int> getPassedResults(Executor *execut, std::vector<SequenceAny *>& outputs, std::vector<std::string>& nameOfOutputs) const;
+#endif
     protected:
       Node *simpleClone(ComposedNode *father, bool editionOnly=true) const;
       void checkLinkPossibility(OutPort *start, const std::list<ComposedNode *>& pointsOfViewStart,
diff --git a/src/engine_swig/pilot.i b/src/engine_swig/pilot.i
index 6262f4773..4b58f258e 100644
--- a/src/engine_swig/pilot.i
+++ b/src/engine_swig/pilot.i
@@ -189,6 +189,8 @@ REFCOUNT_TEMPLATE(CompoInstmap,YACS::ENGINE::ComponentInstance)
 %newobject *::createStructTc;
 %newobject *::createType;
 
+%newobject YACS::ENGINE::SequenceAny::removeUnsetItemsFromThis;
+
 %newobject YACS::ENGINE::TypeCode::interfaceTc;
 %newobject YACS::ENGINE::TypeCode::sequenceTc;
 %newobject YACS::ENGINE::TypeCode::structTc;
@@ -371,6 +373,14 @@ EXCEPTION(YACS::ENGINE::ExecutorSwig::waitPause)
   }
 }
 
+%extend YACS::ENGINE::Any
+{
+  PyObject *getPyObj()
+  {
+    return (PyObject *)getRuntime()->convertNeutral(const_cast<YACS::ENGINE::TypeCode *>(self->getType()),self);
+  }
+}
+
 %newobject YACS::ENGINE::SequenceAny::__getitem__;
 %extend YACS::ENGINE::SequenceAny
 {
@@ -397,3 +407,30 @@ EXCEPTION(YACS::ENGINE::ExecutorSwig::waitPause)
     return a;
   }
 }
+
+%extend YACS::ENGINE::ForEachLoop
+{
+  PyObject *getPassedResults(Executor *execut) const
+  {
+    std::vector<SequenceAny *> ret1;
+    std::vector<std::string> ret2;
+    std::vector<unsigned int> ret0(self->getPassedResults(execut,ret1,ret2));
+    PyObject *ret(PyTuple_New(3));
+    // param 0
+    PyObject *ret0Py(PyList_New(ret0.size()));
+    for(std::size_t i=0;i<ret0.size();i++)
+      PyList_SetItem(ret0Py,i,PyInt_FromLong(ret0[i]));
+    PyTuple_SetItem(ret,0,ret0Py);
+    // param 1
+    PyObject *ret1Py(PyList_New(ret1.size()));
+    for(std::size_t i=0;i<ret1.size();i++)
+      PyList_SetItem(ret1Py,i,SWIG_NewPointerObj(SWIG_as_voidptr(ret1[i]),SWIGTYPE_p_YACS__ENGINE__SequenceAny, SWIG_POINTER_OWN | 0 ));
+    PyTuple_SetItem(ret,1,ret1Py);
+    // param 2
+    PyObject *ret2Py(PyList_New(ret2.size()));
+    for(std::size_t i=0;i<ret2.size();i++)
+      PyList_SetItem(ret2Py,i,PyString_FromString(ret2[i].c_str()));
+    PyTuple_SetItem(ret,2,ret2Py);
+    return ret;
+  }
+}
-- 
2.30.2