0

I'll try to distill this down to the essence of the question, please feel free to ask for details. Also, I'm not a software person to begin with, so my approaches might have some obvious bugs - which I would appreciate pointers to.

So, I have 2 serial ports. One is half-duplex, the other is full-duplex. My goal is to buffer the data from the full duplex port in a pair of queues and send it out over the half-duplex one, running on a Raspberry Pi (Zero ideally, could be 3B+). There is also an initialization routine to establish a connection on one of the ports that has to be serviced before any data gets sent, and some handshaking with control characters to switch drections after a set number of bytes are sent.

So far I wrote two working solutions. The first one uses multiprocessing to service each port in a separate process, and the second one uses asyncio to deal with both at once.I also had a 3rd naive solution which didn't use either, and used 100% CPU on one core with corresponding slow throughput.

The multiprocessing solution works well in that it doesn't use 100% of any core on a laptop OR a Raspberry Pi, but when running on the Pi it only achieves about 1/4 the throughput compared to running on a laptop. Question 1: Why might this be? If CPU speed was the bottleneck, I would expect the cores to run at 100% on the Pi but they don't according to htop.

The asyncio solution turned out to be much faster, it achieves twice the throughput than the multiprocessing one when both are run on a laptop - however it uses nearly 100% of one core as I would expect.

Either way, when I push data through the serial ports at maximum speed, I have not seen under ~12% aggregate CPU utilization (1/8 cores fully or ~6% for 2 cores). This seems extremely high for just buffering some data and servicing serial ports. I understand that nonblocking reads will incur this, which is why my very initial approach was suboptimal.

Question 2: For either of these solutions, how would I go about reducing the CPU usage further so it can run more comfortably on a Raspberry Pi? I think it's possible to use multiprocessing with asyncio (service each port in its own process using async) - but this is getting into the weeds and over my head.

