Use incremental UTF-8 decoder for buffered reading

mininet/node.py

@@ -63,7 +63,7 @@ from time import sleep
 from mininet.log import info, error, warn, debug
 from mininet.util import ( quietRun, errRun, errFail, moveIntf, isShellBuiltin,
                            numCores, retry, mountCgroups, BaseString, decode,
-                           encode, Python3, which )
+                           encode, getincrementaldecoder, Python3, which )
 from mininet.moduledeps import moduleDeps, pathCheck, TUN
 from mininet.link import Link, Intf, TCIntf, OVSIntf
 from re import findall
@@ -106,6 +106,9 @@ class Node( object ):
         self.waiting = False
         self.readbuf = ''
 
+        # Incremental decoder for buffered reading
+        self.decoder = getincrementaldecoder()
+
         # Start command interpreter shell
         self.master, self.slave = None, None  # pylint
         self.startShell()
@@ -229,19 +232,19 @@ class Node( object ):
 
     # Subshell I/O, commands and control
 
-    def read( self, maxbytes=1024 ):
+    def read( self, size=1024 ):
         """Buffered read from node, potentially blocking.
-           maxbytes: maximum number of bytes to return"""
+           size: maximum number of characters to return"""
         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
             self.readbuf = ''
         else:
-            result = self.readbuf[ :maxbytes ]
+            result = self.readbuf[ :size ]
-            self.readbuf = self.readbuf[ maxbytes: ]
+            self.readbuf = self.readbuf[ size: ]
         return result
 
     def readline( self ):

mininet/util.py

@@ -13,23 +13,48 @@ from os import O_NONBLOCK
 import os
 from functools import partial
 import sys
+import codecs
 
 # Python 2/3 compatibility
+
 Python3 = sys.version_info[0] == 3
 BaseString = str if Python3 else getattr( str, '__base__' )
 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:
     # pylint: disable=import-error
     oldpexpect = None
     import pexpect as oldpexpect
-    # pylint: enable=import-error
 
+    # pylint: enable=import-error
     class Pexpect( object ):
         "Custom pexpect that is compatible with str"
         @staticmethod
@@ -119,20 +144,21 @@ def errRun( *cmd, **kwargs ):
     out, err = '', ''
     poller = poll()
     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
     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 )
         errDone = False
     while not outDone or not errDone:
         readable = poller.poll()
         for fd, event in readable:
-            f = fdtofile[ fd ]
+            f = fdToFile[ fd ]
+            decoder = fdToDecoder[ fd ]
             if event & POLLIN:
-                data = f.read( 1024 )
+                data = decoder.decode( f.read( 1024 ) )
-                if Python3:
-                    data = data.decode( Encoding )
                 if echo:
                     output( data )
                 if f == popen.stdout:
@@ -187,8 +213,10 @@ def isShellBuiltin( cmd ):
         cmd = cmd[ :space]
     return cmd in isShellBuiltin.builtIns
 
+
 isShellBuiltin.builtIns = None
 
+
 # Interface management
 #
 # Interfaces are managed as strings which are simply the
@@ -375,7 +403,7 @@ def netParse( ipstr ):
     if '/' in ipstr:
         ip, pf = ipstr.split( '/' )
         prefixLen = int( pf )
-    #if no prefix is specified, set the prefix to 24
+    # if no prefix is specified, set the prefix to 24
     else:
         ip = ipstr
         prefixLen = 24
@@ -418,9 +446,11 @@ def pmonitor(popens, timeoutms=500, readline=True,
        terminates: when all EOFs received"""
     poller = poll()
     fdToHost = {}
+    fdToDecoder = {}
     for host, popen in popens.items():
         fd = popen.stdout.fileno()
         fdToHost[ fd ] = host
+        fdToDecoder[ fd ] = getincrementaldecoder()
         poller.register( fd, POLLIN | POLLHUP )
         flags = fcntl( fd, F_GETFL )
         fcntl( fd, F_SETFL, flags | O_NONBLOCK )
@@ -429,13 +459,14 @@ def pmonitor(popens, timeoutms=500, readline=True,
         if fds:
             for fd, event in fds:
                 host = fdToHost[ fd ]
+                decoder = fdToDecoder[ fd ]
                 popen = popens[ host ]
                 if event & POLLIN or event & POLLHUP:
                     while True:
                         try:
                             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:
                             line = ''
                         if line == '':
@@ -450,19 +481,19 @@ def pmonitor(popens, timeoutms=500, readline=True,
 # Other stuff we use
 def sysctlTestAndSet( name, limit ):
     "Helper function to set sysctl limits"
-    #convert non-directory names into directory names
+    # convert non-directory names into directory names
     if '/' not in name:
         name = '/proc/sys/' + name.replace( '.', '/' )
-    #read limit
+    # read limit
     with open( name, 'r' ) as readFile:
         oldLimit = readFile.readline()
         if isinstance( limit, int ):
-            #compare integer limits before overriding
+            # compare integer limits before overriding
             if int( oldLimit ) < limit:
                 with open( name, 'w' ) as writeFile:
                     writeFile.write( "%d" % limit )
         else:
-            #overwrite non-integer limits
+            # overwrite non-integer limits
             with open( name, 'w' ) as writeFile:
                 writeFile.write( limit )
 
@@ -479,21 +510,21 @@ def fixLimits():
     try:
         rlimitTestAndSet( RLIMIT_NPROC, 8192 )
         rlimitTestAndSet( RLIMIT_NOFILE, 16384 )
-        #Increase open file limit
+        # Increase open file limit
         sysctlTestAndSet( 'fs.file-max', 10000 )
-        #Increase network buffer space
+        # Increase network buffer space
         sysctlTestAndSet( 'net.core.wmem_max', 16777216 )
         sysctlTestAndSet( 'net.core.rmem_max', 16777216 )
         sysctlTestAndSet( 'net.ipv4.tcp_rmem', '10240 87380 16777216' )
         sysctlTestAndSet( 'net.ipv4.tcp_wmem', '10240 87380 16777216' )
         sysctlTestAndSet( 'net.core.netdev_max_backlog', 5000 )
-        #Increase arp cache size
+        # Increase arp cache size
         sysctlTestAndSet( 'net.ipv4.neigh.default.gc_thresh1', 4096 )
         sysctlTestAndSet( 'net.ipv4.neigh.default.gc_thresh2', 8192 )
         sysctlTestAndSet( 'net.ipv4.neigh.default.gc_thresh3', 16384 )
-        #Increase routing table size
+        # Increase routing table size
         sysctlTestAndSet( 'net.ipv4.route.max_size', 32768 )
-        #Increase number of PTYs for nodes
+        # Increase number of PTYs for nodes
         sysctlTestAndSet( 'kernel.pty.max', 20000 )
     # pylint: disable=broad-except
     except Exception: