openrvdas.logger.readers.composed_reader

No module-level documentation available.
  1#!/usr/bin/env python3
  2
  3import logging
  4import threading
  5
  6from logger.readers.reader import Reader  # noqa: E402
  7
  8# How long to a reader thread should lie dormant before shutting down
  9# and counting on getting restarted again if/when needed. We need this
 10# so that our readers eventually terminate.
 11READER_TIMEOUT_WAIT = 0.25
 12
 13
 14################################################################################
 15class ComposedReader(Reader):
 16    """
 17    Read lines from one or more Readers (in parallel) and process their
 18    responses through zero or more Transforms (in series).
 19
 20    NOTE: we make the rash assumption that transforms are thread-safe,
 21    that is, that no mischief or corrupted internal state will result if
 22    more than one thread calls a transform at the same time. To be
 23    thread-safe, a transform must protect any changes to its internal
 24    state with a non-re-entrant thread lock, as described in the threading
 25    module.
 26
 27    Also NOTE: Most of the messy logic in this class comes from the desire
 28    to only call read() on our component readers when we actually need new
 29    records (NOTE: this desire may be misplaced!).
 30
 31    So when we get a request, we fire up threads and ask each of our
 32    readers for a record. We return the first one we get, and let the
 33    others pile up in a queue that we'll feed from the next time we're
 34    asked.
 35
 36    But we don't want to fire up a new thread for each reader every time
 37    the queue is empty, so we have threads (in run_reader()) hang out for
 38    a little while, waiting for another queue_needs_record event. If they
 39    get one, the call their own read() methods again. If they haven't been
 40    called on in READER_TIMEOUT_WAIT seconds, they exit, but will get
 41    fired up again by read() if/when the queue is empty and we're is asked
 42    for another record.
 43
 44    It's important to have the run_reader threads time out, or any process
 45    using a ComposedReader will never naturally terminate.
 46    """
 47    ############################
 48
 49    def __init__(self, readers, transforms=[], **kwargs):
 50        """
 51        Instantiation:
 52        ```
 53        reader = ComposedReader(readers, transforms=[], check_format=True)
 54
 55        readers        A single Reader or a list of Readers.
 56
 57        transforms     A single Transform or list of zero or more Transforms.
 58        ```
 59        Use:
 60        ```
 61        record = reader.read()
 62        ```
 63        Sample:
 64        ```
 65        reader = ComposedReader(readers=[UDPReader(port=6221),
 66                                         UDPReader(port=6223)],
 67                                transforms=[TimestampTransform()])
 68        ```
 69        """
 70        super().__init__(**kwargs)
 71
 72        # Make readers a list, even if it's only a single reader.
 73        self.readers = readers if type(readers) is list else [readers]
 74        self.num_readers = len(self.readers)
 75
 76        # Transforms can be empty. But if not empty, make it a list, even
 77        # if it's only a single transform.
 78        if not isinstance(transforms, list):
 79            self.transforms = [transforms]
 80        else:
 81            self.transforms = transforms
 82
 83        # List where we're going to store reader threads
 84        self.reader_threads = [None] * self.num_readers
 85
 86        # Whether reader[i] has returned EOF since we've last asked it
 87        self.reader_returned_eof = [False] * self.num_readers
 88
 89        # One lock per reader, to save us from accidental re-entry
 90        self.reader_locks = [threading.Lock() for i in range(self.num_readers)]
 91
 92        # Queue where we'll store extra records, and lock so only one
 93        # thread can touch queue at a time
 94        self.queue = []
 95        self.queue_lock = threading.Lock()
 96
 97        # The two events, queue_has_record and queue_needs_record interact
 98        # in a sort of a dance:
 99        #
