AdaoExchangeLayer.cxx

   1 // Copyright (C) 2019 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.
   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, anthony.geay@edf.fr, EDF R&D
  20
  21 #include "AdaoExchangeLayer.hxx"
  22 #include "AdaoExchangeLayerException.hxx"
  23 #include "AdaoModelKeyVal.hxx"
  24 #include "PyObjectRAII.hxx"
  25 #include "Python.h"
  26
  27 #include "py2cpp/py2cpp.hxx"
  28
  29 #include <semaphore.h>
  30
  31 #include <iostream>
  32 #include <sstream>
  33 #include <clocale>
  34 #include <cstdlib>
  35 #include <thread>
  36 #include <future>
  37
  38 struct DataExchangedBetweenThreads // data written by subthread and read by calling thread
  39 {
  40 public:
  41   DataExchangedBetweenThreads();
  42   ~DataExchangedBetweenThreads();
  43 public:
  44   sem_t _sem;
  45   sem_t _sem_result_is_here;
  46   volatile bool _finished = false;
  47   volatile PyObject *_data = nullptr;
  48 };
  49
  50 /////////////////////////////////////////////
  51
  52 struct AdaoCallbackSt
  53 {
  54   PyObject_HEAD
  55   DataExchangedBetweenThreads *_data;
  56 };
  57
  58 static PyObject *adaocallback_call(AdaoCallbackSt *self, PyObject *args, PyObject *kw)
  59 {
  60   if(!PyTuple_Check(args))
  61     throw AdaoExchangeLayerException("Input args is not a tuple as expected !");
  62   if(PyTuple_Size(args)!=1)
  63     throw AdaoExchangeLayerException("Input args is not a tuple of size 1 as expected !");
  64   PyObjectRAII zeobj(PyObjectRAII::FromBorrowed(PyTuple_GetItem(args,0)));
  65   if(zeobj.isNull())
  66     throw AdaoExchangeLayerException("Retrieve of elt #0 of input tuple has failed !");
  67   volatile PyObject *ret(nullptr);
  68   PyThreadState *tstate(PyEval_SaveThread());// GIL is acquired (see ExecuteAsync). Before entering into non python section. Release lock
  69   {
  70     self->_data->_finished = false;
  71     self->_data->_data = zeobj;
  72     sem_post(&self->_data->_sem);
  73     sem_wait(&self->_data->_sem_result_is_here);
  74     ret = self->_data->_data;
  75   }
  76   PyEval_RestoreThread(tstate);//End of parallel section. Reaquire the GIL and restore the thread state
  77   return (PyObject *)ret;
  78 }
  79
  80 static int adaocallback___init__(PyObject *self, PyObject *args, PyObject *kwargs) { return 0; }
  81
  82 static PyObject *adaocallback___new__(PyTypeObject *type, PyObject *args, PyObject *kwargs)
  83 {
  84   return (PyObject *)( type->tp_alloc(type, 0) );
  85 }
  86
  87 static void adaocallback_dealloc(PyObject *self)
  88 {
  89   Py_TYPE(self)->tp_free(self);
  90 }
  91
  92 PyTypeObject AdaoCallbackType = {
  93   PyVarObject_HEAD_INIT(&PyType_Type, 0)
  94   "adaocallbacktype",
  95   sizeof(AdaoCallbackSt),
  96   0,
  97   adaocallback_dealloc,       /*tp_dealloc*/
  98   0,                          /*tp_print*/
  99   0,                          /*tp_getattr*/
 100   0,                          /*tp_setattr*/
 101   0,                          /*tp_compare*/
 102   0,                          /*tp_repr*/
 103   0,                          /*tp_as_number*/
 104   0,                          /*tp_as_sequence*/
 105   0,                          /*tp_as_mapping*/
 106   0,                          /*tp_hash*/
 107   (ternaryfunc)adaocallback_call,  /*tp_call*/
 108   0,                          /*tp_str*/
 109   0,                          /*tp_getattro*/
 110   0,                          /*tp_setattro*/
 111   0,                          /*tp_as_buffer*/
 112   Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC | Py_TPFLAGS_BASETYPE,  /*tp_flags*/
 113   0,                          /*tp_doc*/
 114   0,                          /*tp_traverse*/
 115   0,                          /*tp_clear*/
 116   0,                          /*tp_richcompare*/
 117   0,                          /*tp_weaklistoffset*/
 118   0,                          /*tp_iter*/
 119   0,                          /*tp_iternext*/
 120   0,                          /*tp_methods*/
 121   0,                          /*tp_members*/
 122   0,                          /*tp_getset*/
 123   0,                          /*tp_base*/
 124   0,                          /*tp_dict*/
 125   0,                          /*tp_descr_get*/
 126   0,                          /*tp_descr_set*/
 127   0,                          /*tp_dictoffset*/
 128   adaocallback___init__,      /*tp_init*/
 129   PyType_GenericAlloc,        /*tp_alloc*/
 130   adaocallback___new__,       /*tp_new*/
 131   PyObject_GC_Del,            /*tp_free*/
 132 };
 133
 134 /////////////////////////////////////////////
 135
 136 DataExchangedBetweenThreads::DataExchangedBetweenThreads()
 137 {
 138   if(sem_init(&_sem,0,0)!=0)// put value to 0 to lock by default
 139     throw AdaoExchangeLayerException("Internal constructor : Error on initialization of semaphore !");
 140   if(sem_init(&_sem_result_is_here,0,0)!=0)// put value to 0 to lock by default
 141     throw AdaoExchangeLayerException("Internal constructor : Error on initialization of semaphore !");
 142 }
 143
 144 DataExchangedBetweenThreads::~DataExchangedBetweenThreads()
 145 {
 146   sem_destroy(&_sem);
 147   sem_destroy(&_sem_result_is_here);
 148 }
 149
 150 class AdaoCallbackKeeper
 151 {
 152 public:
 153   void assign(AdaoCallbackSt *pt, DataExchangedBetweenThreads *data)
 154   {
 155     release();
 156     _pt = pt;
 157     _pt->_data = data;
 158   }
 159   PyObject *getPyObject() const { return reinterpret_cast<PyObject*>(_pt); }
 160   ~AdaoCallbackKeeper() { release(); }
 161 private:
 162   void release() { if(_pt) { Py_XDECREF(_pt); } }
 163 private:
 164   AdaoCallbackSt *_pt = nullptr;
 165 };
 166
 167 class AdaoExchangeLayer::Internal
 168 {
 169 public:
 170   Internal():_context(PyObjectRAII::FromNew(PyDict_New()))
 171   {
 172     AutoGIL agil;
 173     PyObject *mainmod(PyImport_AddModule("__main__"));
 174     PyObject *globals(PyModule_GetDict(mainmod));
 175     PyObject *bltins(PyEval_GetBuiltins());
 176     PyDict_SetItemString(_context,"__builtins__",bltins);
 177   }
 178 public:
 179   PyObjectRAII _context;
 180   PyObjectRAII _generate_case_func;
 181   PyObjectRAII _decorator_func;
 182   PyObjectRAII _adao_case;
 183   PyObjectRAII _execute_func;
 184   AdaoCallbackKeeper _py_call_back;
 185   std::future< void > _fut;
 186   PyThreadState *_tstate = nullptr;
 187   DataExchangedBetweenThreads _data_btw_threads;
 188 };
 189
 190 wchar_t **ConvertToWChar(int argc, const char *argv[])
 191 {
 192   wchar_t **ret(new wchar_t*[argc]);
 193   for(int i=0;i<argc;++i)
 194     {
 195       std::size_t len(strlen(argv[i])+1);
 196       wchar_t *elt(new wchar_t[len]);
 197       ret[i]=elt;
 198       std::mbstowcs(elt, argv[i], len);
 199     }
 200   return ret;
 201 }
 202
 203 void FreeWChar(int argc, wchar_t **tab)
 204 {
 205   for(int i=0;i<argc;++i)
 206     delete [] tab[i];
 207   delete [] tab;
 208 }
 209
 210 AdaoExchangeLayer::AdaoExchangeLayer()
 211 {
 212 }
 213
 214 AdaoExchangeLayer::~AdaoExchangeLayer()
 215 {
 216   AutoGIL agil;
 217   delete _internal;
 218 }
 219
 220 void AdaoExchangeLayer::init()
 221 {
 222   initPythonIfNeeded();
 223 }
 224
 225 PyObject *AdaoExchangeLayer::getPythonContext() const
 226 {
 227   if(!_internal)
 228     throw AdaoExchangeLayerException("getPythonContext : not initialized !");
 229   return _internal->_context;
 230 }
 231
 232 std::string AdaoExchangeLayer::printContext() const
 233 {
 234   AutoGIL agil;
 235   PyObject *obj(this->getPythonContext());
 236   if(!PyDict_Check(obj))
 237     throw AdaoExchangeLayerException("printContext : not a dict !");
 238   PyObject *key(nullptr), *value(nullptr);
 239   Py_ssize_t pos(0);
 240   std::ostringstream oss;
 241   while( PyDict_Next(obj, &pos, &key, &value) )
 242     {
 243       if(!PyUnicode_Check(key))
 244         throw AdaoExchangeLayerException("printContext : not a string as key !");
 245       oss << PyUnicode_AsUTF8(key) << " = ";
 246       PyObjectRAII reprOfValue(PyObjectRAII::FromNew(PyObject_Repr(value)));
 247       oss << PyUnicode_AsUTF8(reprOfValue);
 248       oss << std::endl;
 249     }
 250   return oss.str();
 251 }
 252
 253 /*!
 254  * AdaoExchangeLayer is based on multithreaded paradigm.
 255  * Master thread (thread calling this method) and slave thread (thread calling ADAO algo)
 256  * are calling both python interpretor. Consequence all python calls have to be surrounded with AGIL.
 257  *
 258  * User consequence : To avoid deadlocks this method release GIL. The downstream python calls must be with AGIL.
 259  *
 260  * This method initialize python interpretor if not already the case.
 261  * At the end of this method the lock is released to be ready to perform RAII on GIL
 262  * easily.
 263  */
 264 void AdaoExchangeLayer::initPythonIfNeeded()
 265 {
 266   if (!Py_IsInitialized())
 267     {
 268       const char *TAB[]={"AdaoExchangeLayer"};
 269       wchar_t **TABW(ConvertToWChar(1,TAB));
 270       // Python is not initialized
 271       Py_SetProgramName(const_cast<wchar_t *>(TABW[0]));
 272       Py_Initialize(); // Initialize the interpreter
 273       PySys_SetArgv(1,TABW);
 274       FreeWChar(1,TABW);
 275       PyEval_InitThreads();
 276       delete _internal;
 277       _internal = new Internal;
 278       _internal->_tstate=PyEval_SaveThread(); // release the lock acquired in AdaoExchangeLayer::initPythonIfNeeded by PyEval_InitThreads()
 279     }
 280   else
 281     {
 282       delete _internal;
 283       _internal = new Internal;
 284       if( PyGILState_Check() )// is the GIL already acquired (typically by a PyEval_InitThreads) ?
 285         _internal->_tstate=PyEval_SaveThread(); // release the lock acquired upstream
 286     }
 287 }
 288
 289 class Visitor1 : public AdaoModel::PythonLeafVisitor
 290 {
 291 public:
 292   Visitor1(PyObjectRAII func, PyObject *context):_func(func),_context(context)
 293   {
 294   }
 295
 296   void visit(AdaoModel::MainModel *godFather, AdaoModel::PyObjKeyVal *obj) override
 297   {
 298     if(obj->getKey()=="Matrix" || obj->getKey()=="DiagonalSparseMatrix")
 299       {
 300         std::ostringstream oss; oss << "__" << _cnt++;
 301         std::string varname(oss.str());
 302         obj->setVal(Py_None);
 303         PyDict_SetItemString(_context,varname.c_str(),Py_None);
 304         obj->setVarName(varname);
 305         return ;
 306       }
 307     if(obj->getKey()=="OneFunction")
 308       {
 309         std::ostringstream oss; oss << "__" << _cnt++;
 310         std::string varname(oss.str());
 311         obj->setVal(_func);
 312         PyDict_SetItemString(_context,varname.c_str(),_func);
 313         obj->setVarName(varname);
 314         return ;
 315       }
 316   }
 317 private:
 318   unsigned int _cnt = 0;
 319   PyObjectRAII _func;
 320   PyObject *_context = nullptr;
 321 };
 322
 323 void AdaoExchangeLayer::setFunctionCallbackInModel(AdaoModel::MainModel *model)
 324 {
 325   AutoGIL agil;
 326   const char DECORATOR_FUNC[]="def DecoratorAdao(cppFunc):\n"
 327       "    def evaluator( xserie ):\n"
 328       "        import numpy as np\n"
 329       "        yserie = [np.array(elt) for elt in cppFunc(xserie)]\n"
 330       "        return yserie\n"
 331       "    return evaluator\n";
 332   this->_internal->_py_call_back.assign(PyObject_GC_New(AdaoCallbackSt,&AdaoCallbackType),
 333       &this->_internal->_data_btw_threads);
 334   PyObject *callbackPyObj(this->_internal->_py_call_back.getPyObject());
 335   //
 336   {
 337       PyObjectRAII res(PyObjectRAII::FromNew(PyRun_String(DECORATOR_FUNC,Py_file_input,this->_internal->_context,this->_internal->_context)));
 338       PyObjectRAII decoratorGenerator( PyObjectRAII::FromBorrowed(PyDict_GetItemString(this->_internal->_context,"DecoratorAdao")) );
 339       if(decoratorGenerator.isNull())
 340         throw AdaoExchangeLayerException("Fail to locate DecoratorAdao function !");
 341       PyObjectRAII args(PyObjectRAII::FromNew(PyTuple_New(1)));
 342       { PyTuple_SetItem(args,0,callbackPyObj); Py_XINCREF(callbackPyObj); }
 343       this->_internal->_decorator_func = PyObjectRAII::FromNew(PyObject_CallObject(decoratorGenerator,args));
 344       if(this->_internal->_decorator_func.isNull())
 345         throw AdaoExchangeLayerException("Fail to generate result of DecoratorAdao function !");
 346   }
 347   //
 348   Visitor1 visitor(this->_internal->_decorator_func,this->_internal->_context);
 349   model->visitPythonLeaves(&visitor);
 350 }
 351
 352 void AdaoExchangeLayer::loadTemplate(AdaoModel::MainModel *model)
 353 {
 354   AutoGIL agil;
 355   {
 356     std::string sciptPyOfModelMaker(model->pyStr());
 357     PyObjectRAII res(PyObjectRAII::FromNew(PyRun_String(sciptPyOfModelMaker.c_str(),Py_file_input,this->_internal->_context,this->_internal->_context)));
 358     PyErr_Print();
 359     _internal->_adao_case = PyObjectRAII::FromNew(PyDict_GetItemString(this->_internal->_context,"case"));
 360   }
 361   if(_internal->_adao_case.isNull())
 362     throw AdaoExchangeLayerException("Fail to generate ADAO case object !");
 363   //
 364   _internal->_execute_func=PyObjectRAII::FromNew(PyObject_GetAttrString(_internal->_adao_case,"execute"));
 365   if(_internal->_execute_func.isNull())
 366     throw AdaoExchangeLayerException("Fail to locate execute function of ADAO case object !");
 367 }
 368
 369 void ExecuteAsync(PyObject *pyExecuteFunction, DataExchangedBetweenThreads *data)
 370 {
 371   {
 372     AutoGIL gil; // launched in a separed thread -> protect python calls
 373     PyObjectRAII args(PyObjectRAII::FromNew(PyTuple_New(0)));
 374     PyObjectRAII nullRes(PyObjectRAII::FromNew(PyObject_CallObject(pyExecuteFunction,args)));// go to adaocallback_call
 375     PyErr_Print();
 376   }
 377   data->_finished = true;
 378   data->_data = nullptr;
 379   sem_post(&data->_sem);
 380 }
 381
 382 void AdaoExchangeLayer::execute()
 383 {
 384   _internal->_fut = std::async(std::launch::async,ExecuteAsync,_internal->_execute_func,&_internal->_data_btw_threads);
 385 }
 386
 387 bool AdaoExchangeLayer::next(PyObject *& inputRequested)
 388 {
 389   sem_wait(&_internal->_data_btw_threads._sem);
 390   if(_internal->_data_btw_threads._finished)
 391     {
 392       inputRequested = nullptr;
 393       return false;
 394     }
 395   else
 396     {
 397       inputRequested = (PyObject *)_internal->_data_btw_threads._data;
 398       return true;
 399     }
 400 }
 401
 402 void AdaoExchangeLayer::setResult(PyObject *outputAssociated)
 403 {
 404   _internal->_data_btw_threads._data = outputAssociated;
 405   _internal->_data_btw_threads._finished = false;
 406   sem_post(&_internal->_data_btw_threads._sem_result_is_here);
 407 }
 408
 409 PyObject *AdaoExchangeLayer::getResult()
 410 {
 411   _internal->_fut.wait();
 412   if(_internal->_tstate)
 413     PyEval_RestoreThread(_internal->_tstate);
 414   AutoGIL gil;
 415   // now retrieve case.get("Analysis")[-1]
 416   PyObjectRAII get_func_of_adao_case(PyObjectRAII::FromNew(PyObject_GetAttrString(_internal->_adao_case,"get")));
 417   if(get_func_of_adao_case.isNull())
 418     throw AdaoExchangeLayerException("Fail to locate \"get\" method from ADAO case !");
 419   PyObjectRAII all_intermediate_results;
 420   {// retrieve return data from case.get("Analysis")
 421     PyObjectRAII args(PyObjectRAII::FromNew(PyTuple_New(1)));
 422     PyTuple_SetItem(args,0,PyUnicode_FromString("Analysis"));
 423     all_intermediate_results=PyObjectRAII::FromNew(PyObject_CallObject(get_func_of_adao_case,args));
 424     if(all_intermediate_results.isNull())
 425       throw AdaoExchangeLayerException("Fail to retrieve result of case.get(\"Analysis\") !");
 426   }
 427   PyObjectRAII optimum;
 428   {
 429     PyObjectRAII param(PyObjectRAII::FromNew(PyLong_FromLong(-1)));
 430     optimum=PyObjectRAII::FromNew(PyObject_GetItem(all_intermediate_results,param));
 431     if(optimum.isNull())
 432       throw AdaoExchangeLayerException("Fail to retrieve result of last element of case.get(\"Analysis\") !");
 433   }
 434   /*PyObjectRAII code(PyObjectRAII::FromNew(Py_CompileString("case.get(\"Analysis\")[-1]","retrieve result",Py_file_input)));
 435   if(code.isNull())
 436     throw AdaoExchangeLayerException("Fail to compile code to retrieve result after ADAO computation !");
 437     PyObjectRAII res(PyObjectRAII::FromNew(PyEval_EvalCode(code,_internal->_context,_internal->_context)));*/
 438   return optimum.retn();
 439 }