Avoid accumulation of data in udp socket or read newest data from udp socket

Question

I am trying to send data continuously from a c++ code to a python code. I used udp sockets to send the data. The rate of sending is at a faster rate than the receiving rate as it is a simple sensor code. So the data sent is accumulated in the socket. When I try to read the data it returns an old data. How can I read the newest data from the socket or delete the old data when the new data is sent?

Answer 1

How can I read the newest data from the socket or delete the old data when the new data is sent?

Read a packet of data from the socket and place it into a buffer. Keep reading packets from the socket, placing each packet into the buffer each time (replacing whatever packet-data was in the buffer previously), until there is no more data left to read -- non-blocking-I/O mode is useful for this, as a non-blocking recv() will throw a socket.error exception with code EWOULDBLOCK when you've run out of data in the socket's incoming-data-buffer. Once you've read all the data, whatever is left in your buffer is the newest data, so go ahead and use that data.

Sketch/example code follows (untested, may contain errors):

  sock = socket.socket(family, socket.SOCK_DGRAM)

  [... bind socket, etc... ]

  # Receive-UDP-data event-loop begins here
  sock.setblocking(False)
  while True:
     newestData = None

     keepReceiving = True
     while keepReceiving:
        try:
           data, fromAddr = sock.recvfrom(2048)
           if data:
              newestData = data
        except socket.error as why:
           if why.args[0] == EWOULDBLOCK:
              keepReceiving = False
           else:
              raise why

     if (newestData):
        # code to handle/parse (newestData) here

Answer 2

The data from the publisher goes to a buffer and the slower subscriber reads the buffer piece by piece in a first-in-first-out manner. To make it act as last-in-first-out (maybe this description is not precise because in your case we only care about the last data), you can revise a UDP protocol with asyncio create_datagram_endpoint() so that the subscriber clears the current buffer queue and receive the latest data from the new queue.

The following is an example, I tested it on macOS 11.4 with Python 3.9.6.

UdpProtocol.py is the customized UDP protocol object.

import asyncio
class UdpProtocol:
    def __init__(self):
        self.packets = asyncio.Queue()

    def connection_made(self, transport):
        print("connection made")

    def datagram_received(self, data, addr):
        # clear the current queue and the accumulated data
        self.packets._queue.clear()
        # put latest data to the queue
        self.packets.put_nowait((data, addr))

    def connection_lost(self, transport):
        print("connection lost")

    def error_received(self, exc):
        pass

    async def recvfrom(self):
        # get data from the queue
        return await self.packets.get()

This is the publisher.

import asyncio
from UdpProtocol import UdpProtocol

async def main():
    server_address = ("127.0.0.1", 8000)
    loop = asyncio.get_running_loop()
    transport, protocol = await loop.create_datagram_endpoint(
        UdpProtocol, local_addr=None, remote_addr=server_address)

    idx = 0
    while True:
        transport.sendto(str(idx).encode(), server_address)
        print(idx)
        idx += 1
        await asyncio.sleep(0.1)

if __name__ == "__main__":
    asyncio.run(main())

This is the subscriber.

import asyncio
from UdpProtocol import UdpProtocol

async def main():
    server_address = ("127.0.0.1", 8000)
    loop = asyncio.get_running_loop()
    transport, protocol = await loop.create_datagram_endpoint(
        UdpProtocol, local_addr=server_address, remote_addr=None)
    while True:
        data, addr = await protocol.recvfrom()
        print(data, addr)
        await asyncio.sleep(1)

if __name__ == "__main__":
    asyncio.run(main())

Answer 3

I used the idea from [here] (Buffer size for reading UDP packets in Python) and adjust the receive buffer in the python socket. There were some additional points to be noted that are discussed in the thread as well.

sock = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)

This worked well for my application. As for my application, I send data from LabVIEW to python and data should not accumulate. In my case losing packets doesn't affect my application, so I basically reduce the receive buffer and time out the socket to do so.

Answer 4

Once you receive the packet read the packet and send back an acknowledgement. Let the Sender wait until it receives the acknowledgement before it sends the next packet. By this way u can prevent the accumulation of packets at receiving side.