100        #  has = False, need = False: Everything is quiescent
101        #  has = False, need = True:  A request has been made, call readers
102        #  has = True,  need = True:  Momentary condition when we get needed rec
103        #  has = True,  need = False: We've got spare records in the queue
104        #
105        # Set when the queue is empty and we need a record
106        self.queue_needs_record = threading.Event()
107
108        # Set when a reader adds something to the queue
109        self.queue_has_record = threading.Event()
110
111    ############################
112    def read(self):
113        """
114        Get the next record from queue or readers.
115        """
116        # If we only have one reader, there's no point making things
117        # complicated. Just read, transform, return.
118        if len(self.readers) == 1:
119            return self._apply_transforms(self.readers[0].read())
120
121        # Do we have anything in the queue? Note: safe to check outside of
122        # lock, because we're the only method that actually *removes*
123        # anything. So if tests True here, we're assured that there's
124        # something there, and we lock before retrieving it. Advantage of
125        # doing it this way is that we don't tie up queue lock while
126        # processing transforms.
127        if self.queue:
128            logging.debug('read() - read requested; queue len is %d',
129                          len(self.queue))
130            with self.queue_lock:
131                record = self.queue.pop(0)
132                return self._apply_transforms(record)
133
134        # If here, nothing's in the queue. Note that, if we wanted to be
135        # careful to never unnecessarily ask for more records, we should
136        # put a lock around this, but the failure mode is somewhat benign:
137        # we ask for more records when some are already on the way.
138        logging.debug('read() - read requested and nothing in the queue.')
139
140        # Some threads may have timed out while waiting to be called to
141        # action; restart them.
142        for i in range(len(self.readers)):
143            if not self.reader_threads[i] \
144               or not self.reader_threads[i].is_alive() \
145               and not self.reader_returned_eof[i]:
146                logging.info('read() - starting thread for Reader #%d', i)
147                self.reader_returned_eof[i] = False
148                thread = threading.Thread(target=self._run_reader, args=(i,),
149                                          daemon=True)
150                self.reader_threads[i] = thread
151                thread.start()
152
153        # Now notify all threads that we do in fact need a record.
154        self.queue_needs_record.set()
155
156        # Keep checking/sleeping until we've either got a record in the
157        # queue or all readers have given us an EOF.
158        while False in self.reader_returned_eof:
159            logging.debug('read() - waiting for queue lock')
160            with self.queue_lock:
161                logging.debug('read() - acquired queue lock, queue length is %d',
162                              len(self.queue))
163                if self.queue:
164                    record = self.queue.pop(0)
165                    if not self.queue:
166                        self.queue_has_record.clear()  # only set/clear inside queue_lock
167
168                    logging.debug('read() - got record')
169                    return self._apply_transforms(record)
170                else:
171                    self.queue_has_record.clear()
172
173            # If here, nothing in queue yet. Wait
174            logging.debug('read() - clear of queue lock, waiting for record')
175            self.queue_has_record.wait(READER_TIMEOUT_WAIT)
176
177            if not self.queue_has_record.is_set():
178                logging.debug('read() - timed out waiting for record. Looping')
179            logging.debug('read() - readers returned EOF: %s',
180                          self.reader_returned_eof)
181
182        # All readers have given us an EOF
183        logging.debug('read() - all threads returned None; returning None')
184        return None
185
186    ############################
187    def _run_reader(self, index):
188        """
189        Cycle through reading records from a readers[i] and putting them in queue.
190        """
191        while True:
192            logging.debug('    Reader #%d waiting until record needed.', index)
193            self.queue_needs_record.wait(READER_TIMEOUT_WAIT)
194
195            # If we timed out waiting for someone to need a record, go
196            # home. We'll get started up again if needed.
197            if not self.queue_needs_record.is_set():
198                logging.debug('    Reader #%d timed out - exiting.', index)
199                return
200
201            # Else someone needs a record - leap into action
202            logging.debug('    Reader #%d waking up - record needed!', index)
203
204            # Guard against re-entry
205            with self.reader_locks[index]:
206                record = self.readers[index].read()
207
208                # If reader returns None, it's done and has no more data for
209                # us. Note that it's given us an EOF and exit.
210                if record is None:
211                    logging.info('    Reader #%d returned None, is done', index)
212                    self.reader_returned_eof[index] = True
213                    return
214
215            logging.debug('    Reader #%d has record, released reader_lock.', index)
216
217            # Add record to queue and note that an append event has
218            # happened.
219            with self.queue_lock:
220                # No one else can mess with queue while we add record. Once we've
221                # added it, set flag to say there's something in the queue.
222                logging.debug('    Reader #%d has queue lock - adding and notifying.',
223                              index)
224                self.queue.append(record)
225                self.queue_has_record.set()
226                self.queue_needs_record.clear()
227
228            # Now clear of queue_lock
229            logging.debug('    Reader #%d released queue_lock - looping', index)
230
231    ############################
232    def _apply_transforms(self, record):
233        """
234        Apply the transforms in series.
235        """
236        if record:
237            for t in self.transforms:
238                record = t.transform(record)
239                if not record:
240                    break
241        return record
READER_TIMEOUT_WAIT = 0.25
class ComposedReader(logger.readers.reader.Reader): 
 16class ComposedReader(Reader):
 17    """
 18    Read lines from one or more Readers (in parallel) and process their
 19    responses through zero or more Transforms (in series).
 20
 21    NOTE: we make the rash assumption that transforms are thread-safe,
 22    that is, that no mischief or corrupted internal state will result if
 23    more than one thread calls a transform at the same time. To be
 24    thread-safe, a transform must protect any changes to its internal
 25    state with a non-re-entrant thread lock, as described in the threading
 26    module.
 27
 28    Also NOTE: Most of the messy logic in this class comes from the desire
 29    to only call read() on our component readers when we actually need new
 30    records (NOTE: this desire may be misplaced!).
 31
 32    So when we get a request, we fire up threads and ask each of our
 33    readers for a record. We return the first one we get, and let the
 34    others pile up in a queue that we'll feed from the next time we're
 35    asked.
 36
 37    But we don't want to fire up a new thread for each reader every time
 38    the queue is empty, so we have threads (in run_reader()) hang out for
 39    a little while, waiting for another queue_needs_record event. If they
 40    get one, the call their own read() methods again. If they haven't been
 41    called on in READER_TIMEOUT_WAIT seconds, they exit, but will get
 42    fired up again by read() if/when the queue is empty and we're is asked
 43    for another record.
 44
 45    It's important to have the run_reader threads time out, or any process
 46    using a ComposedReader will never naturally terminate.
 47    """
 48    ############################
 49
 50    def __init__(self, readers, transforms=[], **kwargs):
 51        """
 52        Instantiation:
 53        ```
 54        reader = ComposedReader(readers, transforms=[], check_format=True)
 55
 56        readers        A single Reader or a list of Readers.
 57
 58        transforms     A single Transform or list of zero or more Transforms.
 59        ```
 60        Use:
 61        ```
 62        record = reader.read()
 63        ```
 64        Sample:
 65        ```
 66        reader = ComposedReader(readers=[UDPReader(port=6221),
 67                                         UDPReader(port=6223)],
 68                                transforms=[TimestampTransform()])
 69        ```
 70        """
 71        super().__init__(**kwargs)
 72
 73        # Make readers a list, even if it's only a single reader.
 74        self.readers = readers if type(readers) is list else [readers]
 75        self.num_readers = len(self.readers)
 76
 77        # Transforms can be empty. But if not empty, make it a list, even
 78        # if it's only a single transform.
 79        if not isinstance(transforms, list):
 80            self.transforms = [transforms]
 81        else:
 82            self.transforms = transforms
 83
 84        # List where we're going to store reader threads
 85        self.reader_threads = [None] * self.num_readers
 86
 87        # Whether reader[i] has returned EOF since we've last asked it
 88        self.reader_returned_eof = [False] * self.num_readers
 89
 90        # One lock per reader, to save us from accidental re-entry
 91        self.reader_locks = [threading.Lock() for i in range(self.num_readers)]
 92
 93        # Queue where we'll store extra records, and lock so only one
 94        # thread can touch queue at a time
 95        self.queue = []
 96        self.queue_lock = threading.Lock()
 97
 98        # The two events, queue_has_record and queue_needs_record interact
 99        # in a sort of a dance:
