UDP "readinto" nonblocking fails to raise error

[ctimmer]

It appears that under certain conditions "readinto" will return the same (last) record repeatedly.
Tested with RPI_PICO_W-20230426-v1.20.0.uf2

The basic server logic is (full code at the end):
poll input, when a UDP message is available
execute reads until there are no more messages, queuing results
process message(s) from the queue to update a display
repeat

Test results
"recvfrom" (test below) worked as expected.
"readinto" continuously returned the same (only) message and was only stopped when the queue was full. The next poll/read cycle did not return this message. Setting buffer_length to zero before the "readinto" and testing it after did not work around this problem, buffer_length was still set as if a new message was read.

recvfrom (1 input message) :
Code:

                print ("recvfrom")
                (buffer, addre        #self.udp_socket.setblocking (False)ss_ip) = udp_socket.recvfrom (1500)
                buffer = json.loads (buffer.decode())

recvfrom print output:

recvfrom
{'jsonrpc': '2.0', 'params': [{'area': 'dev.aqi', 'text': '13.0'}, {'area': 'dev.hum', 'text': ' 51.5'}, {'area': 'dev.temp_c', 'text': ' 17.3'}, {'area': 'dev.temp_f', 'text': ' 63.2'}, {'area_id': 'dev.time'}], 'method': 'update_area_list'}
recvfrom

readinto (1 input message) :
Code:

                print ("readinto")
                buffer_length = udp_socket.readinto(self.byte_buffer)
                buffer = json.loads (str (memoryview(self.byte_buffer[0:buffer_length]), 'utf8'))

readinto print output:

readinto
{'jsonrpc': '2.0', 'params': [{'area': 'dev.aqi', 'text': '13.5'}, {'area': 'dev.hum', 'text': ' 55.6'}, {'area': 'dev.temp_c', 'text': ' 16.2'}, {'area': 'dev.temp_f', 'text': ' 61.1'}, {'area_id': 'dev.time'}], 'method': 'update_area_list'}
readinto
{'jsonrpc': '2.0', 'params': [{'area': 'dev.aqi', 'text': '13.5'}, {'area': 'dev.hum', 'text': ' 55.6'}, {'area': 'dev.temp_c', 'text': ' 16.2'}, {'area': 'dev.temp_f', 'text': ' 61.1'}, {'area_id': 'dev.time'}], 'method': 'update_area_list'}
.
.
REPEATED 20 times (queue full)
.
.
readinto
{'jsonrpc': '2.0', 'params': [{'area': 'dev.aqi', 'text': '13.5'}, {'area': 'dev.hum', 'text': ' 55.6'}, {'area': 'dev.temp_c', 'text': ' 16.2'}, {'area': 'dev.temp_f', 'text': ' 61.1'}, {'area_id': 'dev.time'}], 'method': 'update_area_list'}

Full code:

import sys
import select
import socket
import json
import time

import display_config
from comm_modules.update_queue import UpdateQueue

PORT = 5010
QUEUE_LENGTH = 20
if "PORT" in display_config.PARAMETERS :
    PORT = display_config.PARAMETERS ["PORT"]
if "QUEUE_LENGTH" in display_config.PARAMETERS :
    QUEUE_LENGTH = display_config.PARAMETERS ["QUEUE_LENGTH"]

#-------------------------------------------------------------------------------
# UpdateUDPServer
#-------------------------------------------------------------------------------
class UpdateUDPServer :
    def __init__ (self ,
                  update_display ,
                  port = PORT ,
                  queue_length = QUEUE_LENGTH
                  ) :
        self.update_display = update_display
        self.update_queue = UpdateQueue (queue_length)
        self.running = True
        self.udp_socket = None
        self.port = port
        self.ip_stats = {}                                  # for debugging
        self.byte_buffer = bytearray (1500)     # readinto buffer

    def start_server (self, timeout = 5000) :
        self.udp_socket = socket.socket (socket.AF_INET, socket.SOCK_DGRAM)
        self.udp_socket.bind (('0.0.0.0', self.port))
        poller = select.poll()
        poller.register(self.udp_socket, select.POLLIN)  # only event
        while self.running :
            events = poller.poll (timeout)
            for event in events :
                #print (event)
                if event[1] == select.POLLIN :
                    self.udp_input (event[0])
                else :
                    poller.unregister (event[0])

    #-----------------------------------------------------------------------
    # udp_input
    #-----------------------------------------------------------------------
    def udp_input (self, udp_socket) :
        udp_socket.setblocking (False)
        while True :
            buffer = None
            try :
                #
                print ("recvfrom")
                (buffer, address_ip) = udp_socket.recvfrom (1500)
                buffer = json.loads (buffer.decode())
                '''
                print ("readinto")
                buffer_length = udp_socket.readinto(self.byte_buffer)
                buffer = json.loads (str (memoryview(self.byte_buffer[0:buffer_length]), 'utf8'))
                '''
                print (buffer)
                if "method" not in buffer :
                    print ("no method")
                    continue
                if buffer["method"] == "shutdown" :
                    self.running = False                     # All done
                    continue
                if "params" not in buffer :
                    print ("no params")
                    continue
                queue_entry = {
                    "method" : buffer["method"] ,
                    "params" : buffer["params"]
                    }
                self.update_queue.push_queue (queue_entry)
                if self.update_queue.full_queue () :         # is q full
                    break ;
            except :
                #print ("recvfrom: exception")
                #time.sleep_ms (100)
                break
        if not self.update_queue.empty_queue () :
            self.update_display.process_update_queue (self.update_queue)

## UpdateUDPServer ##

[moefear85]

yeah I think this has the same underlying root issue as #736

[ctimmer]

Verified the same issue occurs on:
MicroPython v1.21.0 on 2023-10-06; Raspberry Pi Pico W with RP2040

Curt

[shariltumin]

The socket.readinto() will read data from the socket until EOF, i.e. until the socket is closed or until the buffer is full. This is what we can read from the online manual

socket.read([size])

Read up to size bytes from the socket. Return a bytes object. If size is not given, it reads all data available from the socket until EOF; as such the method will not return until the socket is closed. This function tries to read as much data as requested (no “short reads”). This may be not possible with non-blocking socket though, and then less data will be returned.

socket.readinto(buf[, nbytes])

Read bytes into the buf. If nbytes is specified then read at most that many bytes. Otherwise, read at most len(buf) bytes. Just as read(), this method follows “no short reads” policy.

Return value: number of bytes read and stored into buf.

[ctimmer]

Thanks for the input. I'm not seeing the behavior outlined in the manual, the same (only) msssage keeps getting passed back to my application. If an error was raised or bytes read == zero, I could handle that properly.

Curt