title: Receive UDP packets with python, packet loss
body: <p>I'm having a lot of packet loss using UDP in python. I know I should use TCP if I don't want packet loss, but I don't have (full) controll over the sender.</p>
  
  <p>It's a camera that sends 15 images per second using UDP multicast.</p>
  
  <p>Below you see the code I've written now.
  It uses multiprocessing to allow the producer and consumer function to work in parallel. Producer function catches the packets, consumer function processes them and writes the images to .bmp files.</p>
  
  <p>I've written a class PacketStream which writes the bytes from the packages to a .bmp file.</p>
  
  <p>When the camera sends a new image, it first sends one packet, with first byte = 0x01. This contains information about the image.
  Then 612 packets are sent with first byte = 0x02. These contain the bytes from the image (508 bytes/packet).</p>
  
  <p>Since 15 images are sent per second, ~9000 packets are sent per second. Altough this happens at a faster rate in bursts per image, at ~22 packets/ms.</p>
  
  <p>I can receive all packets perfectly using tcpdump or wireshark.
  But using the code below, packets are missed.
  Surely my windows 7 pc should be able to handle this? I'm also using it on a raspberry pi 3, and there more or less the same number of packets is missed. Therefore I think it's a problem with the code.</p>
  
  <p>I've tried lots of different things, like threading instead of multiprocessing, Pipe instead of Queue.</p>
  
  <p>I also tried increasing the socket buffer with </p>
  
  <pre><code>sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 3000000)
  </code></pre>
  
  <p>to no avail.</p>
  
  <p>Is this at all possible in python?</p>
  
  <p>Thanks in advance,</p>
  
  <pre><code>import time
  from multiprocessing import Process, Queue
  import socket
  import struct
  from PIL import Image
  
  
  class PacketStream:
      def __init__(self, output_path):
          self.output_path = output_path
          self.data_buffer = ''
          self.img_id = -1  # -1 = waiting for start of new image
  
      def process(self, data):
          message_id = data[0]
          if message_id == '\x01':
              self.wrap_up_last_image()
              self.img_id = ord(data[3])
              self.data_buffer = ''
          if message_id == '\x02':
              self.data_buffer += data[6:]
  
      def wrap_up_last_image(self):
          if self.img_id &gt; 0:
              n_bytes = len(self.data_buffer)
              if n_bytes == 307200:
                  global i
                  write_image(self.output_path + str(i).zfill(7) + '_' + str(self.img_id).zfill(3) + '.bmp',
                              self.data_buffer)
                  i += 1
              else:
                  print 'Image lost: %s bytes missing.' % (307200 - n_bytes)
  
  
  def write_image(path, data):
      im = Image.frombuffer('L', (640, 480), bytearray(data), 'raw', 'L', 0, 1)
      im.save(path)
      print time.time(), path
  
  
  def producer(q):
      # setup socket
      MCAST_GRP = '239.255.83.71'
      MCAST_PORT = 2271
      sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
      sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
      sock.bind(('', MCAST_PORT))
      mreq = struct.pack('4sl', socket.inet_aton(MCAST_GRP), socket.INADDR_ANY)
      sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
      while True:
          q.put(sock.recv(512))
  
  
  def consumer(q):
      packet_stream = PacketStream('D:/bmpdump/')
      while True:
          data = q.get()
          packet_stream.process(data)
  
  i = 0
  if __name__ == '__main__':
      q = Queue()
  
      t1 = Process(target=producer, args=(q,))
      t1.daemon = True  # so they stop when the main prog stops
      t1.start()
      t2 = Process(target=consumer, args=(q,))
      t2.daemon = True
      t2.start()
  
      time.sleep(10.0)
  
      print 'Program finished.'
  </code></pre>
  
  <p>EDIT</p>
  
  <p>Thanks for all the suggestions.</p>
  
  <p>1) I tried threading+queue already, also the ''.join(), didn't seem to make much difference. I'm quite sure now the problem is that the producer thread doesn't get enough priority. I can't find how to increase this using Python? Is this even possible?</p>
  
  <p>2) I managed to lose only about 10% using the code below. Processor is at ~25% (on the raspberry pi) The key is to consume the data when there's a pause in the packet stream, i.e. when the last data package has arrived</p>
  
  <pre><code>import time
  import socket
  import struct
  from PIL import Image
  
  
  def write_image(path, data):
      im = Image.frombuffer('L', (640, 480), bytearray(data), 'raw', 'L', 0, 1)
      im.save(path)
      print time.time(), path
  
  def consume(data_buffer):
      img_id = ord(data_buffer[0][1])
      real_data_buffer = [data[6:] for data in data_buffer]
      data_string = ''.join(real_data_buffer)
  
      global i
      write_image('/media/pi/exthdd_02/bmpdump/' + str(i).zfill(7) + '_' + str(img_id).zfill(3) + '.bmp', data_string)
      i += 1
  
  def producer(sock):
      print 'Producer start'
      data_buffer = []
      while True:
          data = sock.recvfrom(512)[0]
          if data[0] == '\x01':
              data_buffer = []
          else:
              data_buffer.append(data)
          if len(data_buffer) == 612:
              consume(data_buffer)
  
  
  # image counter
  i = 0
  
  # setup socket
  MCAST_GRP = '239.255.83.71'
  MCAST_PORT = 2271
  sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
  sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  sock.bind((MCAST_GRP, MCAST_PORT))
  mreq = struct.pack("4sl", socket.inet_aton(MCAST_GRP), socket.INADDR_ANY)
  sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
  sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 30000000)
  
  producer(sock)
  </code></pre>