Async serial communication/threads sharing data

[Nick Craig-Wood]

Jean-Paul Calderone <exarkun at divmod.com> wrote:
>  On Sun, 22 Mar 2009 12:30:04 -0500, Nick Craig-Wood <nick at craig-wood.com> wrote:
> >I wrote a serial port to TCP proxy (with logging) with twisted.  The
> >problem I had was that twisted serial ports didn't seem to have any
> >back pressure.  By that I mean I could pump data into a 9600 baud
> >serial port at 10 Mbit/s.  Twisted would then buffer the data for me
> >using 10s or 100s or Megabytes of RAM.  No data would be lost, but
> >there would be hours of latency and my program would use up all my RAM
> >and explode.
> >
> >What I wanted to happen was for twisted to stop taking the data when
> >the serial port buffer was full and to only take the data at 9600
> >baud.
> 
>  This is what Twisted's producers and consumers let you do.  There's a
>  document covering these features:
> 
>  http://twistedmatrix.com/projects/core/documentation/howto/producers.html
> 
>  In the case of a TCP to serial forwarder, you don't actually have to
>  implement either a producer or a consumer, since both the TCP connection
>  and the serial connection are already both producers and consumers.  All
>  you need to do is hook them up to each other so that when the send buffer
>  of one fills up, the other one gets paused, and when the buffer is empty
>  again, it gets resumed.

I eventually came up with this which seems to work, but I'm not sure
it is the best way of doing it as it had to mess about with the
twisted internals to get the number of bytes in the serial port
output buffer.

class SerialPort(protocol.Protocol):
    """Create a serial port connection and pass data from it to a known list of TCP ports."""

    def __init__(self, port, reactor, baudrate, log_name=None):
        self.tcp_ports = []
        self.serial = serialport.SerialPort(self, reactor, port, baudrate, rtscts=0)
        self.log = None
        if log_name is not None:
            self.log = file('%s-0' % log_name, 'w')

    def write(self, data):
        """Write data to the serial port."""
        self.serial.write(data)
        if self.log:
            self.log.write(data)
        self.throttle()

    def throttle(self):
        """
        Pause the inputs if there is too much data in the output buffer
        """
        bytes_in_buffer = len(self.serial.dataBuffer) +	self.serial._tempDataLen
        too_full = bytes_in_buffer > 1024
        for tcp_port in self.tcp_ports:
            if too_full:
                tcp_port.transport.pauseProducing()
            else:
                tcp_port.transport.resumeProducing()
        if too_full:
            reactor.callLater(0.1, self.throttle)

-- 
Nick Craig-Wood <nick at craig-wood.com> -- http://www.craig-wood.com/nick