Direct interaction with subprocess - the curse of blocking I/O
Scott David Daniels
Scott.Daniels at Acm.Org
Fri Jul 3 11:43:47 EDT 2009
yam850 wrote:
> I made a python method/function for non blocking read from a file
> object.... I am happy to see comments.
OK, here's a fairly careful set of comments with a few caveats:
Does this work on windows? The top comment should say where you
know it works. Does this code correctly read a single line?
perhaps you need to check for a final '\n' whenever you actually
get characters, and break out there. The rest below simply is
style-changing suggestions; take what you like and leave the rest.
>
> def non_blocking_readline(f_read=sys.stdin, timeout_select=0.0):
> """to readline non blocking from the file object 'f_read'
> for 'timeout_select' see module 'select'"""
> import select
Typically put imports at the top of the source
XXX text_lines = '' # empty string
Either no comment here or say _why_ it is empty.
(A) text_lines = [] # for result accumulation
if collecting pieces (see below)
(B) text_lines = '' # for result accumulation
if collecting a single string
> while True: # as long as there are bytes to read
> try: # try select
> rlist, wlist, xlist = select.select([f_read], [], [],
> timeout_select)
XXX except: # select ERROR
XXX print >>sys.stderr, ("non_blocking_read select ERROR")
I'd not hide the details of the exception like that. Don't do empty
excepts. Either don't catch it at all here, (I prefer that) or do
something like the following to capture the error:
except Exception, why:
print >>sys.stderr, "non_blocking_read select: %s" % why
> break
XXX if DEBUG: print("rlist=%s, wlist=%s, xlist=%s" % (repr(rlist),
XXX repr(wlist), repr(xlist)))
Don't be scared of vertical space; if you must, define a function DPRINT
to ignore args; use %r to get repr
So, either:
if DEBUG:
print("rlist=%r, wlist=%r, xlist=%r" % (
rlist, wlist, xlist))
or:
elsewhere:
def DPRINT(format_str, *args, **kwargs):
'''Conditionally print formatted output based on DEBUG'''
if DEBUG:
print(format_str % (args or kwargs))
and then here:
DPRINT("rlist=%r, wlist=%r, xlist=%r", rlist, wlist, xlist)
XXX if len(rlist) > 0:
Idiomatic Python: use the fact that empty containers evaluate false:
if rlist:
> text_read = f_read.readline() # get a line
XXX if DEBUG: print("after read/readline text_read:'%s', len=
> %s" % (text_read, repr(len(text_read))))
Similar comment to above:
if DEBUG:
print("after read/readline text_read:%r, len=%s" % (
text_read, len(text_read))
or:
DPRINT("after read/readline text_read:%r, len=%s",
text_read, len(text_read))
XXX if len(text_read) > 0: # there were some bytes
if text_read: # there were some bytes
XXX text_lines = "".join([text_lines, text_read])
Note the join is a good idea only if you have several elements.
For a single concatenation, "=" is just fine. The (A) case combines
at the end, the (B) case if you expect multiple concatenates are rare.
(A) text_lines.append(text_read)
(B) text_lines += text_read
XXX if DEBUG: print("text_lines:'%s'" % (text_lines))
Similar to above
if DEBUG:
print("text_lines:%r" % text_lines)
or:
DPRINT("text_lines:%r", text_lines)
XXX else:
XXX break # there was no byte in a line
XXX else:
XXX break # there was no byte in the f_read
To make the flow of control above more clear (one case continues, others
get out), I'd also replace the above with:
continue # In one case we keep going
break # No new chars found, let's get out.
XXX if text_lines == '':
XXX return None
XXX else:
XXX return text_lines
Simplify using 'or's semantics:
(A) return ''.join(text_lines) or None
(B) return text_lines or None
So, with everything applied:
import select
def DPRINT(format_str, *args, **kwargs):
'''Conditionally print formatted output based on DEBUG'''
if DEBUG:
print(format_str % (args or kwargs))
def non_blocking_readline(f_read=sys.stdin, timeout_select=0.0):
"""to readline non blocking from the file object 'f_read'
for 'timeout_select' see module 'select'
"""
text_lines = '' # for result accumulation
while True: # as long as there are bytes to read
rlist, wlist, xlist = select.select([f_read], [], [],
timeout_select)
DPRINT("rlist=%r, wlist=%r, xlist=%r",
rlist, wlist, xlist)
if rlist:
text_read = f_read.readline() # get a line
DPRINT("after read/readline text_read:%r, len=%s",
text_read, len(text_read))
if text_read: # there were some bytes
text_lines += text_read
DPRINT("text_lines:%r", text_lines)
continue # Got some chars, keep going.
break # Nothing new found, let's get out.
return text_lines or None
--Scott David Daniels
Scott.Daniels at Acm.Org
More information about the Python-list
mailing list