100        #
101        #  has = False, need = False: Everything is quiescent
102        #  has = False, need = True:  A request has been made, call readers
103        #  has = True,  need = True:  Momentary condition when we get needed rec
104        #  has = True,  need = False: We've got spare records in the queue
105        #
106        # Set when the queue is empty and we need a record
107        self.queue_needs_record = threading.Event()
108
109        # Set when a reader adds something to the queue
110        self.queue_has_record = threading.Event()
111
112    ############################
113    def read(self):
114        """
115        Get the next record from queue or readers.
116        """
117        # If we only have one reader, there's no point making things
118        # complicated. Just read, transform, return.
119        if len(self.readers) == 1:
120            return self._apply_transforms(self.readers[0].read())
121
122        # Do we have anything in the queue? Note: safe to check outside of
123        # lock, because we're the only method that actually *removes*
124        # anything. So if tests True here, we're assured that there's
125        # something there, and we lock before retrieving it. Advantage of
126        # doing it this way is that we don't tie up queue lock while
127        # processing transforms.
128        if self.queue:
129            logging.debug('read() - read requested; queue len is %d',
130                          len(self.queue))
131            with self.queue_lock:
132                record = self.queue.pop(0)
133                return self._apply_transforms(record)
134
135        # If here, nothing's in the queue. Note that, if we wanted to be
136        # careful to never unnecessarily ask for more records, we should
137        # put a lock around this, but the failure mode is somewhat benign:
138        # we ask for more records when some are already on the way.
139        logging.debug('read() - read requested and nothing in the queue.')
140
141        # Some threads may have timed out while waiting to be called to
142        # action; restart them.
143        for i in range(len(self.readers)):
144            if not self.reader_threads[i] \
145               or not self.reader_threads[i].is_alive() \
146               and not self.reader_returned_eof[i]:
147                logging.info('read() - starting thread for Reader #%d', i)
148                self.reader_returned_eof[i] = False
149                thread = threading.Thread(target=self._run_reader, args=(i,),
150                                          daemon=True)
151                self.reader_threads[i] = thread
152                thread.start()
153
154        # Now notify all threads that we do in fact need a record.
155        self.queue_needs_record.set()
156
157        # Keep checking/sleeping until we've either got a record in the
158        # queue or all readers have given us an EOF.
159        while False in self.reader_returned_eof:
160            logging.debug('read() - waiting for queue lock')
161            with self.queue_lock:
162                logging.debug('read() - acquired queue lock, queue length is %d',
163                              len(self.queue))
164                if self.queue:
165                    record = self.queue.pop(0)
166                    if not self.queue:
167                        self.queue_has_record.clear()  # only set/clear inside queue_lock
168
169                    logging.debug('read() - got record')
170                    return self._apply_transforms(record)
171                else:
172                    self.queue_has_record.clear()
173
174            # If here, nothing in queue yet. Wait
175            logging.debug('read() - clear of queue lock, waiting for record')
176            self.queue_has_record.wait(READER_TIMEOUT_WAIT)
177
178            if not self.queue_has_record.is_set():
179                logging.debug('read() - timed out waiting for record. Looping')
180            logging.debug('read() - readers returned EOF: %s',
181                          self.reader_returned_eof)
182
183        # All readers have given us an EOF
184        logging.debug('read() - all threads returned None; returning None')
185        return None
186
187    ############################
188    def _run_reader(self, index):
189        """
190        Cycle through reading records from a readers[i] and putting them in queue.
191        """
192        while True:
193            logging.debug('    Reader #%d waiting until record needed.', index)
194            self.queue_needs_record.wait(READER_TIMEOUT_WAIT)
195
196            # If we timed out waiting for someone to need a record, go
197            # home. We'll get started up again if needed.
198            if not self.queue_needs_record.is_set():
199                logging.debug('    Reader #%d timed out - exiting.', index)
200                return
201
202            # Else someone needs a record - leap into action
203            logging.debug('    Reader #%d waking up - record needed!', index)
204
205            # Guard against re-entry
206            with self.reader_locks[index]:
207                record = self.readers[index].read()
208
209                # If reader returns None, it's done and has no more data for
210                # us. Note that it's given us an EOF and exit.
211                if record is None:
212                    logging.info('    Reader #%d returned None, is done', index)
213                    self.reader_returned_eof[index] = True
214                    return
215
216            logging.debug('    Reader #%d has record, released reader_lock.', index)
217
218            # Add record to queue and note that an append event has
219            # happened.
220            with self.queue_lock:
221                # No one else can mess with queue while we add record. Once we've
222                # added it, set flag to say there's something in the queue.
223                logging.debug('    Reader #%d has queue lock - adding and notifying.',
224                              index)
225                self.queue.append(record)
226                self.queue_has_record.set()
227                self.queue_needs_record.clear()
228
229            # Now clear of queue_lock
230            logging.debug('    Reader #%d released queue_lock - looping', index)
231
232    ############################
233    def _apply_transforms(self, record):
234        """
235        Apply the transforms in series.
236        """
237        if record:
238            for t in self.transforms:
239                record = t.transform(record)
240                if not record:
241                    break
242        return record

