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, Thread

    13

    14

import six

    15

from six.moves import _thread

    16

    17

    18

class job(object):

    19

    """

    20

    job to be executed by a worker

    21

    """

    22

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

    23

        self.job = (call, args, kwargs)

    24

        self.result = None

    25

        self.success = None

    26

        self.exc_info = None

    27

    28

    def do(self):

    29

        """

    30

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

    31

        """

    32

        try:

    33

            call, args, kwargs = self.job

    34

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

    35

            self.success = True

    36

        except Exception:

    37

            self.success = False

    38

            self.exc_info = sys.exc_info()

    39

    40

    41

class worker(object):

    42

    """

    43

    serialize main thread load/unload of PLC shared objects

    44

    """

    45

    def __init__(self):

    46

        # Only one job at a time

    47

        self._finish = False

    48

        self._threadID = None

    49

        self.mutex = Lock()

    50

        self.todo = Condition(self.mutex)

    51

        self.done = Condition(self.mutex)

    52

        self.free = Condition(self.mutex)

    53

        self.job = None

    54

        self.enabled = False

    55

        self.stopper = None

    56

        self.own_thread = None

    57

    58

    def reraise(self, job):

    59

        """

    60

        reraise exception happend in a job

    61

        @param job: job where original exception happend

    62

        """

    63

        exc_type = job.exc_info[0]

    64

        exc_value = job.exc_info[1]

    65

        exc_traceback = job.exc_info[2]

    66

        six.reraise(exc_type, exc_value, exc_traceback)

    67

    68

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

    69

        """

    70

        meant to be called by worker thread (blocking)

    71

        """

    72

        self._threadID = _thread.get_ident()

    73

        self.mutex.acquire()

    74

        self.enabled = True

    75

        if args or kwargs:

    76

            _job = job(*args, **kwargs)

    77

            _job.do()

    78

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

    79

            if not _job.success:

    80

                self.reraise(_job)

    81

    82

        while not self._finish:

    83

            self.todo.wait()

    84

            if self.job is not None:

    85

                self.job.do()

    86

                self.done.notify()

    87

            else:

    88

                break

    89

    90

        self.mutex.release()

    91

    92

    def interleave(self, waker, stopper, *args, **kwargs):

    93

        """

    94

        as for twisted reactor's interleave, it passes all jobs to waker func

    95

        additionaly, it creates a new thread to wait for new job.

    96

        """

    97

        self.feed = Condition(self.mutex)

    98

        self._threadID = _thread.get_ident()

    99

        self.stopper = stopper

   100

   101

        def wakerfeedingloop():

   102

            self.mutex.acquire()

   103

            self.enabled = True

   104

            if args or kwargs:

   105

                def first_job_todo():

   106

                    _job = job(*args, **kwargs)

   107

                    _job.do()

   108

                    if not _job.success:

   109

                        self.reraise(_job)

   110

                    self.mutex.acquire()

   111

                    self.feed.notify()

   112

                    self.mutex.release()

   113

                waker(first_job_todo)

   114

                self.feed.wait()

   115

   116

            while not self._finish:

   117

                self.todo.wait()

   118

                def job_todo():

   119

                    self.mutex.acquire()

   120

                    if self.job is not None:

   121

                        self.job.do()

   122

                        self.feed.notify()

   123

                        self.done.notify()

   124

                    self.mutex.release()

   125

                if self._finish:

   126

                    break

   127

                waker(job_todo)

   128

                self.feed.wait()

   129

   130

            self.mutex.release()

   131

        self.own_thread = Thread(target = wakerfeedingloop)

   132

        self.own_thread.start()

   133

   134

    def stop(self):

   135

        """

   136

        !interleave

   137

        """

   138

        self.mutex.acquire()

   139

        self._finish = True

   140

        self.enabled = False

   141

        self.job = None

   142

        self.todo.notify()

   143

        self.done.notify()

   144

        self.mutex.release()

   145

        self.own_thread.join()

   146

   147

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

   148

        """

   149

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

   150

        if job execution raise exception, re-raise same exception

   151

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

   152

        blocking until job done

   153

        """

   154

   155

        _job = job(*args, **kwargs)

   156

   157

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

   158

            # if caller is worker thread execute immediately

   159

            _job.do()

   160

        else:

   161

            # otherwise notify and wait for completion

   162

            self.mutex.acquire()

   163

            if not self.enabled:

   164

                self.mutex.release()

   165

                raise EOFError("Worker is disabled")

   166

   167

            while self.job is not None:

   168

                self.free.wait()

   169

   170

            self.job = _job

   171

            self.todo.notify()

   172

            self.done.wait()

   173

            self.job = None

   174

            self.free.notify()

   175

            self.mutex.release()

   176

   177

        if _job.success is None:

   178

            raise EOFError("Worker job was interrupted")

   179

   180

        if _job.success:

   181

            return _job.result

   182

        else:

   183

            self.reraise(_job)

   184

   185

    def quit(self):

   186

        """

   187

        unblocks main thread, and terminate execution of runloop()

   188

        """

   189

        # mark queue

   190

        self._finish = True

   191

        self.mutex.acquire()

   192

        self.enabled = False

   193

        self.job = None

   194

        self.todo.notify()

   195

        self.done.notify()

   196

        self.mutex.release()

   197

   198

    def finish(self):

   199

        if self.own_thread is None:

   200

            self.quit()

   201

        if self.stopper is not None:

   202

            self.stopper()