beremiz

Intermediate state while implementing runtime worker to ensure that PLCObject Load and Unload methods always run main thread.

2018-04-12, Edouard Tisserant
081265cda5b1
Parents edd7a3a06d86
Children 88048a0dd0c3
Intermediate state while implementing runtime worker to ensure that PLCObject Load and Unload methods always run main thread.
3 files changed, 189 insertions(+), 67 deletions(-)
  • +56 -50
    Beremiz_service.py
  • +132 -16
    runtime/PLCObject.py
  • +1 -1
    runtime/__init__.py
    • --- a/Beremiz_service.py Thu Apr 12 16:38:09 2018 +0200
    +++ b/Beremiz_service.py Thu Apr 12 22:20:38 2018 +0200
    @@ -35,7 +35,7 @@
    import __builtin__
    import Pyro.core as pyro
    -from runtime import PLCObject, ServicePublisher
    +from runtime import PLCObject, ServicePublisher, MainWorker
    import util.paths as paths
    @@ -401,7 +401,7 @@
    class Server(object):
    def __init__(self, servicename, ip_addr, port,
    - workdir, argv, autostart=False,
    + workdir, argv,
    statuschange=None, evaluator=default_evaluator,
    pyruntimevars=None):
    self.continueloop = True
    @@ -413,12 +413,11 @@
    self.argv = argv
    self.plcobj = None
    self.servicepublisher = None
    - self.autostart = autostart
    self.statuschange = statuschange
    self.evaluator = evaluator
    self.pyruntimevars = pyruntimevars
    - def Loop(self):
    + def PyroLoop(self):
    while self.continueloop:
    pyro.initServer()
    self.daemon = pyro.Daemon(host=self.ip_addr, port=self.port)
    @@ -426,7 +425,27 @@
    # taking too small timeout value may cause
    # unwanted diconnection when IDE is kept busy for long periods
    self.daemon.setTimeout(60)
    - self.Start()
    +
    + uri = self.daemon.connect(self.plcobj, "PLCObject")
    +
    + print(_("Pyro port :"), self.port)
    + print(_("Pyro object's uri :"), uri)
    +
    + # Beremiz IDE detects daemon start by looking
    + # for self.workdir in the daemon's stdout.
    + # Therefore don't delete the following line
    + print(_("Current working directory :"), self.workdir)
    +
    + # Configure and publish service
    + # Not publish service if localhost in address params
    + if self.servicename is not None and \
    + self.ip_addr is not None and \
    + self.ip_addr != "localhost" and \
    + self.ip_addr != "127.0.0.1":
    + print(_("Publishing service on local network"))
    + self.servicepublisher = ServicePublisher.ServicePublisher()
    + self.servicepublisher.RegisterService(self.servicename, self.ip_addr, self.port)
    +
    self.daemon.requestLoop()
    self.daemon.sock.close()
    @@ -440,38 +459,8 @@
    self.plcobj.UnLoadPLC()
    self._stop()
    - def Start(self):
    - self.plcobj = PLCObject(self.workdir, self.daemon, self.argv,
    - self.statuschange, self.evaluator,
    - self.pyruntimevars)
    -
    - uri = self.daemon.connect(self.plcobj, "PLCObject")
    -
    - print(_("Pyro port :"), self.port)
    - print(_("Pyro object's uri :"), uri)
    -
    - # Beremiz IDE detects daemon start by looking
    - # for self.workdir in the daemon's stdout.
    - # Therefore don't delete the following line
    - print(_("Current working directory :"), self.workdir)
    -
    - # Configure and publish service
    - # Not publish service if localhost in address params
    - if self.servicename is not None and \
    - self.ip_addr is not None and \
    - self.ip_addr != "localhost" and \
    - self.ip_addr != "127.0.0.1":
    - print(_("Publishing service on local network"))
    - self.servicepublisher = ServicePublisher.ServicePublisher()
    - self.servicepublisher.RegisterService(self.servicename, self.ip_addr, self.port)
    -
    - self.plcobj.AutoLoad()
    - if self.plcobj.GetPLCstatus()[0] != "Empty":
    - if self.autostart:
    - self.plcobj.StartPLC()
    - self.plcobj.StatusChange()
    -
    - sys.stdout.flush()
    + def RegisterPLCObject(self, plcobj):
    + self.plcobj = plcobj
    def _stop(self):
    if self.plcobj is not None:
    @@ -529,13 +518,13 @@
    return o.res
    pyroserver = Server(servicename, given_ip, port,
    - WorkingDir, argv, autostart,
    + WorkingDir, argv,
    statuschange, evaluator, pyruntimevars)
    taskbar_instance = BeremizTaskBarIcon(pyroserver, enablewx)
    else:
    pyroserver = Server(servicename, given_ip, port,
    - WorkingDir, argv, autostart,
    + WorkingDir, argv,
    statuschange, pyruntimevars=pyruntimevars)
    @@ -631,19 +620,36 @@
    except Exception:
    LogMessageAndException(_("WAMP client startup failed. "))
    +plcobj = PLCObject(pyroserver)
    +
    +plcobj.AutoLoad()
    +if plcobj.GetPLCstatus()[0] == "Stopped":
    + if autostart:
    + plcobj.StartPLC()
    +plcobj.StatusChange()
    +
    +pyro_thread = Thread(target=pyroserver.PyroLoop)
    +pyro_thread.start()
    +
    +sys.stdout.flush()
    if havetwisted or havewx:
    - pyro_thread = Thread(target=pyroserver.Loop)
    - pyro_thread.start()
    + if havetwisted:
    + # reactor._installSignalHandlersAgain()
    + def ui_thread_target():
    + # FIXME: had to disable SignaHandlers install because
    + # signal not working in non-main thread
    + reactor.run(installSignalHandlers=False)
    + else :
    + ui_thread_target = app.MainLoop
    - if havetwisted:
    - reactor.run()
    - elif havewx:
    - app.MainLoop()
    -else:
    - try:
    - pyroserver.Loop()
    - except KeyboardInterrupt:
    - pass
    + ui_thread = Thread(target = ui_thread_target)
    + ui_thread.start()
    +
    +try:
    + MainWorker.runloop()
    +except KeyboardInterrupt:
    + pass
    +
    pyroserver.Quit()
    sys.exit(0)
    --- a/runtime/PLCObject.py Thu Apr 12 16:38:09 2018 +0200
    +++ b/runtime/PLCObject.py Thu Apr 12 22:20:38 2018 +0200
    @@ -23,7 +23,8 @@
    from __future__ import absolute_import
    -from threading import Timer, Thread, Lock, Semaphore, Event
    +import thread
    +from threading import Timer, Thread, Lock, Semaphore, Event, Condition
    import ctypes
    import os
    import sys
    @@ -60,28 +61,133 @@
    sys.stdout.flush()
    +class job(object):
    + """
    + job to be executed by a worker
    + """
    + def __init__(self,call,*args,**kwargs):
    + self.job = (call,args,kwargs)
    + self.result = None
    + self.success = False
    + self.exc_info = None
    +
    + def do(self):
    + """
    + do the job by executing the call, and deal with exceptions
    + """
    + try :
    + call, args, kwargs = self.job
    + self.result = call(*args,**kwargs)
    + self.success = True
    + except Exception:
    + self.success = False
    + self.exc_info = sys.exc_info()
    +
    +
    +class worker(object):
    + """
    + serialize main thread load/unload of PLC shared objects
    + """
    + def __init__(self):
    + # Only one job at a time
    + self._finish = False
    + self._threadID = None
    + self.mutex = Lock()
    + self.todo = Condition(self.mutex)
    + self.done = Condition(self.mutex)
    + self.job = None
    +
    + def runloop(self):
    + """
    + meant to be called by worker thread (blocking)
    + """
    + self._threadID = thread.get_ident()
    + self.mutex.acquire()
    + while not self._finish:
    + self.todo.wait()
    + if self.job is not None:
    + self.job.do()
    + self.done.notify_all()
    + self.mutex.release()
    +
    + def call(self, *args, **kwargs):
    + print("call", args, kwargs)
    + """
    + creates a job, execute it in worker thread, and deliver result.
    + if job execution raise exception, re-raise same exception
    + meant to be called by non-worker threads, but this is accepted.
    + blocking until job done
    + """
    +
    + _job = job(*args,**kwargs)
    +
    + if self._threadID == thread.get_ident():
    + # if caller is worker thread execute immediately
    + _job.do()
    + else:
    + # otherwise notify and wait for completion
    + self.mutex.acquire()
    +
    + while self.job is not None:
    + self.done.wait()
    +
    + self.job = _job
    + self.todo.notify()
    + self.done.wait()
    + _job = self.job
    + self.job = None
    + self.mutex.release()
    +
    + if _job.success:
    + return _job.result
    + else:
    + raise _job.exc_info[0], _job.exc_info[1], _job.exc_info[2]
    +
    + def quit(self):
    + """
    + unblocks main thread, and terminate execution of runloop()
    + """
    + # mark queue
    + self._finish = True
    + self.mutex.acquire()
    + self.job = None
    + self.todo.notify()
    + self.mutex.release()
    +
    +
    +MainWorker = worker()
    +
    +
    +def RunInMain(func):
    + def func_wrapper(*args,**kwargs):
    + return MainWorker.call(func, *args, **kwargs)
    + return func_wrapper
    +
    +
    class PLCObject(pyro.ObjBase):
    - def __init__(self, workingdir, daemon, argv, statuschange, evaluator, pyruntimevars):
    + def __init__(self, server):
    +
    pyro.ObjBase.__init__(self)
    - self.evaluator = evaluator
    - self.argv = [workingdir] + argv # force argv[0] to be "path" to exec...
    - self.workingdir = workingdir
    + self.evaluator = server.evaluator
    + self.argv = [server.workdir] + server.argv # force argv[0] to be "path" to exec...
    + self.workingdir = server.workdir
    self.PLCStatus = "Empty"
    self.PLClibraryHandle = None
    self.PLClibraryLock = Lock()
    self.DummyIteratorLock = None
    # Creates fake C funcs proxies
    - self._FreePLC()
    - self.daemon = daemon
    - self.statuschange = statuschange
    + self._InitPLCStubCalls()
    + self.daemon = server.daemon
    + self.statuschange = server.statuschange
    self.hmi_frame = None
    - self.pyruntimevars = pyruntimevars
    + self.pyruntimevars = server.pyruntimevars
    self._loading_error = None
    self.python_runtime_vars = None
    self.TraceThread = None
    self.TraceLock = Lock()
    self.TraceWakeup = Event()
    self.Traces = []
    + server.RegisterPLCObject(self)
    def AutoLoad(self):
    # Get the last transfered PLC if connector must be restart
    @@ -145,6 +251,7 @@
    def _GetLibFileName(self):
    return os.path.join(self.workingdir, self.CurrentPLCFilename)
    + @RunInMain
    def LoadPLC(self):
    """
    Load PLC library
    @@ -233,19 +340,18 @@
    except Exception:
    self._loading_error = traceback.format_exc()
    PLCprint(self._loading_error)
    + self._FreePLC()
    return False
    + @RunInMain
    def UnLoadPLC(self):
    self.PythonRuntimeCleanup()
    self._FreePLC()
    - def _FreePLC(self):
    + def _InitPLCStubCalls(self):
    """
    - Unload PLC library.
    - This is also called by __init__ to create dummy C func proxies
    + create dummy C func proxies
    """
    - self.PLClibraryLock.acquire()
    - # Forget all refs to library
    self._startPLC = lambda x, y: None
    self._stopPLC = lambda: None
    self._ResetDebugVariables = lambda: None
    @@ -259,11 +365,22 @@
    self._GetLogCount = None
    self._LogMessage = None
    self._GetLogMessage = None
    + self._PLClibraryHandle = None
    self.PLClibraryHandle = None
    +
    + def _FreePLC(self):
    + """
    + Unload PLC library.
    + This is also called by __init__ to create dummy C func proxies
    + """
    + self.PLClibraryLock.acquire()
    +
    # Unload library explicitely
    if getattr(self, "_PLClibraryHandle", None) is not None:
    dlclose(self._PLClibraryHandle)
    - self._PLClibraryHandle = None
    +
    + # Forget all refs to library
    + self._InitPLCStubCalls()
    self.PLClibraryLock.release()
    return False
    @@ -448,7 +565,6 @@
    self.PLCStatus = "Stopped"
    else:
    self.PLCStatus = "Broken"
    - self._FreePLC()
    self.StatusChange()
    return self.PLCStatus == "Stopped"
    --- a/runtime/__init__.py Thu Apr 12 16:38:09 2018 +0200
    +++ b/runtime/__init__.py Thu Apr 12 22:20:38 2018 +0200
    @@ -24,5 +24,5 @@
    from __future__ import absolute_import
    import os
    -from runtime.PLCObject import PLCObject, PLCprint
    +from runtime.PLCObject import PLCObject, PLCprint, MainWorker
    import runtime.ServicePublisher