Merge pull request #891 from benfrankel/master

Use incremental utf-8 decoder for buffered reading
This commit is contained in:
lantz
2019-07-11 17:15:37 -07:00
committed by GitHub
2 changed files with 68 additions and 34 deletions
+12 -9
View File
@@ -63,7 +63,7 @@ from time import sleep
from mininet.log import info, error, warn, debug from mininet.log import info, error, warn, debug
from mininet.util import ( quietRun, errRun, errFail, moveIntf, isShellBuiltin, from mininet.util import ( quietRun, errRun, errFail, moveIntf, isShellBuiltin,
numCores, retry, mountCgroups, BaseString, decode, numCores, retry, mountCgroups, BaseString, decode,
encode, Python3, which ) encode, getincrementaldecoder, Python3, which )
from mininet.moduledeps import moduleDeps, pathCheck, TUN from mininet.moduledeps import moduleDeps, pathCheck, TUN
from mininet.link import Link, Intf, TCIntf, OVSIntf from mininet.link import Link, Intf, TCIntf, OVSIntf
from re import findall from re import findall
@@ -106,6 +106,9 @@ class Node( object ):
self.waiting = False self.waiting = False
self.readbuf = '' self.readbuf = ''
# Incremental decoder for buffered reading
self.decoder = getincrementaldecoder()
# Start command interpreter shell # Start command interpreter shell
self.master, self.slave = None, None # pylint self.master, self.slave = None, None # pylint
self.startShell() self.startShell()
@@ -229,19 +232,19 @@ class Node( object ):
# Subshell I/O, commands and control # Subshell I/O, commands and control
def read( self, maxbytes=1024 ): def read( self, size=1024 ):
"""Buffered read from node, potentially blocking. """Buffered read from node, potentially blocking.
maxbytes: maximum number of bytes to return""" size: maximum number of characters to return"""
count = len( self.readbuf ) count = len( self.readbuf )
if count < maxbytes: if count < size:
data = decode( os.read( self.stdout.fileno(), maxbytes - count ) ) data = os.read( self.stdout.fileno(), size - count )
self.readbuf += data self.readbuf += self.decoder.decode( data )
if maxbytes >= len( self.readbuf ): if size >= len( self.readbuf ):
result = self.readbuf result = self.readbuf
self.readbuf = '' self.readbuf = ''
else: else:
result = self.readbuf[ :maxbytes ] result = self.readbuf[ :size ]
self.readbuf = self.readbuf[ maxbytes: ] self.readbuf = self.readbuf[ size: ]
return result return result
def readline( self ): def readline( self ):
+45 -14
View File
@@ -13,23 +13,48 @@ from os import O_NONBLOCK
import os import os
from functools import partial from functools import partial
import sys import sys
import codecs
# Python 2/3 compatibility # Python 2/3 compatibility
Python3 = sys.version_info[0] == 3 Python3 = sys.version_info[0] == 3
BaseString = str if Python3 else getattr( str, '__base__' ) BaseString = str if Python3 else getattr( str, '__base__' )
Encoding = 'utf-8' if Python3 else None Encoding = 'utf-8' if Python3 else None
def decode( s ): class NullCodec( object ):
"Decode a byte string if needed for Python 3" "Null codec for Python 2"
return s.decode( Encoding ) if Python3 else s @staticmethod
def encode( s ): def decode( buf ):
"Encode a byte string if needed for Python 3" "Null decode"
return s.encode( Encoding ) if Python3 else s return buf
@staticmethod
def encode( buf ):
"Null encode"
return buf
if Python3:
def decode( buf ):
"Decode buffer for Python 3"
return buf.decode( Encoding )
def encode( buf ):
"Encode buffer for Python 3"
return buf.encode( Encoding )
getincrementaldecoder = codecs.getincrementaldecoder( Encoding )
else:
decode, encode = NullCodec.decode, NullCodec.encode
def getincrementaldecoder():
"Return null codec for Python 2"
return NullCodec
try: try:
# pylint: disable=import-error # pylint: disable=import-error
oldpexpect = None oldpexpect = None
import pexpect as oldpexpect import pexpect as oldpexpect
# pylint: enable=import-error
# pylint: enable=import-error
class Pexpect( object ): class Pexpect( object ):
"Custom pexpect that is compatible with str" "Custom pexpect that is compatible with str"
@staticmethod @staticmethod
@@ -119,20 +144,21 @@ def errRun( *cmd, **kwargs ):
out, err = '', '' out, err = '', ''
poller = poll() poller = poll()
poller.register( popen.stdout, POLLIN ) poller.register( popen.stdout, POLLIN )
fdtofile = { popen.stdout.fileno(): popen.stdout } fdToFile = { popen.stdout.fileno(): popen.stdout }
fdToDecoder = { popen.stdout.fileno(): getincrementaldecoder() }
outDone, errDone = False, True outDone, errDone = False, True
if popen.stderr: if popen.stderr:
fdtofile[ popen.stderr.fileno() ] = popen.stderr fdToFile[ popen.stderr.fileno() ] = popen.stderr
fdToDecoder[ popen.stderr.fileno() ] = getincrementaldecoder()
poller.register( popen.stderr, POLLIN ) poller.register( popen.stderr, POLLIN )
errDone = False errDone = False
while not outDone or not errDone: while not outDone or not errDone:
readable = poller.poll() readable = poller.poll()
for fd, event in readable: for fd, event in readable:
f = fdtofile[ fd ] f = fdToFile[ fd ]
decoder = fdToDecoder[ fd ]
if event & POLLIN: if event & POLLIN:
data = f.read( 1024 ) data = decoder.decode( f.read( 1024 ) )
if Python3:
data = data.decode( Encoding )
if echo: if echo:
output( data ) output( data )
if f == popen.stdout: if f == popen.stdout:
@@ -187,8 +213,10 @@ def isShellBuiltin( cmd ):
cmd = cmd[ :space] cmd = cmd[ :space]
return cmd in isShellBuiltin.builtIns return cmd in isShellBuiltin.builtIns
isShellBuiltin.builtIns = None isShellBuiltin.builtIns = None
# Interface management # Interface management
# #
# Interfaces are managed as strings which are simply the # Interfaces are managed as strings which are simply the
@@ -418,9 +446,11 @@ def pmonitor(popens, timeoutms=500, readline=True,
terminates: when all EOFs received""" terminates: when all EOFs received"""
poller = poll() poller = poll()
fdToHost = {} fdToHost = {}
fdToDecoder = {}
for host, popen in popens.items(): for host, popen in popens.items():
fd = popen.stdout.fileno() fd = popen.stdout.fileno()
fdToHost[ fd ] = host fdToHost[ fd ] = host
fdToDecoder[ fd ] = getincrementaldecoder()
poller.register( fd, POLLIN | POLLHUP ) poller.register( fd, POLLIN | POLLHUP )
flags = fcntl( fd, F_GETFL ) flags = fcntl( fd, F_GETFL )
fcntl( fd, F_SETFL, flags | O_NONBLOCK ) fcntl( fd, F_SETFL, flags | O_NONBLOCK )
@@ -429,12 +459,13 @@ def pmonitor(popens, timeoutms=500, readline=True,
if fds: if fds:
for fd, event in fds: for fd, event in fds:
host = fdToHost[ fd ] host = fdToHost[ fd ]
decoder = fdToDecoder[ fd ]
popen = popens[ host ] popen = popens[ host ]
if event & POLLIN or event & POLLHUP: if event & POLLIN or event & POLLHUP:
while True: while True:
try: try:
f = popen.stdout f = popen.stdout
line = decode( f.readline() if readline line = decoder.decode( f.readline() if readline
else f.read( readmax ) ) else f.read( readmax ) )
except IOError: except IOError:
line = '' line = ''