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