openrvdas.logger.readers.timeout_reader

  1#!/usr/bin/env python3
  2
  3import threading
  4import time
  5
  6from logger.readers.reader import Reader  # noqa: E402
  7
  8
  9##############################
 10class ReaderTimeout(StopIteration):
 11    """A custom exception we can raise when we hit timeout."""
 12    pass
 13
 14
 15################################################################################
 16class TimeoutReader(Reader):
 17    """Instantiated with a client Reader instance (such as a
 18    NetworkReader), an interval, a timeout and optional message. When its
 19    read() method is called, it iteratively calls its passed reader's
 20    read() method every interval seconds, discarding the received
 21    output. It only returns if/when the client reader fails to return a
 22    record within timeout seconds, in which case it returns either the
 23    passed timeout message or a default one, warning that no records have
 24    been received within the specified timeout.
 25
 26    In general, it's better if you can structure your logger configuration
 27    so that it uses TimeoutWriters rather than TimeoutReaders. The former
 28    are more robust and less computationally intensive.
 29    """
 30    ############################
 31
 32    def __init__(self, reader, timeout, message=None, resume_message=None,
 33                 empty_is_okay=False, none_is_okay=False, **kwargs):
 34        """
 35        ```
 36        reader         A client reader instance
 37
 38        timeout        Timeout interval in seconds
 39
 40        message        Message to be returned if client reader fails to return
 41                       a record within the timeout interval
 42
 43        resume_message Message to be returned when client returns a record after
 44                       having timed out
 45
 46        empty_is_okay If True, receiving an empty record is sufficient to reset
 47                      the timer.
 48        none_is_okay  If True, receiving a 'None' record is sufficient to reset
 49                        the timer.
 50        ```
 51        Sample:
 52        ```
 53        gyr1_reader = ComposedReader(NetworkReader(':6224'),
 54                                     RegexFilterTransform('^gyr1'))
 55        reader = TimeoutReader(reader=gyr1_reader,
 56                               timeout=15,
 57                               message='No Gyroscope records received for 15 seconds')
 58        ```
 59        """
 60        super().__init__(**kwargs)
 61
 62        self.reader = reader
 63        self.timeout = timeout
 64        self.message = message or ('Timeout: no %s record received in %d seconds'
 65                                   % (reader, timeout))
 66        self.resume_message = resume_message or ('Timeout: %s record received'
 67                                                 % reader)
 68        self.empty_is_okay = empty_is_okay
 69        self.none_is_okay = none_is_okay
 70
 71        # When we got our last record (or were instantiated)
 72        self.last_record = time.time()
 73
 74        # Keep track of whether we're currently timed out or not
 75        self.timed_out = False
 76
 77        # Protect self.last_record and self.timed_out
 78        self.timeout_lock = threading.Lock()
 79
 80        # To let us cleanly exit _timeout_thread; this gets set to False
 81        # when our read() method is called.
 82        self.keep_reading = False
 83
 84        # Placeholder for the timeout loop we'll run in a separate thread
 85        # when called.
 86        self.timeout_thread = None
 87
 88    ############################
 89    def __del__(self):
 90        self.quit()
 91
 92    ############################
 93    def quit(self):
 94        self.keep_reading = False
 95
 96    ############################
 97    def _timeout_thread(self):
 98        """Repeatedly call the client read() method, and keep track of when we
 99        get records from it.
100        """
101        while self.keep_reading:
102            # Loop until we get a record that matches our standards
103            record = None
104            while not record:
105                record = self.reader.read()
106                if self.empty_is_okay:
107                    break
108                if record is None and self.none_is_okay:
109                    break
110
111            # We've got a record!
112            with self.timeout_lock:
113                self.last_record = time.time()
114                # If we were timed out, we aren't anymore, and the read will
115                # return. Stop trying to read from our client (until we get
116                # another read() request).
117                if self.timed_out:
118                    self.keep_reading = False
119
120    ############################
121    def read(self):
122        """Block and return either a 'timeout' message or a 'resumed' message,
123        only when our client reader either hasn't given us a record in N
124        seconds or has resumed giving us records after having timed out.
125        """
126        # Don't sleep the full interval because we want to quickly catch
127        # if we get a 'resume'
128        max_sleep_interval = 1  # Don't sleep more than one second
129
130        # We want the timeout_thread to start calling records for us.
131        self.keep_reading = True
132
133        # Set our 'last_record' to 'now' so that we start counting toward
134        # the timeout from now
135        with self.timeout_lock:
136            self.last_record = time.time()
137
138        # Start up timeout_thread if it's not running
139        if not self.timeout_thread or not self.timeout_thread.is_alive():
140            self.timeout_thread = threading.Thread(target=self._timeout_thread,
141                                                   name='timeout_thread', daemon=True)
142            self.timeout_thread.start()
143
144        # Loop until 1) we've timed out, 2) we've resumed after a timeout,
145        # or 3) we've been told to quit
146        while self.keep_reading:
147            now = time.time()
148            with self.timeout_lock:
149                time_since_last_record = now - self.last_record
150
151                # If we're timed out and we've seen a record inside our
152                # timeout window, we're not timed out anymore. Return a
153                # 'resumed' message and stop trying to read.
154                if self.timed_out and time_since_last_record < self.timeout:
155                    self.timed_out = False
156                    self.keep_reading = False
157                    return self.resume_message
158
159                # Otherwise, figure how long to sleep before we need to see
160                # our next record.
161                time_to_sleep = self.timeout - time_since_last_record
162
163                # If we're overdue for a record...
164                if time_to_sleep < 0:
165                    # If we weren't already timed out, we are now. Send message.
166                    if not self.timed_out:
167                        self.timed_out = True
168                        self.keep_reading = False
169                        return self.message
170
171                    # Check again in a little while
172                    time_to_sleep = max_sleep_interval
173
174            # Whether or not we're timed out, snooze a bit before checking
175            # again.
176            time.sleep(min(time_to_sleep, max_sleep_interval))