openwebrx-clone/csdr/pipe.py

157 lines
5.0 KiB
Python
Raw Permalink Normal View History

2020-08-14 17:54:07 +00:00
import os
import select
import time
import threading
import logging
logger = logging.getLogger(__name__)
class Pipe(object):
READ = "r"
WRITE = "w"
NONE = None
@staticmethod
def create(path, t, encoding=None):
if t == Pipe.READ:
return ReadingPipe(path, encoding=encoding)
elif t == Pipe.WRITE:
return WritingPipe(path, encoding=encoding)
elif t == Pipe.NONE:
return Pipe(path, None, encoding=encoding)
def __init__(self, path, direction, encoding=None):
self.doOpen = True
self.path = "{base}_{myid}".format(base=path, myid=id(self))
self.direction = direction
self.encoding = encoding
self.file = None
os.mkfifo(self.path)
def open(self):
2020-08-14 18:20:07 +00:00
"""
this method opens the file descriptor with an added O_NONBLOCK flag. This gives us a special behaviour for
FIFOS, when they are not opened by the opposing side:
- opening a pipe for writing will throw an OSError with errno = 6 (ENXIO). This is handled specially in the
WritingPipe class.
- opening a pipe for reading will pass through this method instantly, even if the opposing end has not been
opened yet, but the resulting file descriptor will behave as if O_NONBLOCK is set (even if we remove it
immediately here), resulting in empty reads until data is available. This is handled specially in the
ReadingPipe class.
"""
2021-01-20 16:01:46 +00:00
2020-08-14 17:54:07 +00:00
def opener(path, flags):
fd = os.open(path, flags | os.O_NONBLOCK)
os.set_blocking(fd, True)
return fd
2020-08-14 18:20:07 +00:00
self.file = open(self.path, self.direction, encoding=self.encoding, opener=opener)
2020-08-14 17:54:07 +00:00
def close(self):
self.doOpen = False
try:
if self.file is not None:
self.file.close()
os.unlink(self.path)
except FileNotFoundError:
# it seems like we keep calling this twice. no idea why, but we don't need the resulting error.
pass
except Exception:
logger.exception("Pipe.close()")
def __str__(self):
return self.path
class WritingPipe(Pipe):
def __init__(self, path, encoding=None):
self.queue = []
self.queueLock = threading.Lock()
super().__init__(path, "w", encoding=encoding)
self.open()
def open_and_dequeue(self):
2020-08-14 18:20:07 +00:00
"""
This method implements a retry loop that can be interrupted in case the Pipe gets shutdown before actually
being connected.
2020-08-14 17:54:07 +00:00
2020-08-14 18:20:07 +00:00
After the pipe is opened successfully, all data that has been queued is sent in the order it was passed into
write().
"""
retries = 0
while self.file is None and self.doOpen and retries < 10:
try:
super().open()
except OSError as error:
# ENXIO = FIFO has not been opened for reading
if error.errno == 6:
2021-01-20 16:01:46 +00:00
time.sleep(0.1)
2020-08-14 18:20:07 +00:00
retries += 1
else:
raise
# if doOpen is false, opening has been canceled, so no warning in that case.
2020-08-14 17:54:07 +00:00
if self.file is None:
2020-08-14 18:20:07 +00:00
if self.doOpen:
logger.warning("could not open FIFO %s", self.path)
2020-08-14 17:54:07 +00:00
return
with self.queueLock:
for i in self.queue:
self.file.write(i)
self.file.flush()
self.queue = None
def open(self):
2020-08-14 18:20:07 +00:00
"""
This sends the opening operation off to a background thread. If we were to block the thread here, another pipe
may be waiting in the queue to be opened on the opposing side, resulting in a deadlock
"""
2020-08-14 18:22:25 +00:00
threading.Thread(target=self.open_and_dequeue, name="csdr_pipe_thread").start()
2020-08-14 17:54:07 +00:00
def write(self, data):
2020-08-14 18:20:07 +00:00
"""
This method queues all data to be written until the file is actually opened. As soon as a file is available,
it becomes a passthrough.
"""
2020-08-14 17:54:07 +00:00
if self.file is None:
with self.queueLock:
self.queue.append(data)
return
r = self.file.write(data)
self.file.flush()
return r
class ReadingPipe(Pipe):
def __init__(self, path, encoding=None):
super().__init__(path, "r", encoding=encoding)
def open(self):
2020-08-14 18:20:07 +00:00
"""
This method implements an interruptible loop that waits for the file descriptor to be opened and the first
batch of data coming in using repeated select() calls.
:return:
"""
2020-08-14 17:54:07 +00:00
if not self.doOpen:
return
super().open()
2020-08-14 18:20:07 +00:00
while self.doOpen:
(read, _, _) = select.select([self.file], [], [], 1)
if self.file in read:
break
2020-08-14 17:54:07 +00:00
def read(self):
if self.file is None:
self.open()
return self.file.read()
def readline(self):
if self.file is None:
self.open()
return self.file.readline()