В настоящее время я работаю над стажировкой по предмету колледжа, в которой мне нужно использовать два сокета UDP для отправки и получения видеоданных с компьютера.
Чтобы добиться этого, я создал следующий класс с именем UDPConnection.Он в основном использует два сокета UDP, один для получения и один для отправки данных, а также два потока.Первый читает все данные, полученные из первого сокета UDP, и сохраняет их в буфере, чтобы я мог получить их в любой момент.Второй считывает данные из выходного буфера и записывает их в отправляющий сокет, поэтому мне просто нужно вставить данные в выходной сокет, чтобы избежать возможных блокировок.Оба потока используют События, чтобы узнать, нужно ли им останавливать / возобновлять.
Проблема, с которой я сталкиваюсь, заключается в том, что иногда между отправляемыми данными и задержкой (даже более 10 секунд, когда компьютеры находятся на расстоянии 20 см)получено.Когда это происходит, выходной буфер начинает заполняться, а пакеты почти не принимаются.Похоже, что поток-отправитель не может передать данные, хранящиеся в буфере вывода, в сеть с достаточной скоростью, поскольку он даже получает исключения тайм-аута, но, поскольку я использую сокет UDP, я не знаю, почему это может бытьпроисходит.Разве сокет не должен выдвигать данные или отбрасывать их, если он не может этого сделать?Может быть, в моем коде есть еще одна ошибка, которую я не заметил?
Я проверил все события, чтобы убедиться, что ни один из потоков не приостановлен или что-то еще, но они работают отлично.Вот мой код (функции, используемые потоками: _sendData и _rcvData):
class UDPConnection(object):
def __init__(self, orIP, dstIp, orPort, dstPort, logger, maxTimeout = 2, rcvDataLen = 65535):
super(UDPConnection, self).__init__()
self.buffer = []
self.outputBuffer = queue.Queue(maxsize=120)
self.orIP = orIP
self.dstIp = dstIp
self.orPort = orPort
self.dstPort = dstPort
self.maxTimeout = maxTimeout
self.logger = logger
self.rcvDataLen = rcvDataLen
self.pausedEvent = Event()
self.closedEvent = Event()
self.rcvThread = None
self.sendThread = None
# Initialize the connection and the send/receive threads
def start(self):
# Evento que indica que la conexion debe estar cerrada
self.closedEvent.clear()
# Inicializamos para que no se detengan los hilos
self.pausedEvent.set()
self.sendSocket = None
self.rcvSocket = None
# Receive socket - To receive information
try:
self.rcvSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.rcvSocket.bind((self.orIP, self.orPort))
self.rcvSocket.settimeout(self.maxTimeout)
except socket.error as e:
self.logger.exception('UDPConnection - Receive socket {}:{}.'.format(self.orIP, self.orPort))
return False
time.sleep(2)
self.logger.debug('UDPConnection - Escuchando en {}:{}'.format(self.orIP, self.orPort))
self.rcvThread = Thread(target=self._rcvData)
self.rcvThread.start()
self.sendThread = Thread(target=self._sendData)
self.sendThread.start()
# Send socket - To send information
try:
self.sendSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sendSocket.connect((self.dstIp, self.dstPort))
self.sendSocket.settimeout(self.maxTimeout)
except socket.error as e:
self.logger.exception('UDPConnection - Send socket {}:{}.'.format(self.dstIp, self.dstPort))
return False
self.logger.debug('UDPConnection - Enviando en {}:{}'.format(self.orIP, self.orPort))
return True
# Pauses the connection
def pause(self):
self.pausedEvent.clear()
# Resumes the connection
def resume(self):
self.pausedEvent.set()
# Closes the connection
def quit(self):
# In case it's paused, we prevent it from getting stucked
self.pausedEvent.set()
self.closedEvent.set()
time.sleep(self.maxTimeout)
self.sendSocket.close()
self.rcvSocket.close()
return True
def isClosed(self):
return self.closedEvent.is_set()
def isPaused(self):
return not self.pausedEvent.is_set()
def isRunning(self):
return self.pausedEvent.is_set() and not self.closedEvent.is_set()
# Auxiliar function that pushes an object into the reception buffer.
# Used for inherited classes
def _bufferPush(self, data):
self.buffer.append(data)
# Auxiliar function that gets an element from the reception buffer.
# Used for inherited classes
def _bufferPop(self):
r = self.buffer[0]
del self.buffer[0]
# Auxliar function that allows us to know the number of elements
# on the reception buffer.
def _bufferLen(self):
bufLen = len(self.buffer)
return bufLen
# Function that empties the output buffer
def _emptyOutputBuffer(self):
while True:
try:
self.outputBuffer.get(False)
except queue.Empty:
break
# Used to get data from the output socket and push it into the network.
# If the connection is paused, it removes all the data from the output buffer and waits till it gets resumed
def _sendData(self):
while not self.isClosed():
if self.isPaused():
self._emptyOutputBuffer()
self.pausedEvent.wait()
try:
data = self.outputBuffer.get(True, self.maxTimeout)
self.sendSocket.send(data)
except queue.Empty as e:
continue
except socket.error as e:
# Socket timeout, perdemos el frame.
print('Timeout while sending the frame')
self._emptyOutputBuffer()
continue
# Function used to get data from the network and push it into the received buffer.
def _rcvData(self):
while not self.isClosed():
# Comprobamos si la conexion esta o no parada
self.pausedEvent.wait()
try:
request = self.rcvSocket.recv(self.rcvDataLen)
self._bufferPush(request)
except socket.error as e:
continue
# Function used by the user to send data
def send(self, data):
if self.isRunning():
if not self.outputBuffer.full():
self.outputBuffer.put_nowait(data)
# Function used by the user to receive data
def rcv(self, timeout=5):
start = time.time()
bufLen = 0
while (time.time()-start) < timeout:
bufLen = self._bufferLen()
if bufLen != 0:
break
if bufLen == 0:
raise BlockingIOError('Timeout passed.')
return self._bufferPop()
Тогда
Может ли кто-нибудь помочь мне, пожалуйста?
Большое спасибо!