runtime/Worker.py
author Andrey Skvortsov <andrej.skvortzov@gmail.com>
Wed, 13 Mar 2019 14:27:24 +0300
changeset 2542 a3ec35ee94e7
parent 2537 eb4a4cc41914
child 2604 c8a25a3a7f8b
permissions -rw-r--r--
Fix crash in runtime if PSK secret is missing

./Beremiz_service.py -s $PWD/psk2.txt -n beremiz /tmp/beremiz
Beremiz_service: 1.2-1378c18402c3+

Traceback (most recent call last):
File "./Beremiz_service.py", line 511, in <module>
ensurePSK(servicename, PSKpath)
File "/home/developer/WorkData/PLC/beremiz/beremiz/runtime/Stunnel.py", line 32, in ensurePSK
PSKgen(ID, PSKpath)
File "/home/developer/WorkData/PLC/beremiz/beremiz/runtime/Stunnel.py", line 23, in PSKgen
call(restart_stunnel_cmdline)
File "/home/developer/WorkData/PLC/beremiz/beremiz/runtime/spawn_subprocess.py", line 116, in call
pid = posix_spawn.posix_spawnp(cmd[0], cmd)
File "/home/developer/.local/lib/python2.7/site-packages/posix_spawn/_impl.py", line 120, in posix_spawnp
return _posix_spawn(lib.posix_spawnp, *args, **kwargs)
File "/home/developer/.local/lib/python2.7/site-packages/posix_spawn/_impl.py", line 111, in _posix_spawn
_check_error(res, path)
File "/home/developer/.local/lib/python2.7/site-packages/posix_spawn/_impl.py", line 10, in _check_error
raise OSError(errno, os.strerror(errno), path)
OSError: [Errno 2] No such file or directory: '/etc/init.d/S50stunnel'
#!/usr/bin/env python
# -*- coding: utf-8 -*-

# This file is part of Beremiz runtime.
#
# Copyright (C) 2018: Edouard TISSERANT
#
# See COPYING.Runtime file for copyrights details.

from __future__ import absolute_import
import sys
from threading import Lock, Condition
import six
from six.moves import _thread


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.free = Condition(self.mutex)
        self.job = None

    def reraise(self, job):
        """
        reraise exception happend in a job
        @param job: job where original exception happend
        """
        exc_type = job.exc_info[0]
        exc_value = job.exc_info[1]
        exc_traceback = job.exc_info[2]
        six.reraise(exc_type, exc_value, exc_traceback)

    def runloop(self, *args, **kwargs):
        """
        meant to be called by worker thread (blocking)
        """
        self._threadID = _thread.get_ident()
        self.mutex.acquire()
        if args or kwargs:
            _job = job(*args, **kwargs)
            _job.do()
            if not _job.success:
                self.reraise(_job)

        while not self._finish:
            self.todo.wait()
            if self.job is not None:
                self.job.do()
                self.done.notify()
            else:
                break

        self.mutex.release()

    def call(self, *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.free.wait()

            self.job = _job
            self.todo.notify()
            self.done.wait()
            self.job = None
            self.free.notify()
            self.mutex.release()

        if _job.success:
            return _job.result
        else:
            self.reraise(_job)

    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()