1

#!/usr/bin/env python

     2

# -*- coding: utf-8 -*-

     3

     4

# This file is part of Beremiz runtime.

     5

#

     6

# Copyright (C) 2018: Edouard TISSERANT

     7

#

     8

# See COPYING.Runtime file for copyrights details.

     9

    10

from __future__ import absolute_import

    11

import sys

    12

from threading import Lock, Condition

    13

import six

    14

from six.moves import _thread

    15

    16

    17

class job(object):

    18

    """

    19

    job to be executed by a worker

    20

    """

    21

    def __init__(self, call, *args, **kwargs):

    22

        self.job = (call, args, kwargs)

    23

        self.result = None

    24

        self.success = None

    25

        self.exc_info = None

    26

    27

    def do(self):

    28

        """

    29

        do the job by executing the call, and deal with exceptions

    30

        """

    31

        try:

    32

            call, args, kwargs = self.job

    33

            self.result = call(*args, **kwargs)

    34

            self.success = True

    35

        except Exception:

    36

            self.success = False

    37

            self.exc_info = sys.exc_info()

    38

    39

    40

class worker(object):

    41

    """

    42

    serialize main thread load/unload of PLC shared objects

    43

    """

    44

    def __init__(self):

    45

        # Only one job at a time

    46

        self._finish = False

    47

        self._threadID = None

    48

        self.mutex = Lock()

    49

        self.todo = Condition(self.mutex)

    50

        self.done = Condition(self.mutex)

    51

        self.free = Condition(self.mutex)

    52

        self.job = None

    53

        self.enabled = False

    54

    55

    def reraise(self, job):

    56

        """

    57

        reraise exception happend in a job

    58

        @param job: job where original exception happend

    59

        """

    60

        exc_type = job.exc_info[0]

    61

        exc_value = job.exc_info[1]

    62

        exc_traceback = job.exc_info[2]

    63

        six.reraise(exc_type, exc_value, exc_traceback)

    64

    65

    def runloop(self, *args, **kwargs):

    66

        """

    67

        meant to be called by worker thread (blocking)

    68

        """

    69

        self._threadID = _thread.get_ident()

    70

        self.mutex.acquire()

    71

        self.enabled = True

    72

        if args or kwargs:

    73

            _job = job(*args, **kwargs)

    74

            _job.do()

    75

            # _job.success can't be None after do()

    76

            if not _job.success:

    77

                self.reraise(_job)

    78

    79

        while not self._finish:

    80

            self.todo.wait()

    81

            if self.job is not None:

    82

                self.job.do()

    83

                self.done.notify()

    84

            else:

    85

                break

    86

    87

        self.mutex.release()

    88

    89

    def call(self, *args, **kwargs):

    90

        """

    91

        creates a job, execute it in worker thread, and deliver result.

    92

        if job execution raise exception, re-raise same exception

    93

        meant to be called by non-worker threads, but this is accepted.

    94

        blocking until job done

    95

        """

    96

    97

        _job = job(*args, **kwargs)

    98

    99

        if self._threadID == _thread.get_ident():

   100

            # if caller is worker thread execute immediately

   101

            _job.do()

   102

        else:

   103

            # otherwise notify and wait for completion

   104

            self.mutex.acquire()

   105

            if not self.enabled:

   106

                self.mutex.release()

   107

                raise EOFError("Worker is disabled")

   108

   109

            while self.job is not None:

   110

                self.free.wait()

   111

   112

            self.job = _job

   113

            self.todo.notify()

   114

            self.done.wait()

   115

            self.job = None

   116

            self.free.notify()

   117

            self.mutex.release()

   118

   119

        if _job.success is None:

   120

            raise EOFError("Worker job was interrupted")

   121

   122

        if _job.success:

   123

            return _job.result

   124

        else:

   125

            self.reraise(_job)

   126

   127

    def quit(self):

   128

        """

   129

        unblocks main thread, and terminate execution of runloop()

   130

        """

   131

        # mark queue

   132

        self._finish = True

   133

        self.mutex.acquire()

   134

        self.enabled = False

   135

        self.job = None

   136

        self.todo.notify()

   137

        self.done.notify()

   138

        self.mutex.release()