<s> import errno <EOL> import sys <EOL> import logging <EOL> import select <EOL> import socket <EOL> import time <EOL> import threading <EOL> from pox . lib . util import makePinger <EOL> from pox . lib . ioworker . io_worker import IOWorker <EOL> log = logging . getLogger ( "<STR_LIT>" ) <EOL> class STSIOWorker ( IOWorker ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , socket , on_close ) : <EOL> IOWorker . __init__ ( self ) <EOL> self . socket = socket <EOL> self . closed = False <EOL> self . on_close = on_close <EOL> def fileno ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . socket . fileno ( ) <EOL> def send ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return IOWorker . send ( self , data ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> IOWorker . close ( self ) <EOL> self . on_close ( self ) <EOL> class IOMaster ( object ) : <EOL> """<STR_LIT>""" <EOL> _select_timeout = <NUM_LIT:5> <EOL> _BUF_SIZE = <NUM_LIT> <EOL> def __init__ ( self ) : <EOL> self . _workers = set ( ) <EOL> self . pinger = makePinger ( ) <EOL> self . closed = False <EOL> self . _close_requested = False <EOL> self . _in_select = <NUM_LIT:0> <EOL> def create_worker_for_socket ( self , socket ) : <EOL> '''<STR_LIT>''' <EOL> def on_close ( worker ) : <EOL> worker . socket . close ( ) <EOL> worker . closed = True <EOL> self . _workers . discard ( worker ) <EOL> worker = STSIOWorker ( socket , on_close = on_close ) <EOL> self . _workers . add ( worker ) <EOL> return worker <EOL> def monkey_time_sleep ( self ) : <EOL> """<STR_LIT>""" <EOL> self . original_time_sleep = time . sleep <EOL> time . _orig_sleep = time . sleep <EOL> time . sleep = self . sleep <EOL> def raw_input ( self , prompt ) : <EOL> """<STR_LIT>""" <EOL> _io_master = self <EOL> class BackgroundIOThread ( threading . Thread ) : <EOL> def __init__ ( self ) : <EOL> threading . Thread . __init__ ( self , name = "<STR_LIT>" ) <EOL> self . done = False <EOL> def run ( self ) : <EOL> while not self . done : <EOL> _io_master . select ( None ) <EOL> self . sleep ( <NUM_LIT> ) <EOL> io_thread = BackgroundIOThread ( ) <EOL> io_thread . daemon = False <EOL> io_thread . start ( ) <EOL> try : <EOL> return raw_input ( prompt ) <EOL> finally : <EOL> """<STR_LIT>""" <EOL> io_thread . done = True <EOL> self . _ping ( ) <EOL> io_thread . join ( ) <EOL> def _ping ( self ) : <EOL> if self . pinger : <EOL> self . pinger . ping ( ) <EOL> def close_all ( self ) : <EOL> if self . _in_select > <NUM_LIT:0> : <EOL> self . _close_requested = True <EOL> self . _ping ( ) <EOL> else : <EOL> self . _do_close_all ( ) <EOL> def _do_close_all ( self ) : <EOL> for w in list ( self . _workers ) : <EOL> try : <EOL> w . close ( ) <EOL> except Exception as e : <EOL> log . warn ( "<STR_LIT>" , w , e . strerror , e . errno ) <EOL> if time . sleep is self . sleep : <EOL> time . sleep = self . original_time_sleep <EOL> if ( self . pinger ) : <EOL> self . pinger . ping ( ) <EOL> if hasattr ( self . pinger , "<STR_LIT>" ) : <EOL> self . pinger . close ( ) <EOL> self . pinger = None <EOL> self . closed = True <EOL> def poll ( self ) : <EOL> self . select ( <NUM_LIT:0> ) <EOL> def sleep ( self , timeout ) : <EOL> '''<STR_LIT>''' <EOL> start = time . time ( ) <EOL> while not self . closed : <EOL> elapsed = time . time ( ) - start <EOL> remaining = timeout - elapsed <EOL> if remaining < <NUM_LIT> : <EOL> break <EOL> self . select ( remaining ) <EOL> def deschedule_worker ( self , io_worker ) : <EOL> self . _workers . discard ( io_worker ) <EOL> def reschedule_worker ( self , io_worker ) : <EOL> self . _workers . add ( io_worker ) <EOL> def grab_workers_rwe ( self ) : <EOL> read_sockets = list ( self . _workers ) + [ self . pinger ] <EOL> write_sockets = [ worker for worker in self . _workers if worker . _ready_to_send ] <EOL> exception_sockets = list ( self . _workers ) <EOL> return ( read_sockets , write_sockets , exception_sockets ) <EOL> def select ( self , timeout = <NUM_LIT:0> ) : <EOL> '''<STR_LIT>''' <EOL> self . _in_select += <NUM_LIT:1> <EOL> try : <EOL> read_sockets , write_sockets , exception_sockets = self . grab_workers_rwe ( ) <EOL> rlist , wlist , elist = select . select ( read_sockets , write_sockets , exception_sockets , timeout ) <EOL> self . handle_workers_rwe ( rlist , wlist , elist ) <EOL> except select . error : <EOL> sys . stderr . write ( "<STR_LIT>" ) <EOL> except TypeError : <EOL> sys . stderr . write ( "<STR_LIT>" ) <EOL> finally : <EOL> self . _in_select -= <NUM_LIT:1> <EOL> if self . _in_select == <NUM_LIT:0> and self . _close_requested and not self . closed : <EOL> self . _do_close_all ( ) <EOL> def handle_workers_rwe ( self , rlist , wlist , elist ) : <EOL> if self . pinger in rlist : <EOL> self . pinger . pongAll ( ) <EOL> rlist . remove ( self . pinger ) <EOL> for worker in elist : <EOL> worker . close ( ) <EOL> if worker in self . _workers : <EOL> self . _workers . discard ( worker ) <EOL> for worker in rlist : <EOL> try : <EOL> data = worker . socket . recv ( self . _BUF_SIZE ) <EOL> if data : <EOL> worker . _push_receive_data ( data ) <EOL> else : <EOL> log . warn ( "<STR_LIT>" ) <EOL> worker . close ( ) <EOL> self . _workers . discard ( worker ) <EOL> except socket . error as ( s_errno , strerror ) : <EOL> log . error ( "<STR_LIT>" + strerror ) <EOL> worker . close ( ) <EOL> self . _workers . discard ( worker ) <EOL> for worker in wlist : <EOL> try : <EOL> l = worker . socket . send ( worker . send_buf ) <EOL> if l > <NUM_LIT:0> : <EOL> worker . _consume_send_buf ( l ) <EOL> except socket . error as ( s_errno , strerror ) : <EOL> if s_errno != errno . EAGAIN : <EOL> log . error ( "<STR_LIT>" + strerror ) <EOL> worker . close ( ) <EOL> self . _workers . discard ( worker ) </s>
