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

    12

    13

    14

class job(object):

    15

    """

    16

    job to be executed by a worker

    17

    """

    18

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

    19

        self.job = (call, args, kwargs)

    20

        self.result = None

    21

        self.success = None

    22

        self.exc_info = None

    23

    24

    def do(self):

    25

        """

    26

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

    27

        """

    28

        try:

    29

            call, args, kwargs = self.job

    30

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

    31

            self.success = True

    32

        except Exception as e:

    33

            self.success = False

    34

            self.exc_info = e

    35

    36

    37

class worker(object):

    38

    """

    39

    serialize main thread load/unload of PLC shared objects

    40

    """

    41

    def __init__(self):

    42

        # Only one job at a time

    43

        self._finish = False

    44

        self._threadID = None

    45

        self.mutex = Lock()

    46

        self.todo = Condition(self.mutex)

    47

        self.done = Condition(self.mutex)

    48

        self.free = Condition(self.mutex)

    49

        self.job = None

    50

        self.enabled = False

    51

        self.stopper = None

    52

        self.own_thread = None

    53

    54

    def reraise(self, job):

    55

        """

    56

        reraise exception happend in a job

    57

        @param job: job where original exception happend

    58

        """

    59

        raise job.exc_info

    60

    61

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

    62

        """

    63

        meant to be called by worker thread (blocking)

    64

        """

    65

        self._threadID = get_ident()

    66

        self.mutex.acquire()

    67

        self.enabled = True

    68

        if args or kwargs:

    69

            self.job = job(*args, **kwargs)

    70

            self.job.do()

    71

            # fail if first job fails

    72

            if not self.job.success:

    73

                self.reraise(self.job)

    74

            self.job = None

    75

    76

        self.free.notify()

    77

    78

        while not self._finish:

    79

            self.todo.wait_for(lambda: self.job is not None)

    80

            self.job.do()

    81

            self.done.notify()

    82

            self.job = None

    83

            self.free.notify()

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

    94

        self.stopper = stopper

    95

    96

        def do_pending_job():

    97

            self.mutex.acquire()

    98

            self.job.do()

    99

            self.done.notify_all()

   100

            self.mutex.release()

   101

   102

        def wakerfeedingloop():

   103

            self.mutex.acquire()

   104

            self.enabled = True

   105

   106

            # Handle first job

   107

            if args or kwargs:

   108

                self.job = job(*args, **kwargs)

   109

                waker(do_pending_job)

   110

                self.done.wait_for(lambda: self.job.success is not None)

   111

                # fail if first job fails

   112

                if not self.job.success:

   113

                    self.reraise(self.job)

   114

                self.job = None

   115

   116

            self.free.notify()

   117

   118

            while not self._finish:

   119

                self.todo.wait_for(lambda: self.job is not None)

   120

                if self._finish:

   121

                    break

   122

                waker(do_pending_job)

   123

                self.done.wait_for(lambda: self.job.success is not None)

   124

                self.job = None

   125

                self.free.notify()

   126

   127

            self.mutex.release()

   128

   129

        self.own_thread = Thread(target = wakerfeedingloop)

   130

        self.own_thread.start()

   131

   132

    def stop(self):

   133

        """

   134

        !interleave

   135

        """

   136

        self.mutex.acquire()

   137

        self._finish = True

   138

        self.enabled = False

   139

        self.job = None

   140

        self.todo.notify()

   141

        self.done.notify_all()

   142

        self.mutex.release()

   143

        self.own_thread.join()

   144

   145

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

   146

        """

   147

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

   148

        if job execution raise exception, re-raise same exception

   149

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

   150

        blocking until job done

   151

        """

   152

   153

        _job = job(*args, **kwargs)

   154

   155

        if self._threadID == get_ident():

   156

            # if caller is worker thread execute immediately

   157

            _job.do()

   158

        else:

   159

            # otherwise notify and wait for completion

   160

            self.mutex.acquire()

   161

            if not self.enabled:

   162

                self.mutex.release()

   163

                raise EOFError("Worker is disabled")

   164

   165

            self.free.wait_for(lambda: self.job is None)

   166

   167

            self.job = _job

   168

            self.todo.notify()

   169

            self.done.wait_for(lambda: _job.success is not None)

   170

            self.free.notify()

   171

            self.mutex.release()

   172

   173

        if _job.success is None:

   174

            raise EOFError("Worker job was interrupted")

   175

   176

        if _job.success:

   177

            return _job.result

   178

        else:

   179

            self.reraise(_job)

   180

   181

    def quit(self):

   182

        """

   183

        unblocks main thread, and terminate execution of runloop()

   184

        """

   185

        # mark queue

   186

        self._finish = True

   187

        self.mutex.acquire()

   188

        self.enabled = False

   189

        self.job = None

   190

        self.todo.notify()

   191

        self.done.notify()

   192

        self.mutex.release()

   193

   194

    def finish(self):

   195

        if self.own_thread is None:

   196

            self.quit()

   197

        if self.stopper is not None:

   198

            self.stopper()