Read lines from one or more Readers (in parallel) and process their responses through zero or more Transforms (in series).

NOTE: we make the rash assumption that transforms are thread-safe, that is, that no mischief or corrupted internal state will result if more than one thread calls a transform at the same time. To be thread-safe, a transform must protect any changes to its internal state with a non-re-entrant thread lock, as described in the threading module.

Also NOTE: Most of the messy logic in this class comes from the desire to only call read() on our component readers when we actually need new records (NOTE: this desire may be misplaced!).

So when we get a request, we fire up threads and ask each of our readers for a record. We return the first one we get, and let the others pile up in a queue that we'll feed from the next time we're asked.

But we don't want to fire up a new thread for each reader every time the queue is empty, so we have threads (in run_reader()) hang out for a little while, waiting for another queue_needs_record event. If they get one, the call their own read() methods again. If they haven't been called on in READER_TIMEOUT_WAIT seconds, they exit, but will get fired up again by read() if/when the queue is empty and we're is asked for another record.

It's important to have the run_reader threads time out, or any process using a ComposedReader will never naturally terminate.

ComposedReader(readers, transforms=[], **kwargs) 
 50    def __init__(self, readers, transforms=[], **kwargs):
 51        """
 52        Instantiation:
 53        ```
 54        reader = ComposedReader(readers, transforms=[], check_format=True)
 55
 56        readers        A single Reader or a list of Readers.
 57
 58        transforms     A single Transform or list of zero or more Transforms.
 59        ```
 60        Use:
 61        ```
 62        record = reader.read()
 63        ```
 64        Sample:
 65        ```
 66        reader = ComposedReader(readers=[UDPReader(port=6221),
 67                                         UDPReader(port=6223)],
 68                                transforms=[TimestampTransform()])
 69        ```
 70        """
 71        super().__init__(**kwargs)
 72
 73        # Make readers a list, even if it's only a single reader.
 74        self.readers = readers if type(readers) is list else [readers]
 75        self.num_readers = len(self.readers)
 76
 77        # Transforms can be empty. But if not empty, make it a list, even
 78        # if it's only a single transform.
 79        if not isinstance(transforms, list):
 80            self.transforms = [transforms]
 81        else:
 82            self.transforms = transforms
 83
 84        # List where we're going to store reader threads
 85        self.reader_threads = [None] * self.num_readers
 86
 87        # Whether reader[i] has returned EOF since we've last asked it
 88        self.reader_returned_eof = [False] * self.num_readers
 89
 90        # One lock per reader, to save us from accidental re-entry
 91        self.reader_locks = [threading.Lock() for i in range(self.num_readers)]
 92
 93        # Queue where we'll store extra records, and lock so only one
 94        # thread can touch queue at a time
 95        self.queue = []
 96        self.queue_lock = threading.Lock()
 97
 98        # The two events, queue_has_record and queue_needs_record interact
 99        # in a sort of a dance:
