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

    11

from threading import Lock, Condition, Thread

    12

    13

import _thread

    14

    15

    16

class job(object):

    17

    """

    18

    job to be executed by a worker

    19

    """

    20

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

    21

        self.job = (call, args, kwargs)

    22

        self.result = None

    23

        self.success = None

    24

        self.exc_info = None

    25

    26

    def do(self):

    27

        """

    28

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

    29

        """

    30

        try:

    31

            call, args, kwargs = self.job

    32

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

    33

            self.success = True

    34

        except Exception as e:

    35

            self.success = False

    36

            self.exc_info = e

    37

    38

    39

class worker(object):

    40

    """

    41

    serialize main thread load/unload of PLC shared objects

    42

    """

    43

    def __init__(self):

    44

        # Only one job at a time

    45

        self._finish = False

    46

        self._threadID = None

    47

        self.mutex = Lock()

    48

        self.todo = Condition(self.mutex)

    49

        self.done = Condition(self.mutex)

    50

        self.free = Condition(self.mutex)

    51

        self.job = None

    52

        self.enabled = False

    53

        self.stopper = None

    54

        self.own_thread = None

    55

    56

    def reraise(self, job):

    57

        """

    58

        reraise exception happend in a job

    59

        @param job: job where original exception happend

    60

        """

    61

        raise job.exc_info

    62

    63

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

    64

        """

    65

        meant to be called by worker thread (blocking)

    66

        """

    67

        self._threadID = _thread.get_ident()

    68

        self.mutex.acquire()

    69

        self.enabled = True

    70

        if args or kwargs:

    71

            _job = job(*args, **kwargs)

    72

            _job.do()

    73

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

    74

            if not _job.success:

    75

                self.reraise(_job)

    76

    77

        while not self._finish:

    78

            self.todo.wait()

    79

            if self.job is not None:

    80

                self.job.do()

    81

                self.done.notify()

    82

            else:

    83

                break

    84

    85

        self.mutex.release()

    86

    87

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

    88

        """

    89

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

    90

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

    91

        """

    92

        self.feed = Condition(self.mutex)

    93

        self._threadID = _thread.get_ident()

    94

        self.stopper = stopper

    95

    96

        def wakerfeedingloop():

    97

            self.mutex.acquire()

    98

            self.enabled = True

    99

            if args or kwargs:

   100

                def first_job_todo():

   101

                    _job = job(*args, **kwargs)

   102

                    _job.do()

   103

                    if not _job.success:

   104

                        self.reraise(_job)

   105

                    self.mutex.acquire()

   106

                    self.feed.notify()

   107

                    self.mutex.release()

   108

                waker(first_job_todo)

   109

                self.feed.wait()

   110

   111

            while not self._finish:

   112

                self.todo.wait()

   113

                def job_todo():

   114

                    self.mutex.acquire()

   115

                    if self.job is not None:

   116

                        self.job.do()

   117

                        self.feed.notify()

   118

                        self.done.notify()

   119

                    self.mutex.release()

   120

                if self._finish:

   121

                    break

   122

                waker(job_todo)

   123

                self.feed.wait()

   124

   125

            self.mutex.release()

   126

        self.own_thread = Thread(target = wakerfeedingloop)

   127

        self.own_thread.start()

   128

   129

    def stop(self):

   130

        """

   131

        !interleave

   132

        """

   133

        self.mutex.acquire()

   134

        self._finish = True

   135

        self.enabled = False

   136

        self.job = None

   137

        self.todo.notify()

   138

        self.done.notify()

   139

        self.mutex.release()

   140

        self.own_thread.join()

   141

   142

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

   143

        """

   144

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

   145

        if job execution raise exception, re-raise same exception

   146

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

   147

        blocking until job done

   148

        """

   149

   150

        _job = job(*args, **kwargs)

   151

   152

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

   153

            # if caller is worker thread execute immediately

   154

            _job.do()

   155

        else:

   156

            # otherwise notify and wait for completion

   157

            self.mutex.acquire()

   158

            if not self.enabled:

   159

                self.mutex.release()

   160

                raise EOFError("Worker is disabled")

   161

   162

            while self.job is not None:

   163

                self.free.wait()

   164

   165

            self.job = _job

   166

            self.todo.notify()

   167

            self.done.wait()

   168

            self.job = None

   169

            self.free.notify()

   170

            self.mutex.release()

   171

   172

        if _job.success is None:

   173

            raise EOFError("Worker job was interrupted")

   174

   175

        if _job.success:

   176

            return _job.result

   177

        else:

   178

            self.reraise(_job)

   179

   180

    def quit(self):

   181

        """

   182

        unblocks main thread, and terminate execution of runloop()

   183

        """

   184

        # mark queue

   185

        self._finish = True

   186

        self.mutex.acquire()

   187

        self.enabled = False

   188

        self.job = None

   189

        self.todo.notify()

   190

        self.done.notify()

   191

        self.mutex.release()

   192

   193

    def finish(self):

   194

        if self.own_thread is None:

   195

            self.quit()

   196

        if self.stopper is not None:

   197

            self.stopper()