Multiprocessing version (abridged - the control character checking in receivebytes() can probably be optimized - but I'm pretty sure this is not my main problem):

import serial
import time
import sys
import string
from multiprocessing import Queue, Process
from queue import Empty as QueueEmpty

bytecount = 5000

ser = serial.Serial('/dev/ttyUSB1', 115200, timeout=5)
ser1 = serial.Serial('./reader', 57600 , timeout = 0.00005)

tx=Queue()
rx=Queue()

index = 0
index1 = 0

conn_good = 0


def init():
  print("waiting for server...")
  while(ser.read(1) != b'\x12'):   
    continue
  print("got connect byte")
  ser.write(b'\x14')                      
  global conn_good
  conn_good=1

def serviceport():
  print("servicing port!")
  while(1):
    newbytes=ser1.read()
    if(len(newbytes) != 0):
      tx.put(newbytes)
    try:
      rxbyte = rx.get(block=False)
      ser1.write(rxbyte)
    except QueueEmpty:
      continue 

def sendbytes():
  ser.write(b'\x11')                          
  for index1 in range (bytecount, 0, -1):      
    try:                                   
      byte = tx.get(block=False)
      ser.write(byte)
    except QueueEmpty:
      break                       
  ser.write(b'\x13')                          

def receivebytes():
  global conn_good
  rxbyte=ser.read(1)                          
  if(len(rxbyte)==0):
    print("Timing out waiting for remote byte...")
    conn_good = 0
    return
  while(rxbyte != b'\x11'):          
    if(rxbyte == b'\x12'):                  
      print("It's a connect byte, quit and reconnect")  
      conn_good=0
      return            
    rxbyte=ser.read(1)                       
    if(len(rxbyte)==0):
      print("Timing out waiting for start byte...")
      conn_good = 0
      return
  rxbyte=ser.read(1)                                     
  if(len(rxbyte)==0):
    print("Timing out waiting for data or end byte...")
    conn_good = 0
    return                       
  while(rxbyte != b'\x13'): 
    if(rxbyte == b'\x12'):                    
      print("It's a connect byte, quit and reconnect")  
      conn_good=0
      return
    rx.put(rxbyte)
    rxbyte=ser.read(1)                       
    if(len(rxbyte)==0):
      print("Timing out waiting for remote byte...")
      conn_good = 0
      return
  
def serviceradio():
  while(1):
    init()
    while(conn_good == 1):
      receivebytes()
      sendbytes()

if __name__ == '__main__':
  radioprocess = Process(target=serviceradio)
  radioprocess.start()

  serialinputprocess = Process(target=serviceport)
  serialinputprocess.start()

Asyncio version:

import asyncio
import serial
import serial_asyncio

bytes_to_send = 5000

reading = 0
writing = 0

conn_good = 0

# Create asyncio event for signaling data reception
data_received_event = asyncio.Event()

async def read_serial(reader, rx_local):
    print("Reading local port")
    while True:
      msg = await reader.read(1)
      rx_local.put_nowait(msg)
      data_received_event.set()

async def write_serial(writer, tx_local):
    print("Writing local port..")
    while True:
        try:
           if(not tx_local.empty()):
             sendbytes = tx_local.get_nowait()
             writer.write(sendbytes)
           await asyncio.sleep(0.00002)
        except (asyncio.QueueEmpty,asyncio.TimeoutError)
           pass

async def sendbytes(ser, tx_radio_bytes, bytecount):
    global reading
    global conn_good
    print("Sending")
    while(True):  
      while(conn_good == 1 and reading == 0):
        ser.write(b'\x11')  # Send start-byte
        for _ in range(bytecount):
            try:
                byte = tx_radio_bytes.get_nowait()
                ser.write(byte)
            except (asyncio.QueueEmpty,asyncio.TimeoutError):
                break
        ser.write(b'\x13')  
        reading = 1
      await asyncio.sleep(0.001)
      
async def receivebytes(rd, wrt, rx_radio_bytes):
    print("Receiving")
    global conn_good
    global reading
    while(True):
      rxbyte = await rd.read(1)
      #print(rxbyte)         
      if(conn_good == 0): 
        if(rxbyte == b'\x12'):
          print("got connect byte")
          wrt.write(b'\x14')
          conn_good = 1
          continue
        #print("Waiting for server...")
        continue
      
      if(rxbyte == b'\x12'):
        print("Connect-byte out of seq, quit and restart")
        conn_good = 0  
        continue
      
      if (rxbyte == b'\x11'):
        reading = 1
        #print("Got data start byte")
        continue
     
      if (rxbyte == b'\x13'):
        #print("Got data end byte")
        reading = 0
        continue
        
      #print("Adding to receive queue")
      rx_radio_bytes.put_nowait(rxbyte)
        
async def main():
    radio_to_local = asyncio.Queue(maxsize=0)
    local_to_radio = asyncio.Queue(maxsize=0)
    

    # Create serial connection for reading
    reader, writer = await serial_asyncio.open_serial_connection(url='./reader', baudrate=115200)
    radioreader, radiowriter = await serial_asyncio.open_serial_connection(url='/dev/ttyUSB1', baudrate=115200)
    
    read_task = asyncio.create_task(read_serial(reader, local_to_radio))
    write_task = asyncio.create_task(write_serial(writer, radio_to_local))
    
    receive_task = asyncio.create_task(receivebytes(radioreader, radiowriter, radio_to_local))
    radiowrite_task = asyncio.create_task(sendbytes(radiowriter, local_to_radio, bytes_to_send))

    # Run all tasks concurrently
    await asyncio.wait([read_task, write_task, radiowrite_task, receive_task])

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

For this version, increasing the asyncio.sleep() calls reduces CPU usage, but cuts into throughput first - so that's not great.

Question 3: For the multiprocessing version, could I add some sleep times somewhere without missing data to reduce load too? If so, where? I tried adding some in the loops that service the ports, but they have the same issues as my previous point.

Note: I do not want to fully rewrite this code, but I do want to learn how to optimize it for CPU usage - and if I can't, why not.

Cheers, R

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.