100        #
101        #  has = False, need = False: Everything is quiescent
102        #  has = False, need = True:  A request has been made, call readers
103        #  has = True,  need = True:  Momentary condition when we get needed rec
104        #  has = True,  need = False: We've got spare records in the queue
105        #
106        # Set when the queue is empty and we need a record
107        self.queue_needs_record = threading.Event()
108
109        # Set when a reader adds something to the queue
110        self.queue_has_record = threading.Event()

Instantiation:

reader = ComposedReader(readers, transforms=[], check_format=True)

readers        A single Reader or a list of Readers.

transforms     A single Transform or list of zero or more Transforms.

Use:

record = reader.read()

Sample:

reader = ComposedReader(readers=[UDPReader(port=6221),
                                 UDPReader(port=6223)],
                        transforms=[TimestampTransform()])
readers
num_readers
reader_threads
reader_returned_eof
reader_locks
queue
queue_lock
queue_needs_record
queue_has_record
def read(self): 
113    def read(self):
114        """
115        Get the next record from queue or readers.
116        """
117        # If we only have one reader, there's no point making things
118        # complicated. Just read, transform, return.
119        if len(self.readers) == 1:
120            return self._apply_transforms(self.readers[0].read())
121
122        # Do we have anything in the queue? Note: safe to check outside of
123        # lock, because we're the only method that actually *removes*
124        # anything. So if tests True here, we're assured that there's
125        # something there, and we lock before retrieving it. Advantage of
126        # doing it this way is that we don't tie up queue lock while
127        # processing transforms.
128        if self.queue:
129            logging.debug('read() - read requested; queue len is %d',
130                          len(self.queue))
131            with self.queue_lock:
132                record = self.queue.pop(0)
133                return self._apply_transforms(record)
134
135        # If here, nothing's in the queue. Note that, if we wanted to be
136        # careful to never unnecessarily ask for more records, we should
137        # put a lock around this, but the failure mode is somewhat benign:
138        # we ask for more records when some are already on the way.
139        logging.debug('read() - read requested and nothing in the queue.')
140
141        # Some threads may have timed out while waiting to be called to
142        # action; restart them.
143        for i in range(len(self.readers)):
144            if not self.reader_threads[i] \
145               or not self.reader_threads[i].is_alive() \
146               and not self.reader_returned_eof[i]:
147                logging.info('read() - starting thread for Reader #%d', i)
148                self.reader_returned_eof[i] = False
149                thread = threading.Thread(target=self._run_reader, args=(i,),
150                                          daemon=True)
151                self.reader_threads[i] = thread
152                thread.start()
153
154        # Now notify all threads that we do in fact need a record.
155        self.queue_needs_record.set()
156
157        # Keep checking/sleeping until we've either got a record in the
158        # queue or all readers have given us an EOF.
159        while False in self.reader_returned_eof:
160            logging.debug('read() - waiting for queue lock')
161            with self.queue_lock:
162                logging.debug('read() - acquired queue lock, queue length is %d',
163                              len(self.queue))
164                if self.queue:
165                    record = self.queue.pop(0)
166                    if not self.queue:
167                        self.queue_has_record.clear()  # only set/clear inside queue_lock
168
169                    logging.debug('read() - got record')
170                    return self._apply_transforms(record)
171                else:
172                    self.queue_has_record.clear()
173
174            # If here, nothing in queue yet. Wait
175            logging.debug('read() - clear of queue lock, waiting for record')
176            self.queue_has_record.wait(READER_TIMEOUT_WAIT)
177
178            if not self.queue_has_record.is_set():
179                logging.debug('read() - timed out waiting for record. Looping')
180            logging.debug('read() - readers returned EOF: %s',
181                          self.reader_returned_eof)
182
183        # All readers have given us an EOF
184        logging.debug('read() - all threads returned None; returning None')
185        return None

Get the next record from queue or readers.