Python 线程/队列快速数据 I/O 丢失或重复 Get() 数据包

Question

编写了一个简单的程序，它记录来自带有以太网屏蔽的 Arduino Mega 的数据，每个 48 字节数据包的传输速率通常为 0.0008-0.001 秒。我创建了一个线程，它打开一个套接字并从 arduino 获取一个数据包并将其放入队列中连续循环，还在控制台中打印一条调试消息，根据控制台我还没有看到它跳过一个节拍从套接字接收数据包。还有另一个线程从队列中获取等待数据包并将其打印到控制台。然而，第二个线程有时会跳过或复制数据包，因为它将它们从队列中拉出，因此我会丢失有价值的数据。我不确定我在这里做错了什么，因为我尝试了几次不同的调用 put 和 get 迭代。在我为多进程重写之前，也许有人可以阐明这个问题？

import sys, os, time, datetime  
import socket from threading 
import Thread from PyQt4 
import QtCore, QtGui 
import queue  
import binascii

q = queue.Queue(maxsize=0) 
Running = False

class Window(QtGui.QWidget):
    def __init__(self):
        QtGui.QWidget.__init__(self)
        self.button = QtGui.QPushButton('Start', self)
        self.button.clicked.connect(self.handleButton)
        layout = QtGui.QVBoxLayout(self)
        layout.addWidget(self.button)

    def handleButton(self):
        global Running, thread, thread2, thrAdruino, thrPackets
        if Running == True:
            Running = False
            self.button.setText('Start')
            thrAdruino.isRunning = False
            thrPackets.isRunning = False
            print ('stop')
        else:
            Running = True
            self.button.setText('Stop')

            thrAdruino = talkArduino()
            thread = Thread(target=thrAdruino.run, daemon=True )
            thread.start()

            thrPackets = parsePackets()
            thread2 = Thread(target=thrPackets.run, daemon=True )
            thread2.start()

            print ("Start")   



class talkArduino(Thread):
    def __init__(self):
        self.isRunning = True
    def run(self):
        print('Arduino: Thread Started')
        TCP_RECEIVE_BUFFER = bytearray(48)
        TCP_IP, TCP_PORT, BUFFER_SIZE = '192.168.1.177', 502, 48
        s=socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        print('Arduino: Socket Opening')
        s.connect((TCP_IP, TCP_PORT))
        print('Arduino: Socket Opened, sending start command')
        s.send(bytes(1))
        while self.isRunning == True:
            s.recv_into(TCP_RECEIVE_BUFFER)
            if TCP_RECEIVE_BUFFER[0] == 48:
                q.put(TCP_RECEIVE_BUFFER, block=True, timeout=1)
                print ('Arduino: ',  binascii.hexlify(TCP_RECEIVE_BUFFER) )
        s.close()
        print('Arduino: Closed Socket')



class parsePackets(Thread):
    def __init__(self):
        self.isRunning = True
    def run(self):
        print ('packet parser thread started')
        print (q.qsize())
        buffer1 = bytearray()
        while self.isRunning == True:
            try:
                print (q.qsize())
                buffer1 = q.get(block=True, timeout=1)
                print ("Packets: ", binascii.hexlify(buffer1) )
            except q.Empty:
                pass



if __name__ == '__main__':

    import sys
    app = QtGui.QApplication(sys.argv)
    window = Window()
    window.show()
    sys.exit(app.exec_())

为了完整起见，我还包括了 Arduino Mega Sketch。

#include <SPI.h>
#include <Ethernet.h>

// Enter a MAC address and IP address for your controller below.
// The IP address will be dependent on your local network:
byte mac[] = {0xDE, 0xAD, 0xBE, 0xEF, 0xFE, 0xED };
IPAddress ip(192,168,1,177);
byte Buffer[] = { 0x30, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
                  0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
                  0x20, 0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28, 0x29, 0x2a, 0x1b, 0x2c, 0x2d, 0x2e, 0x2f};
// Initialize the Ethernet server library
// with the IP address and port you want to use 
// (port 80 is default for HTTP):
EthernetServer server(502);

// Define various ADC prescaler
const unsigned char PS_16 = (1 << ADPS2);
const unsigned char PS_32 = (1 << ADPS2) | (1 << ADPS0);
const unsigned char PS_64 = (1 << ADPS2) | (1 << ADPS1);
const unsigned char PS_128 = (1 << ADPS2) | (1 << ADPS1) | (1 << ADPS0);


