<s> import collections <EOL> import errno <EOL> import fcntl <EOL> import gevent <EOL> import gevent . socket <EOL> import os <EOL> PIPE_BUF_BYTES = None <EOL> OS_PIPE_SZ = None <EOL> def _configure_buffer_sizes ( ) : <EOL> """<STR_LIT>""" <EOL> global PIPE_BUF_BYTES <EOL> global OS_PIPE_SZ <EOL> PIPE_BUF_BYTES = <NUM_LIT> <EOL> OS_PIPE_SZ = None <EOL> if not hasattr ( fcntl , '<STR_LIT>' ) : <EOL> import platform <EOL> if platform . system ( ) == '<STR_LIT>' : <EOL> fcntl . F_SETPIPE_SZ = <NUM_LIT> <EOL> try : <EOL> with open ( '<STR_LIT>' , '<STR_LIT:r>' ) as f : <EOL> OS_PIPE_SZ = min ( int ( f . read ( ) ) , <NUM_LIT> * <NUM_LIT> ) <EOL> PIPE_BUF_BYTES = max ( OS_PIPE_SZ , PIPE_BUF_BYTES ) <EOL> except : <EOL> pass <EOL> _configure_buffer_sizes ( ) <EOL> def set_buf_size ( fd ) : <EOL> """<STR_LIT>""" <EOL> if OS_PIPE_SZ and hasattr ( fcntl , '<STR_LIT>' ) : <EOL> fcntl . fcntl ( fd , fcntl . F_SETPIPE_SZ , OS_PIPE_SZ ) <EOL> def _setup_fd ( fd ) : <EOL> """<STR_LIT>""" <EOL> flags = fcntl . fcntl ( fd , fcntl . F_GETFL ) <EOL> fcntl . fcntl ( fd , fcntl . F_SETFL , flags | os . O_NONBLOCK ) <EOL> set_buf_size ( fd ) <EOL> class ByteDeque ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _dq = collections . deque ( ) <EOL> self . byteSz = <NUM_LIT:0> <EOL> def add ( self , b ) : <EOL> self . _dq . append ( b ) <EOL> self . byteSz += len ( b ) <EOL> def get ( self , n ) : <EOL> assert n <= self . byteSz , '<STR_LIT>' <EOL> if ( n == self . byteSz and len ( self . _dq ) == <NUM_LIT:1> and <EOL> isinstance ( self . _dq [ <NUM_LIT:0> ] , bytes ) ) : <EOL> self . byteSz = <NUM_LIT:0> <EOL> return self . _dq . popleft ( ) <EOL> out = bytearray ( n ) <EOL> remaining = n <EOL> while remaining > <NUM_LIT:0> : <EOL> part = self . _dq . popleft ( ) <EOL> delta = remaining - len ( part ) <EOL> offset = n - remaining <EOL> if delta == <NUM_LIT:0> : <EOL> out [ offset : ] = part <EOL> remaining = <NUM_LIT:0> <EOL> elif delta > <NUM_LIT:0> : <EOL> out [ offset : ] = part <EOL> remaining = delta <EOL> elif delta < <NUM_LIT:0> : <EOL> cleave = len ( part ) + delta <EOL> out [ offset : ] = buffer ( part , <NUM_LIT:0> , cleave ) <EOL> self . _dq . appendleft ( buffer ( part , cleave ) ) <EOL> remaining = <NUM_LIT:0> <EOL> else : <EOL> assert False <EOL> self . byteSz -= n <EOL> assert len ( out ) == n <EOL> return bytes ( out ) <EOL> def get_all ( self ) : <EOL> return self . get ( self . byteSz ) <EOL> class NonBlockBufferedReader ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fp ) : <EOL> self . _fp = fp <EOL> self . _fd = fp . fileno ( ) <EOL> self . _bd = ByteDeque ( ) <EOL> self . got_eof = False <EOL> _setup_fd ( self . _fd ) <EOL> def _read_chunk ( self , sz ) : <EOL> chunk = None <EOL> try : <EOL> chunk = os . read ( self . _fd , sz ) <EOL> self . _bd . add ( chunk ) <EOL> except EnvironmentError , e : <EOL> if e . errno in [ errno . EAGAIN , errno . EWOULDBLOCK ] : <EOL> assert chunk is None <EOL> gevent . socket . wait_read ( self . _fd ) <EOL> else : <EOL> raise <EOL> self . got_eof = ( chunk == '<STR_LIT>' ) <EOL> def read ( self , size = None ) : <EOL> if size is None : <EOL> while not self . got_eof : <EOL> self . _read_chunk ( PIPE_BUF_BYTES ) <EOL> return self . _bd . get_all ( ) <EOL> elif size > <NUM_LIT:0> : <EOL> while True : <EOL> if self . _bd . byteSz >= size : <EOL> return self . _bd . get ( size ) <EOL> elif self . _bd . byteSz <= size and self . got_eof : <EOL> return self . _bd . get_all ( ) <EOL> else : <EOL> assert not self . got_eof <EOL> if size == PIPE_BUF_BYTES : <EOL> to_read = PIPE_BUF_BYTES - self . _bd . byteSz <EOL> self . _read_chunk ( to_read ) <EOL> else : <EOL> self . _read_chunk ( PIPE_BUF_BYTES ) <EOL> else : <EOL> assert False <EOL> def close ( self ) : <EOL> if self . closed : <EOL> return <EOL> self . _fp . close ( ) <EOL> try : <EOL> del self . _fp <EOL> except AttributeError : <EOL> pass <EOL> try : <EOL> del self . _bd <EOL> except AttributeError : <EOL> pass <EOL> self . _fd = - <NUM_LIT:1> <EOL> def fileno ( self ) : <EOL> return self . _fd <EOL> @ property <EOL> def closed ( self ) : <EOL> return self . _fd == - <NUM_LIT:1> <EOL> class NonBlockBufferedWriter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fp ) : <EOL> self . _fp = fp <EOL> self . _fd = fp . fileno ( ) <EOL> self . _bd = ByteDeque ( ) <EOL> _setup_fd ( self . _fd ) <EOL> def _partial_flush ( self , max_retain ) : <EOL> byts = self . _bd . get_all ( ) <EOL> cursor = buffer ( byts ) <EOL> flushed = False <EOL> while len ( cursor ) > max_retain : <EOL> try : <EOL> n = os . write ( self . _fd , cursor ) <EOL> flushed = True <EOL> cursor = buffer ( cursor , n ) <EOL> except EnvironmentError , e : <EOL> if e . errno in [ errno . EAGAIN , errno . EWOULDBLOCK ] : <EOL> gevent . socket . wait_write ( self . _fd ) <EOL> else : <EOL> raise <EOL> assert self . _bd . byteSz == <NUM_LIT:0> <EOL> if len ( cursor ) > <NUM_LIT:0> : <EOL> self . _bd . add ( cursor ) <EOL> return flushed <EOL> def write ( self , data ) : <EOL> self . _bd . add ( data ) <EOL> flushed = True <EOL> while flushed and self . _bd . byteSz > PIPE_BUF_BYTES : <EOL> flushed = self . _partial_flush ( <NUM_LIT> ) <EOL> def flush ( self ) : <EOL> while self . _bd . byteSz > <NUM_LIT:0> : <EOL> self . _partial_flush ( <NUM_LIT:0> ) <EOL> def fileno ( self ) : <EOL> return self . _fd <EOL> def close ( self ) : <EOL> if self . closed : <EOL> return <EOL> self . _fp . close ( ) <EOL> try : <EOL> del self . _fp <EOL> except AttributeError : <EOL> pass <EOL> try : <EOL> del self . _bd <EOL> except AttributeError : <EOL> pass <EOL> self . _fd = - <NUM_LIT:1> <EOL> @ property <EOL> def closed ( self ) : <EOL> return self . _fd == - <NUM_LIT:1> </s>