void setup() {
 // Open serial communications and wait for port to open:
  Serial.begin(9600);
   while (!Serial) {
    ; // wait for serial port to connect. Needed for Leonardo only
  }

  // set up the ADC
  ADCSRA &= ~PS_128;  // remove bits set by Arduino library
  // you can choose a prescaler from above.
  // PS_16, PS_32, PS_64 or PS_128
  ADCSRA |= PS_32;    // set our own prescaler to 64 

  // start the Ethernet connection and the server:
  SPI.setClockDivider(SPI_CLOCK_DIV2);
  Ethernet.begin(mac, ip);
  server.begin();
  Serial.print("server is at ");
  Serial.println(Ethernet.localIP());
}


void loop() {
  unsigned int i;
  unsigned int AI;
  // listen for incoming clients
  EthernetClient client = server.available();
  if (client) {
    Serial.println("new client");

    // an http request ends with a blank line
    while (client.connected()) {
          for(i=0;i<16;i++) {
            AI =analogRead(i);
            Buffer[i*2+4] = highByte(AI);
            Buffer[i*2+5] = lowByte(AI);
          }
          Buffer[1] +=1;
          client.write(Buffer, 48);
    }
    // give the web browser time to receive the data
    delay(1);
    // close the connection:
    client.stop();
    Serial.println("client disonnected");
  }
}

Answer 1

当您put() 和get() 时，您是在放置和获取对对象的引用——而不是对象的副本。这一行引用了原始的 bytearray()

q.put(TCP_RECEIVE_BUFFER, block=True, timeout=1)

当你得到：

buffer1 = q.get(block=True, timeout=1)    

buffer1 现在是对与 TCP_RECEIVE_BUFFER 相同的字节数组的引用。如果第一个线程改变了缓冲区的内容，那么第二个线程看到的是相同的数据。

会发生什么取决于两个线程如何交错。您描述的症状可以通过编写两个线程的可能顺序来查看。

正常情况下，get() 打印正确：

# case 1 - what you want to happen
s.recv_into(TCP_RECEIVE_BUFFER)  # receive = "packet1"
q.put(TCP_RECEIVE_BUFFER, block=True, timeout=1) # push reference to buffer - 1 item on Q
buffer1 = q.get(block=True, timeout=1) # pop reference to buffer - 0 item on Q
# buffer1 is correct at this point = "packet1"

get() 如何丢包：

# case 2 - get is delayed a bit - 2 recv's before first get()
s.recv_into(TCP_RECEIVE_BUFFER) # write into buffer "packet1"
q.put(TCP_RECEIVE_BUFFER, block=True, timeout=1) # push reference to buffer - 1 item on Q
s.recv_into(TCP_RECEIVE_BUFFER) # write into buffer "packet2" - overwrites buffer
buffer1 = q.get(block=True, timeout=1) # pop reference to buffer - 0 item onQ   
# buffer1 has missed one - the buffer now contains "packet2" - never will see "packet1"

get() 如何两次看到同一个数据包：

# case 3 - get is delayed a bit longer - 2 puts() before first get()
s.recv_into(TCP_RECEIVE_BUFFER) # write into buffer "packet1"
q.put(TCP_RECEIVE_BUFFER, block=True, timeout=1) # push reference to buffer - 1 item on Q
s.recv_into(TCP_RECEIVE_BUFFER) # write into buffer "packet2" - overwrites buffer
q.put(TCP_RECEIVE_BUFFER, block=True, timeout=1) # push reference to buffer - 2 item on Q
buffer1 = q.get(block=True, timeout=1) # pop reference to buffer - 1 item on Q    
# buffer1 has missed one - the buffer now contains "packet2"
buffer1 = q.get(block=True, timeout=1) # pop reference to buffer - 0 item on Q    
# buffer1 has missed one - the buffer still contains "packet2"

要解决此问题，第一个线程必须将数据的唯一副本推送到 Q 上并销毁其引用。您将数据提供给 Q，然后 Q 拥有它，直到接收者接受它。每次只需创建一个新缓冲区即可解决：

    while self.isRunning == True:
        s.recv_into(TCP_RECEIVE_BUFFER)
        if TCP_RECEIVE_BUFFER[0] == 48:
            print ('Arduino: ',  binascii.hexlify(TCP_RECEIVE_BUFFER) )
            # you have to print before pushing onto Q
            # once it is on the Q, it is no longer safe for this thread to use
            q.put(TCP_RECEIVE_BUFFER, block=True, timeout=1)
            # do not use the object any more - you have given it away
            # create a new object for the next receive
            TCP_RECEIVE_BUFFER = bytearray(48)