<s> """<STR_LIT>""" <EOL> __author__ = "<STR_LIT>" <EOL> __email__ = "<STR_LIT>" <EOL> __copyright__ = "<STR_LIT>" <EOL> __contributors__ = [ ] <EOL> __maintainer__ = "<STR_LIT>" <EOL> __license__ = "<STR_LIT>" <EOL> __url__ = "<STR_LIT>" <EOL> __status__ = "<STR_LIT>" <EOL> __version__ = "<STR_LIT>" <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> import time <EOL> import threading <EOL> import functools <EOL> import socket <EOL> import inspect <EOL> import traceback <EOL> import select <EOL> import sys <EOL> import types <EOL> import struct <EOL> import re <EOL> import logging <EOL> import errno <EOL> import platform <EOL> import ssl <EOL> from heapq import heappush , heappop <EOL> from bisect import bisect_left <EOL> import Queue as queue <EOL> import atexit <EOL> import collections <EOL> import cPickle as pickle <EOL> import copy <EOL> if platform . system ( ) == '<STR_LIT>' : <EOL> from errno import WSAEINPROGRESS as EINPROGRESS <EOL> from errno import WSAEWOULDBLOCK as EWOULDBLOCK <EOL> from errno import WSAEINVAL as EINVAL <EOL> from time import clock as _time <EOL> _time ( ) <EOL> else : <EOL> from errno import EINPROGRESS <EOL> from errno import EWOULDBLOCK <EOL> from errno import EINVAL <EOL> from time import time as _time <EOL> def serialize ( obj ) : <EOL> return pickle . dumps ( obj , pickle . HIGHEST_PROTOCOL ) <EOL> def unserialize ( pkl ) : <EOL> return pickle . loads ( pkl ) <EOL> class MetaSingleton ( type ) : <EOL> __instance = None <EOL> def __call__ ( cls , * args , ** kwargs ) : <EOL> if cls . __instance is None : <EOL> cls . __instance = super ( MetaSingleton , cls ) . __call__ ( * args , ** kwargs ) <EOL> return cls . __instance <EOL> class Logger ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , stream = sys . stdout , level = logging . INFO ) : <EOL> """<STR_LIT>""" <EOL> self . name = name <EOL> self . stream = stream <EOL> self . level = level <EOL> self . log_ms = False <EOL> def setLevel ( self , level ) : <EOL> """<STR_LIT>""" <EOL> self . level = level <EOL> def show_ms ( self , flag ) : <EOL> """<STR_LIT>""" <EOL> self . log_ms = bool ( flag ) <EOL> def log ( self , message , * args ) : <EOL> now = time . time ( ) <EOL> if len ( args ) == <NUM_LIT:1> : <EOL> args = args [ <NUM_LIT:0> ] <EOL> message = message % args <EOL> if self . log_ms : <EOL> self . stream . write ( '<STR_LIT>' % <EOL> ( time . strftime ( '<STR_LIT>' , time . localtime ( now ) ) , <EOL> <NUM_LIT:1000> * ( now - long ( now ) ) , self . name , message ) ) <EOL> else : <EOL> self . stream . write ( '<STR_LIT>' % <EOL> ( time . strftime ( '<STR_LIT>' , time . localtime ( now ) ) , <EOL> self . name , message ) ) <EOL> debug = lambda self , message , * args : self . log ( message , * args ) if self . level <= logging . DEBUG else None <EOL> info = lambda self , message , * args : self . log ( message , * args ) if self . level <= logging . INFO else None <EOL> warn = warning = lambda self , message , * args : self . log ( message , * args ) if self . level <= logging . WARNING else None <EOL> def flush ( self ) : <EOL> self . stream . flush ( ) <EOL> shutdown = flush <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_value , trace ) : <EOL> self . shutdown ( ) <EOL> logger = Logger ( '<STR_LIT>' ) <EOL> class _AsyncSocket ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> _default_timeout = None <EOL> def __init__ ( self , sock , blocking = False , keyfile = None , certfile = None , <EOL> ssl_version = ssl . PROTOCOL_SSLv23 ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( sock , AsyncSocket ) : <EOL> logger . warning ( '<STR_LIT>' , sock . _fileno ) <EOL> for k in sock . __slots__ : <EOL> setattr ( self , k , getattr ( sock , k ) ) <EOL> else : <EOL> self . _rsock = sock <EOL> self . _keyfile = keyfile <EOL> self . _certfile = certfile <EOL> self . _ssl_version = ssl_version <EOL> self . _fileno = sock . fileno ( ) <EOL> self . _timeout = <NUM_LIT:0> <EOL> self . _timeout_id = None <EOL> self . _read_coro = None <EOL> self . _read_task = None <EOL> self . _read_result = None <EOL> self . _write_coro = None <EOL> self . _write_task = None <EOL> self . _write_result = None <EOL> self . _notifier = None <EOL> self . ssl_server_ctx = None <EOL> self . recvall = None <EOL> self . sendall = None <EOL> self . recv_msg = None <EOL> self . send_msg = None <EOL> self . _blocking = None <EOL> self . setblocking ( blocking ) <EOL> if _AsyncSocket . _default_timeout : <EOL> self . settimeout ( _AsyncSocket . _default_timeout ) <EOL> def __getattr__ ( self , name ) : <EOL> return getattr ( self . _rsock , name ) <EOL> def setblocking ( self , blocking ) : <EOL> if blocking : <EOL> blocking = True <EOL> else : <EOL> blocking = False <EOL> if self . _blocking == blocking : <EOL> return <EOL> self . _blocking = blocking <EOL> if self . _blocking : <EOL> self . _unregister ( ) <EOL> self . _rsock . setblocking ( <NUM_LIT:1> ) <EOL> if self . _certfile : <EOL> self . _rsock = ssl . wrap_socket ( self . _rsock , keyfile = self . _keyfile , <EOL> certfile = self . _certfile , <EOL> ssl_version = self . _ssl_version ) <EOL> for name in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> setattr ( self , name , getattr ( self . _rsock , name ) ) <EOL> if self . _rsock . type & socket . SOCK_STREAM : <EOL> self . recvall = self . _sync_recvall <EOL> self . sendall = self . _sync_sendall <EOL> self . recv_msg = self . _sync_recv_msg <EOL> self . send_msg = self . _sync_send_msg <EOL> self . _notifier = None <EOL> else : <EOL> self . _rsock . setblocking ( <NUM_LIT:0> ) <EOL> self . recv = self . _async_recv <EOL> self . send = self . _async_send <EOL> self . recvfrom = self . _async_recvfrom <EOL> self . sendto = self . _async_sendto <EOL> self . accept = self . _async_accept <EOL> self . connect = self . _async_connect <EOL> if self . _rsock . type & socket . SOCK_STREAM : <EOL> self . recvall = self . _async_recvall <EOL> self . sendall = self . _async_sendall <EOL> self . recv_msg = self . _async_recv_msg <EOL> self . send_msg = self . _async_send_msg <EOL> self . _notifier = _AsyncNotifier . instance ( ) <EOL> self . _register ( ) <EOL> def _register ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def _unregister ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _notifier : <EOL> self . _notifier . unregister ( self ) <EOL> self . _notifier = None <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _unregister ( ) <EOL> if self . _rsock : <EOL> self . _rsock . close ( ) <EOL> self . _rsock = None <EOL> self . _read_coro = self . _write_coro = None <EOL> def unwrap ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _unregister ( ) <EOL> self . _notifier = None <EOL> self . _read_coro = self . _write_coro = None <EOL> sock , self . _rsock = self . _rsock , None <EOL> return sock <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_value , trace ) : <EOL> self . close ( ) <EOL> def __lt__ ( self , other ) : <EOL> if isinstance ( other , AsyncSocket ) : <EOL> return self . _fileno < other . _fileno <EOL> return <NUM_LIT:1> <EOL> def setdefaulttimeout ( self , timeout ) : <EOL> if isinstance ( timeout , ( int , float ) ) and timeout > <NUM_LIT:0> : <EOL> self . _rsock . setdefaulttimeout ( timeout ) <EOL> self . _default_timeout = timeout <EOL> else : <EOL> logger . warning ( '<STR_LIT>' , timeout ) <EOL> def getdefaulttimeout ( self ) : <EOL> if self . _blocking : <EOL> return self . _rsock . getdefalttimeout ( ) <EOL> else : <EOL> return _AsyncSocket . _default_timeout <EOL> def settimeout ( self , timeout ) : <EOL> if self . _blocking : <EOL> if timeout is None : <EOL> pass <EOL> elif not timeout : <EOL> self . setblocking ( <NUM_LIT:0> ) <EOL> self . settimeout ( <NUM_LIT:0.0> ) <EOL> else : <EOL> self . _rsock . settimeout ( timeout ) <EOL> else : <EOL> if timeout is None : <EOL> self . setblocking ( <NUM_LIT:1> ) <EOL> elif isinstance ( timeout , ( int , float ) ) and timeout >= <NUM_LIT:0> : <EOL> self . _timeout = timeout <EOL> else : <EOL> logger . warning ( '<STR_LIT>' , timeout ) <EOL> def gettimeout ( self ) : <EOL> if self . _blocking : <EOL> return self . _rsock . gettimeout ( ) <EOL> else : <EOL> return self . _timeout <EOL> def _timed_out ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _read_coro : <EOL> buf = None <EOL> if isinstance ( self . _read_result , bytearray ) : <EOL> view = self . _read_task . args [ <NUM_LIT:1> ] <EOL> n = len ( self . _read_result ) - len ( view ) <EOL> if n > <NUM_LIT:0> : <EOL> buf = bytes ( self . _read_result [ : n ] ) <EOL> if buf : <EOL> self . _read_coro . _proceed_ ( buf ) <EOL> else : <EOL> self . _read_coro . throw ( socket . timeout ( '<STR_LIT>' ) ) <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = self . _read_result = self . _read_coro = None <EOL> if self . _write_coro : <EOL> sent = <NUM_LIT:0> <EOL> if isinstance ( self . _write_result , buffer ) : <EOL> sent = self . _write_task . args [ <NUM_LIT:1> ] - len ( self . _write_result ) <EOL> if sent : <EOL> self . _write_coro . _proceed_ ( sent ) <EOL> else : <EOL> self . _write_coro . throw ( socket . timeout ( '<STR_LIT>' ) ) <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = self . _write_result = self . _write_coro = None <EOL> def _eof ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _read_task : <EOL> self . _read_task ( ) <EOL> def _async_recv ( self , bufsize , * args ) : <EOL> """<STR_LIT>""" <EOL> def _recv ( self , bufsize , * args ) : <EOL> try : <EOL> buf = self . _rsock . recv ( bufsize , * args ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ : <EOL> pass <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . _proceed_ ( buf ) <EOL> self . _read_task = functools . partial ( _recv , self , bufsize , * args ) <EOL> self . _read_coro = AsynCoro . cur_coro ( ) <EOL> self . _read_coro . _await_ ( ) <EOL> self . _notifier . add ( self , _AsyncPoller . _Read ) <EOL> if self . _certfile and self . _rsock . pending ( ) : <EOL> try : <EOL> buf = self . _rsock . recv ( bufsize ) <EOL> except socket . error as err : <EOL> if err . args [ <NUM_LIT:0> ] != EWOULDBLOCK : <EOL> self . _read_task = None <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> if buf : <EOL> self . _read_task = None <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . _proceed_ ( buf ) <EOL> def _async_recvall ( self , bufsize , * args ) : <EOL> """<STR_LIT>""" <EOL> def _recvall ( self , view , * args ) : <EOL> try : <EOL> recvd = self . _rsock . recv_into ( view , len ( view ) , * args ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ : <EOL> pass <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> if recvd : <EOL> view = view [ recvd : ] <EOL> if len ( view ) == <NUM_LIT:0> : <EOL> buf = str ( self . _read_result ) <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . _proceed_ ( buf ) <EOL> else : <EOL> if self . _timeout : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _notifier . _add_timeout ( self ) <EOL> self . _read_task = functools . partial ( _recvall , self , view , * args ) <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . _proceed_ ( '<STR_LIT>' ) <EOL> self . _read_result = bytearray ( bufsize ) <EOL> view = memoryview ( self . _read_result ) <EOL> self . _read_task = functools . partial ( _recvall , self , view , * args ) <EOL> self . _read_coro = AsynCoro . cur_coro ( ) <EOL> self . _read_coro . _await_ ( ) <EOL> self . _notifier . add ( self , _AsyncPoller . _Read ) <EOL> if self . _certfile and self . _rsock . pending ( ) : <EOL> try : <EOL> recvd = self . _rsock . recv_into ( view , bufsize ) <EOL> except socket . error as err : <EOL> if err . args [ <NUM_LIT:0> ] != EWOULDBLOCK : <EOL> self . _read_task = self . _read_result = None <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> if recvd == bufsize : <EOL> buf = str ( self . _read_result ) <EOL> self . _read_task = self . _read_result = None <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . _proceed_ ( buf ) <EOL> elif recvd : <EOL> view = view [ recvd : ] <EOL> self . _read_task = functools . partial ( _recvall , self , view , * args ) <EOL> def _sync_recvall ( self , bufsize , * args ) : <EOL> """<STR_LIT>""" <EOL> self . _read_result = bytearray ( bufsize ) <EOL> view = memoryview ( self . _read_result ) <EOL> while len ( view ) > <NUM_LIT:0> : <EOL> recvd = self . _rsock . recv_into ( view , * args ) <EOL> if not recvd : <EOL> self . _read_result = None <EOL> return '<STR_LIT>' <EOL> view = view [ recvd : ] <EOL> buf , self . _read_result = str ( self . _read_result ) , None <EOL> return buf <EOL> def _async_recvfrom ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> def _recvfrom ( self , * args ) : <EOL> try : <EOL> res = self . _rsock . recvfrom ( * args ) <EOL> except : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> coro . _proceed_ ( res ) <EOL> self . _read_task = functools . partial ( _recvfrom , self , * args ) <EOL> self . _read_coro = AsynCoro . cur_coro ( ) <EOL> self . _read_coro . _await_ ( ) <EOL> self . _notifier . add ( self , _AsyncPoller . _Read ) <EOL> def _async_send ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> def _send ( self , * args ) : <EOL> try : <EOL> sent = self . _rsock . send ( * args ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE : <EOL> pass <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . _proceed_ ( sent ) <EOL> self . _write_task = functools . partial ( _send , self , * args ) <EOL> self . _write_coro = AsynCoro . cur_coro ( ) <EOL> self . _write_coro . _await_ ( ) <EOL> self . _notifier . add ( self , _AsyncPoller . _Write ) <EOL> def _async_sendto ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> def _sendto ( self , * args ) : <EOL> try : <EOL> sent = self . _rsock . sendto ( * args ) <EOL> except : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . _proceed_ ( sent ) <EOL> self . _write_task = functools . partial ( _sendto , self , * args ) <EOL> self . _write_coro = AsynCoro . cur_coro ( ) <EOL> self . _write_coro . _await_ ( ) <EOL> self . _notifier . add ( self , _AsyncPoller . _Write ) <EOL> def _async_sendall ( self , data ) : <EOL> """<STR_LIT>""" <EOL> def _sendall ( self , data_len ) : <EOL> try : <EOL> sent = self . _rsock . send ( self . _write_result ) <EOL> if sent < <NUM_LIT:0> : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = self . _write_result = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE : <EOL> pass <EOL> else : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = self . _write_result = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = self . _write_result = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> if sent > <NUM_LIT:0> : <EOL> self . _write_result = self . _write_result [ sent : ] <EOL> if len ( self . _write_result ) == <NUM_LIT:0> : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = self . _write_result = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . _proceed_ ( None ) <EOL> elif self . _timeout : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _notifier . _add_timeout ( self ) <EOL> self . _write_result = buffer ( data ) <EOL> self . _write_task = functools . partial ( _sendall , self , len ( data ) ) <EOL> self . _write_coro = AsynCoro . cur_coro ( ) <EOL> self . _write_coro . _await_ ( ) <EOL> self . _notifier . add ( self , _AsyncPoller . _Write ) <EOL> def _sync_sendall ( self , data ) : <EOL> """<STR_LIT>""" <EOL> buf = buffer ( data ) <EOL> while len ( buf ) > <NUM_LIT:0> : <EOL> sent = self . _rsock . send ( buf ) <EOL> if sent < <NUM_LIT:0> : <EOL> raise socket . error ( '<STR_LIT>' ) <EOL> buf = buf [ sent : ] <EOL> return None <EOL> def _async_accept ( self ) : <EOL> """<STR_LIT>""" <EOL> def _accept ( self ) : <EOL> conn , addr = self . _rsock . accept ( ) <EOL> self . _read_task = None <EOL> self . _notifier . unregister ( self ) <EOL> if self . _certfile : <EOL> if not self . ssl_server_ctx and hasattr ( ssl , '<STR_LIT>' ) : <EOL> self . ssl_server_ctx = ssl . create_default_context ( ssl . Purpose . CLIENT_AUTH ) <EOL> self . ssl_server_ctx . load_cert_chain ( certfile = self . _certfile , <EOL> keyfile = self . _keyfile ) <EOL> def _ssl_handshake ( conn , addr ) : <EOL> try : <EOL> conn . _rsock . do_handshake ( ) <EOL> except ssl . SSLError as err : <EOL> if ( err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ or <EOL> err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE ) : <EOL> pass <EOL> else : <EOL> conn . _read_task = conn . _write_task = None <EOL> coro , conn . _read_coro = conn . _read_coro , None <EOL> conn . _write_coro = None <EOL> conn . close ( ) <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except : <EOL> conn . _read_task = conn . _write_task = None <EOL> coro , conn . _read_coro = conn . _read_coro , None <EOL> conn . _write_coro = None <EOL> conn . close ( ) <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> conn . _read_task = conn . _write_task = None <EOL> coro , conn . _read_coro = conn . _read_coro , None <EOL> conn . _notifier . clear ( conn ) <EOL> coro . _proceed_ ( ( conn , addr ) ) <EOL> conn = AsyncSocket ( conn , blocking = False , keyfile = self . _keyfile , <EOL> certfile = self . _certfile , ssl_version = self . _ssl_version ) <EOL> try : <EOL> if self . ssl_server_ctx : <EOL> conn . _rsock = self . ssl_server_ctx . wrap_socket ( conn . _rsock , <EOL> server_side = True , <EOL> do_handshake_on_connect = False ) <EOL> else : <EOL> conn . _rsock = ssl . wrap_socket ( conn . _rsock , certfile = self . _certfile , <EOL> keyfile = self . _keyfile , <EOL> ssl_version = self . _ssl_version , <EOL> server_side = True , <EOL> do_handshake_on_connect = False ) <EOL> except : <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> conn . close ( ) <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> conn . _read_task = conn . _write_task = functools . partial ( _ssl_handshake , conn , addr ) <EOL> conn . _read_coro = conn . _write_coro = self . _read_coro <EOL> self . _read_coro = None <EOL> conn . _notifier . add ( conn , _AsyncPoller . _Read | _AsyncPoller . _Write ) <EOL> conn . _read_task ( ) <EOL> else : <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> conn = AsyncSocket ( conn , blocking = False ) <EOL> coro . _proceed_ ( ( conn , addr ) ) <EOL> self . _read_task = functools . partial ( _accept , self ) <EOL> self . _read_coro = AsynCoro . cur_coro ( ) <EOL> self . _read_coro . _await_ ( ) <EOL> self . _notifier . add ( self , _AsyncPoller . _Read ) <EOL> def _async_connect ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> def _connect ( self , * args ) : <EOL> err = self . _rsock . getsockopt ( socket . SOL_SOCKET , socket . SO_ERROR ) <EOL> if err : <EOL> self . _notifier . unregister ( self ) <EOL> self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> coro . throw ( socket . error ( err ) ) <EOL> elif self . _certfile : <EOL> def _ssl_handshake ( self ) : <EOL> try : <EOL> self . _rsock . do_handshake ( ) <EOL> except ssl . SSLError as err : <EOL> if ( err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ or <EOL> err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE ) : <EOL> pass <EOL> else : <EOL> self . _read_task = self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> self . _read_coro = None <EOL> self . close ( ) <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except : <EOL> self . _read_task = self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> self . _read_coro = None <EOL> self . close ( ) <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> self . _notifier . clear ( self ) <EOL> self . _read_task = self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> self . _read_coro = None <EOL> coro . _proceed_ ( <NUM_LIT:0> ) <EOL> try : <EOL> self . _rsock = ssl . wrap_socket ( self . _rsock , ca_certs = self . _certfile , <EOL> cert_reqs = ssl . CERT_REQUIRED , server_side = False , <EOL> do_handshake_on_connect = False ) <EOL> except : <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> self . close ( ) <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> else : <EOL> self . _read_task = self . _write_task = functools . partial ( _ssl_handshake , self ) <EOL> self . _read_coro = self . _write_coro <EOL> self . _notifier . add ( self , _AsyncPoller . _Read ) <EOL> self . _write_task ( ) <EOL> else : <EOL> self . _write_task = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> coro . _proceed_ ( <NUM_LIT:0> ) <EOL> self . _write_task = functools . partial ( _connect , self , * args ) <EOL> self . _write_coro = AsynCoro . cur_coro ( ) <EOL> self . _write_coro . _await_ ( ) <EOL> self . _notifier . add ( self , _AsyncPoller . _Write ) <EOL> try : <EOL> self . _rsock . connect ( * args ) <EOL> except socket . error as e : <EOL> if e . args [ <NUM_LIT:0> ] == EINPROGRESS or e . args [ <NUM_LIT:0> ] == EWOULDBLOCK : <EOL> pass <EOL> else : <EOL> raise <EOL> def _async_send_msg ( self , data ) : <EOL> """<STR_LIT>""" <EOL> yield self . sendall ( struct . pack ( '<STR_LIT>' , len ( data ) ) + data ) <EOL> def _sync_send_msg ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return self . _sync_sendall ( struct . pack ( '<STR_LIT>' , len ( data ) ) + data ) <EOL> def _async_recv_msg ( self ) : <EOL> """<STR_LIT>""" <EOL> n = struct . calcsize ( '<STR_LIT>' ) <EOL> try : <EOL> data = yield self . recvall ( n ) <EOL> except socket . error as err : <EOL> if err . args [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> raise StopIteration ( '<STR_LIT>' ) <EOL> else : <EOL> raise <EOL> if len ( data ) != n : <EOL> raise StopIteration ( '<STR_LIT>' ) <EOL> n = struct . unpack ( '<STR_LIT>' , data ) [ <NUM_LIT:0> ] <EOL> assert n >= <NUM_LIT:0> <EOL> try : <EOL> data = yield self . recvall ( n ) <EOL> except socket . error as err : <EOL> if err . args [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> raise StopIteration ( '<STR_LIT>' ) <EOL> else : <EOL> raise <EOL> if len ( data ) != n : <EOL> raise StopIteration ( '<STR_LIT>' ) <EOL> raise StopIteration ( data ) <EOL> def _sync_recv_msg ( self ) : <EOL> """<STR_LIT>""" <EOL> n = struct . calcsize ( '<STR_LIT>' ) <EOL> try : <EOL> data = self . _sync_recvall ( n ) <EOL> except socket . error as err : <EOL> if err . args [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> raise <EOL> if len ( data ) != n : <EOL> return '<STR_LIT>' <EOL> n = struct . unpack ( '<STR_LIT>' , data ) [ <NUM_LIT:0> ] <EOL> assert n >= <NUM_LIT:0> <EOL> try : <EOL> data = self . _sync_recvall ( n ) <EOL> except socket . error as err : <EOL> if err . args [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> raise <EOL> if len ( data ) != n : <EOL> return '<STR_LIT>' <EOL> return data <EOL> def create_connection ( self , host_port , timeout = None , source_address = None ) : <EOL> if timeout is not None : <EOL> self . settimeout ( timeout ) <EOL> if source_address is not None : <EOL> self . _rsock . bind ( source_address ) <EOL> yield self . connect ( host_port ) <EOL> if platform . system ( ) == '<STR_LIT>' : <EOL> try : <EOL> import win32file <EOL> import win32event <EOL> import pywintypes <EOL> import winerror <EOL> except : <EOL> logger . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> class _AsyncPoller ( object ) : <EOL> """<STR_LIT>""" <EOL> __metaclass__ = MetaSingleton <EOL> __instance = None <EOL> _Read = <NUM_LIT> <EOL> _Write = <NUM_LIT> <EOL> _Error = <NUM_LIT> <EOL> @ classmethod <EOL> def instance ( cls ) : <EOL> return cls . __instance <EOL> def __init__ ( self , iocp_notifier ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . __class__ . __instance = self <EOL> self . _fds = { } <EOL> self . _events = { } <EOL> self . _lock = threading . Lock ( ) <EOL> self . polling = False <EOL> self . _terminate = False <EOL> self . rset = set ( ) <EOL> self . wset = set ( ) <EOL> self . xset = set ( ) <EOL> self . iocp_notifier = iocp_notifier <EOL> self . cmd_rsock , self . cmd_wsock = _AsyncPoller . _socketpair ( ) <EOL> self . cmd_rsock . setblocking ( <NUM_LIT:0> ) <EOL> self . cmd_wsock . setblocking ( <NUM_LIT:0> ) <EOL> self . poller = select . select <EOL> self . poll_thread = threading . Thread ( target = self . poll ) <EOL> self . poll_thread . daemon = True <EOL> self . poll_thread . start ( ) <EOL> def unregister ( self , fd , update = True ) : <EOL> fid = fd . _fileno <EOL> if fd . _timeout : <EOL> self . iocp_notifier . _del_timeout ( fd ) <EOL> self . _lock . acquire ( ) <EOL> if update : <EOL> if self . _fds . pop ( fid , None ) != fd : <EOL> self . _lock . release ( ) <EOL> logger . debug ( '<STR_LIT>' , fid ) <EOL> return <EOL> event = self . _events . pop ( fid , <NUM_LIT:0> ) <EOL> else : <EOL> event = self . _events . get ( fd , <NUM_LIT:0> ) <EOL> self . _lock . release ( ) <EOL> if event & _AsyncPoller . _Read : <EOL> self . rset . discard ( fid ) <EOL> if event & _AsyncPoller . _Write : <EOL> self . wset . discard ( fid ) <EOL> if event & _AsyncPoller . _Error : <EOL> self . xset . discard ( fid ) <EOL> if update and self . polling : <EOL> self . cmd_wsock . send ( '<STR_LIT:u>' ) <EOL> def add ( self , fd , event ) : <EOL> fid = fd . _fileno <EOL> if fd . _timeout : <EOL> self . iocp_notifier . _del_timeout ( fd ) <EOL> cur_event = self . _events . get ( fid , <NUM_LIT:0> ) <EOL> if cur_event & _AsyncPoller . _Read : <EOL> self . rset . discard ( fid ) <EOL> if cur_event & _AsyncPoller . _Write : <EOL> self . wset . discard ( fid ) <EOL> if cur_event & _AsyncPoller . _Error : <EOL> self . xset . discard ( fid ) <EOL> event |= cur_event <EOL> self . _events [ fid ] = event <EOL> self . _fds [ fid ] = fd <EOL> if event : <EOL> if event & _AsyncPoller . _Read : <EOL> self . rset . add ( fid ) <EOL> if event & _AsyncPoller . _Write : <EOL> self . wset . add ( fid ) <EOL> if event & _AsyncPoller . _Error : <EOL> self . xset . add ( fid ) <EOL> if fd . _timeout : <EOL> self . iocp_notifier . _add_timeout ( fd ) <EOL> self . iocp_notifier . interrupt ( fd . _timeout ) <EOL> if self . polling : <EOL> self . cmd_wsock . send ( '<STR_LIT:m>' ) <EOL> def clear ( self , fd , event = <NUM_LIT:0> ) : <EOL> fid = fd . _fileno <EOL> cur_event = self . _events . get ( fid , <NUM_LIT:0> ) <EOL> if cur_event : <EOL> if cur_event & _AsyncPoller . _Read : <EOL> self . rset . discard ( fid ) <EOL> if cur_event & _AsyncPoller . _Write : <EOL> self . wset . discard ( fid ) <EOL> if cur_event & _AsyncPoller . _Error : <EOL> self . xset . discard ( fid ) <EOL> if event : <EOL> cur_event &= ~ event <EOL> else : <EOL> cur_event = <NUM_LIT:0> <EOL> self . _events [ fid ] = cur_event <EOL> if cur_event : <EOL> if cur_event & _AsyncPoller . _Read : <EOL> self . rset . add ( fid ) <EOL> if cur_event & _AsyncPoller . _Write : <EOL> self . wset . add ( fid ) <EOL> if cur_event & _AsyncPoller . _Error : <EOL> self . xset . add ( fid ) <EOL> elif fd . _timeout_id : <EOL> self . iocp_notifier . _del_timeout ( fd ) <EOL> if self . polling : <EOL> self . cmd_wsock . send ( '<STR_LIT:m>' ) <EOL> def poll ( self ) : <EOL> self . cmd_rsock = AsyncSocket ( self . cmd_rsock ) <EOL> setattr ( self . cmd_rsock , '<STR_LIT>' , lambda : self . cmd_rsock . _rsock . recv ( <NUM_LIT> ) ) <EOL> self . add ( self . cmd_rsock , _AsyncPoller . _Read ) <EOL> while <NUM_LIT:1> : <EOL> self . polling = True <EOL> rlist , wlist , xlist = self . poller ( self . rset , self . wset , self . xset ) <EOL> self . polling = False <EOL> if self . _terminate : <EOL> break <EOL> events = { } <EOL> for fid in rlist : <EOL> events [ fid ] = _AsyncPoller . _Read <EOL> for fid in wlist : <EOL> events [ fid ] = events . get ( fid , <NUM_LIT:0> ) | _AsyncPoller . _Write <EOL> for fid in xlist : <EOL> events [ fid ] = events . get ( fid , <NUM_LIT:0> ) | _AsyncPoller . _Error <EOL> self . _lock . acquire ( ) <EOL> events = [ ( self . _fds . get ( fid , None ) , event ) <EOL> for ( fid , event ) in events . iteritems ( ) ] <EOL> self . _lock . release ( ) <EOL> iocp_notify = False <EOL> for fd , event in events : <EOL> if fd is None : <EOL> logger . debug ( '<STR_LIT>' ) <EOL> continue <EOL> if event & _AsyncPoller . _Read : <EOL> if fd . _read_task : <EOL> if fd != self . cmd_rsock : <EOL> iocp_notify = True <EOL> fd . _read_task ( ) <EOL> else : <EOL> logger . debug ( '<STR_LIT>' , fd . _fileno ) <EOL> if event & _AsyncPoller . _Write : <EOL> if fd . _write_task : <EOL> iocp_notify = True <EOL> fd . _write_task ( ) <EOL> else : <EOL> logger . debug ( '<STR_LIT>' , fd . _fileno ) <EOL> if event & _AsyncPoller . _Error : <EOL> if fd . _read_coro : <EOL> fd . _read_coro . throw ( socket . error ( _AsyncPoller . _Error ) ) <EOL> if fd . _write_coro : <EOL> fd . _write_coro . throw ( socket . error ( _AsyncPoller . _Error ) ) <EOL> if iocp_notify : <EOL> self . iocp_notifier . interrupt ( ) <EOL> self . rset = set ( ) <EOL> self . wset = set ( ) <EOL> self . xset = set ( ) <EOL> self . cmd_rsock . close ( ) <EOL> self . cmd_wsock . close ( ) <EOL> def terminate ( self ) : <EOL> self . _terminate = True <EOL> self . cmd_wsock . send ( '<STR_LIT:x>' ) <EOL> self . poll_thread . join ( ) <EOL> self . __class__ . __instance = None <EOL> @ staticmethod <EOL> def _socketpair ( ) : <EOL> srv_sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> srv_sock . bind ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) ) <EOL> srv_sock . listen ( <NUM_LIT:1> ) <EOL> sock1 = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> conn_thread = threading . Thread ( target = lambda sock , addr_port : sock . connect ( addr_port ) , <EOL> args = ( sock1 , srv_sock . getsockname ( ) ) ) <EOL> conn_thread . daemon = True <EOL> conn_thread . start ( ) <EOL> sock2 , caddr = srv_sock . accept ( ) <EOL> srv_sock . close ( ) <EOL> return ( sock1 , sock2 ) <EOL> class _AsyncNotifier ( object ) : <EOL> """<STR_LIT>""" <EOL> __metaclass__ = MetaSingleton <EOL> __instance = None <EOL> _Block = win32event . INFINITE <EOL> @ classmethod <EOL> def instance ( cls , * args , ** kwargs ) : <EOL> if cls . __instance is None : <EOL> cls . __instance = cls ( * args , ** kwargs ) <EOL> return cls . __instance <EOL> def __init__ ( self ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . __class__ . __instance = self <EOL> self . iocp = win32file . CreateIoCompletionPort ( win32file . INVALID_HANDLE_VALUE , <EOL> None , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> self . _timeouts = [ ] <EOL> self . poll_timeout = <NUM_LIT:0> <EOL> self . _lock = threading . Lock ( ) <EOL> self . async_poller = _AsyncPoller ( self ) <EOL> self . cmd_rsock , self . cmd_wsock = _AsyncPoller . _socketpair ( ) <EOL> self . cmd_wsock . setblocking ( <NUM_LIT:0> ) <EOL> self . cmd_rsock = AsyncSocket ( self . cmd_rsock ) <EOL> self . cmd_rsock_buf = win32file . AllocateReadBuffer ( <NUM_LIT> ) <EOL> self . cmd_rsock . _read_overlap . object = self . cmd_rsock_recv <EOL> err , n = win32file . WSARecv ( self . cmd_rsock . _fileno , self . cmd_rsock_buf , <EOL> self . cmd_rsock . _read_overlap , <NUM_LIT:0> ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> logger . warning ( '<STR_LIT>' , err ) <EOL> self . _poller_name = '<STR_LIT>' <EOL> def cmd_rsock_recv ( self , err , n ) : <EOL> if n == <NUM_LIT:0> : <EOL> err = winerror . ERROR_CONNECTION_INVALID <EOL> if err : <EOL> logger . warning ( '<STR_LIT>' , err ) <EOL> err , n = win32file . WSARecv ( self . cmd_rsock . _fileno , self . cmd_rsock_buf , <EOL> self . cmd_rsock . _read_overlap , <NUM_LIT:0> ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> logger . warning ( '<STR_LIT>' , err ) <EOL> def interrupt ( self , timeout = None ) : <EOL> if timeout is None : <EOL> self . cmd_wsock . send ( '<STR_LIT:i>' ) <EOL> elif self . poll_timeout == _AsyncNotifier . _Block or timeout < self . poll_timeout : <EOL> self . cmd_wsock . send ( '<STR_LIT:I>' ) <EOL> def register ( self , handle , event = <NUM_LIT:0> ) : <EOL> win32file . CreateIoCompletionPort ( handle , self . iocp , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> def unregister ( self , handle ) : <EOL> pass <EOL> def modify ( self , fd , event ) : <EOL> pass <EOL> def poll ( self , timeout ) : <EOL> self . _lock . acquire ( ) <EOL> if timeout == <NUM_LIT:0> : <EOL> self . poll_timeout = <NUM_LIT:0> <EOL> elif self . _timeouts : <EOL> self . poll_timeout = self . _timeouts [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] - _time ( ) <EOL> if self . poll_timeout < <NUM_LIT> : <EOL> self . poll_timeout = <NUM_LIT:0> <EOL> elif timeout is not None : <EOL> self . poll_timeout = min ( timeout , self . poll_timeout ) <EOL> elif timeout is None : <EOL> self . poll_timeout = _AsyncNotifier . _Block <EOL> else : <EOL> self . poll_timeout = timeout <EOL> timeout = self . poll_timeout <EOL> self . _lock . release ( ) <EOL> if timeout and timeout != _AsyncNotifier . _Block : <EOL> timeout = int ( timeout * <NUM_LIT:1000> ) <EOL> err , n , key , overlap = win32file . GetQueuedCompletionStatus ( self . iocp , timeout ) <EOL> while err != winerror . WAIT_TIMEOUT : <EOL> if overlap and overlap . object : <EOL> overlap . object ( err , n ) <EOL> else : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> err , n , key , overlap = win32file . GetQueuedCompletionStatus ( self . iocp , <NUM_LIT:0> ) <EOL> self . poll_timeout = <NUM_LIT:0> <EOL> if timeout == <NUM_LIT:0> : <EOL> now = _time ( ) <EOL> self . _lock . acquire ( ) <EOL> while self . _timeouts and self . _timeouts [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <= now : <EOL> fd_timeout , fd = self . _timeouts . pop ( <NUM_LIT:0> ) <EOL> if fd . _timeout_id == fd_timeout : <EOL> fd . _timeout_id = None <EOL> fd . _timed_out ( ) <EOL> self . _lock . release ( ) <EOL> def _add_timeout ( self , fd ) : <EOL> if fd . _timeout : <EOL> self . _lock . acquire ( ) <EOL> fd . _timeout_id = _time ( ) + fd . _timeout <EOL> i = bisect_left ( self . _timeouts , ( fd . _timeout_id , fd ) ) <EOL> self . _timeouts . insert ( i , ( fd . _timeout_id , fd ) ) <EOL> self . _lock . release ( ) <EOL> else : <EOL> fd . _timeout_id = None <EOL> def _del_timeout ( self , fd ) : <EOL> if fd . _timeout_id : <EOL> self . _lock . acquire ( ) <EOL> i = bisect_left ( self . _timeouts , ( fd . _timeout_id , fd ) ) <EOL> while i < len ( self . _timeouts ) : <EOL> if self . _timeouts [ i ] == ( fd . _timeout_id , fd ) : <EOL> del self . _timeouts [ i ] <EOL> fd . _timeout_id = None <EOL> break <EOL> if fd . _timeout_id != self . _timeouts [ i ] [ <NUM_LIT:0> ] : <EOL> logger . warning ( '<STR_LIT>' , <EOL> fd . _fileno , fd . _timeout_id ) <EOL> break <EOL> i += <NUM_LIT:1> <EOL> self . _lock . release ( ) <EOL> def terminate ( self ) : <EOL> self . async_poller . terminate ( ) <EOL> self . cmd_rsock . close ( ) <EOL> self . cmd_wsock . close ( ) <EOL> if len ( self . _timeouts ) : <EOL> logger . warning ( '<STR_LIT>' , len ( self . _timeouts ) ) <EOL> win32file . CloseHandle ( self . iocp ) <EOL> self . iocp = None <EOL> self . cmd_rsock_buf = None <EOL> self . __class__ . __instance = None <EOL> class AsyncSocket ( _AsyncSocket ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = _AsyncSocket . __slots__ + ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . _read_overlap = None <EOL> self . _write_overlap = None <EOL> _AsyncSocket . __init__ ( self , * args , ** kwargs ) <EOL> def _register ( self ) : <EOL> if not self . _blocking : <EOL> if self . _rsock . type & socket . SOCK_STREAM : <EOL> self . _read_overlap = pywintypes . OVERLAPPED ( ) <EOL> self . _write_overlap = pywintypes . OVERLAPPED ( ) <EOL> self . _notifier . register ( self . _fileno ) <EOL> else : <EOL> self . _notifier = _AsyncPoller . instance ( ) <EOL> else : <EOL> _AsyncSocket . _register ( self ) <EOL> def _unregister ( self ) : <EOL> if self . _notifier : <EOL> self . _notifier . unregister ( self ) <EOL> if self . _rsock . type & socket . SOCK_STREAM : <EOL> if ( ( self . _read_overlap and self . _read_overlap . object ) or <EOL> ( self . _write_overlap and self . _write_overlap . object ) ) : <EOL> def _cleanup_ ( self , rc , n ) : <EOL> self . _read_overlap . object = self . _write_overlap . object = None <EOL> if rc == winerror . ERROR_OPERATION_ABORTED : <EOL> self . _read_result = self . _write_result = None <EOL> self . _read_coro = self . _write_coro = None <EOL> self . _read_overlap = self . _write_overlap = None <EOL> self . _notifier = None <EOL> elif rc : <EOL> logger . warning ( '<STR_LIT>' , rc ) <EOL> if self . _read_overlap and self . _read_overlap . object : <EOL> self . _read_overlap . object = functools . partial ( _cleanup_ , self ) <EOL> if self . _write_overlap and self . _write_overlap . object : <EOL> self . _read_overlap . object = functools . partial ( _cleanup_ , self ) <EOL> win32file . CancelIo ( self . _fileno ) <EOL> else : <EOL> self . _read_overlap = self . _write_overlap = None <EOL> self . _read_result = self . _write_result = None <EOL> self . _read_coro = self . _write_coro = None <EOL> self . _notifier = None <EOL> else : <EOL> self . _notifier = None <EOL> def _timed_out ( self ) : <EOL> if self . _rsock and self . _rsock . type & socket . SOCK_STREAM : <EOL> if self . _read_overlap or self . _write_overlap : <EOL> win32file . CancelIo ( self . _fileno ) <EOL> if self . _read_coro : <EOL> if self . _rsock and self . _rsock . type & socket . SOCK_DGRAM : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Read ) <EOL> self . _read_task = None <EOL> self . _read_coro . throw ( socket . timeout ( '<STR_LIT>' ) ) <EOL> self . _read_result = self . _read_coro = None <EOL> if self . _write_coro : <EOL> if self . _rsock and self . _rsock . type & socket . SOCK_DGRAM : <EOL> self . _notifier . clear ( self , _AsyncPoller . _Write ) <EOL> self . _write_task = None <EOL> self . _write_coro . throw ( socket . timeout ( '<STR_LIT>' ) ) <EOL> self . _write_result = self . _write_coro = None <EOL> def setblocking ( self , blocking ) : <EOL> _AsyncSocket . setblocking ( self , blocking ) <EOL> if not self . _blocking and self . _rsock . type & socket . SOCK_STREAM : <EOL> self . recv = self . _iocp_recv <EOL> self . send = self . _iocp_send <EOL> self . recvall = self . _iocp_recvall <EOL> self . sendall = self . _iocp_sendall <EOL> self . connect = self . _iocp_connect <EOL> self . accept = self . _iocp_accept <EOL> def _iocp_recv ( self , bufsize , * args ) : <EOL> """<STR_LIT>""" <EOL> def _recv ( self , err , n ) : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> if err or n == <NUM_LIT:0> : <EOL> self . _read_overlap . object = self . _read_result = None <EOL> if err == winerror . ERROR_OPERATION_ABORTED : <EOL> self . _read_coro = None <EOL> else : <EOL> if not err : <EOL> err = winerror . ERROR_CONNECTION_INVALID <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> if err == winerror . ERROR_CONNECTION_INVALID : <EOL> coro . _proceed_ ( '<STR_LIT>' ) <EOL> else : <EOL> coro . throw ( socket . error ( err ) ) <EOL> else : <EOL> buf = self . _read_result [ : n ] <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> coro . _proceed_ ( buf ) <EOL> self . _read_result = win32file . AllocateReadBuffer ( bufsize ) <EOL> self . _read_overlap . object = functools . partial ( _recv , self ) <EOL> self . _read_coro = AsynCoro . cur_coro ( ) <EOL> self . _read_coro . _await_ ( ) <EOL> if self . _timeout : <EOL> self . _notifier . _add_timeout ( self ) <EOL> err , n = win32file . WSARecv ( self . _fileno , self . _read_result , self . _read_overlap , <NUM_LIT:0> ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> self . _read_overlap . object = self . _read_result = self . _read_coro = None <EOL> raise socket . error ( err ) <EOL> def _iocp_send ( self , buf , * args ) : <EOL> """<STR_LIT>""" <EOL> def _send ( self , err , n ) : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> if err or n == <NUM_LIT:0> : <EOL> self . _write_overlap . object = self . _write_result = None <EOL> if err == winerror . ERROR_OPERATION_ABORTED : <EOL> self . _write_coro = None <EOL> else : <EOL> if not err : <EOL> err = winerror . ERROR_CONNECTION_INVALID <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> if coro : <EOL> coro . throw ( socket . error ( err ) ) <EOL> else : <EOL> self . _write_overlap . object = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> if coro : <EOL> coro . _proceed_ ( n ) <EOL> self . _write_overlap . object = functools . partial ( _send , self ) <EOL> self . _write_coro = AsynCoro . cur_coro ( ) <EOL> self . _write_coro . _await_ ( ) <EOL> if self . _timeout : <EOL> self . _notifier . _add_timeout ( self ) <EOL> err , n = win32file . WSASend ( self . _fileno , buf , self . _write_overlap , <NUM_LIT:0> ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> self . _write_overlap . object = self . _write_coro = None <EOL> raise socket . error ( err ) <EOL> def _iocp_recvall ( self , bufsize , * args ) : <EOL> """<STR_LIT>""" <EOL> def _recvall ( self , pending , buf , err , n ) : <EOL> if err or n == <NUM_LIT:0> : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> if err == winerror . ERROR_OPERATION_ABORTED : <EOL> self . _read_coro = None <EOL> else : <EOL> if not err : <EOL> err = winerror . ERROR_CONNECTION_INVALID <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> if err == winerror . ERROR_CONNECTION_INVALID : <EOL> coro . _proceed_ ( '<STR_LIT>' ) <EOL> else : <EOL> coro . throw ( socket . error ( err ) ) <EOL> else : <EOL> self . _read_result . append ( buf [ : n ] ) <EOL> pending -= n <EOL> if pending == <NUM_LIT:0> : <EOL> buf = '<STR_LIT>' . join ( self . _read_result ) <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> coro . _proceed_ ( buf ) <EOL> else : <EOL> buf = win32file . AllocateReadBuffer ( min ( pending , <NUM_LIT> ) ) <EOL> self . _read_overlap . object = functools . partial ( _recvall , self , <EOL> pending , buf ) <EOL> err , n = win32file . WSARecv ( self . _fileno , buf , self . _read_overlap , <NUM_LIT:0> ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> coro . throw ( socket . error ( err ) ) <EOL> self . _read_result = [ ] <EOL> buf = win32file . AllocateReadBuffer ( min ( bufsize , <NUM_LIT> ) ) <EOL> self . _read_overlap . object = functools . partial ( _recvall , self , bufsize , buf ) <EOL> self . _read_coro = AsynCoro . cur_coro ( ) <EOL> self . _read_coro . _await_ ( ) <EOL> if self . _timeout : <EOL> self . _notifier . _add_timeout ( self ) <EOL> err , n = win32file . WSARecv ( self . _fileno , buf , self . _read_overlap , <NUM_LIT:0> ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> self . _read_overlap . object = self . _read_result = self . _read_coro = None <EOL> raise socket . error ( err ) <EOL> def _iocp_sendall ( self , data ) : <EOL> """<STR_LIT>""" <EOL> def _sendall ( self , err , n ) : <EOL> if err or n == <NUM_LIT:0> : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _write_overlap . object = self . _write_result = None <EOL> if err == winerror . ERROR_OPERATION_ABORTED : <EOL> self . _write_coro = None <EOL> else : <EOL> if not err : <EOL> err = winerror . ERROR_CONNECTION_INVALID <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> if coro : <EOL> coro . throw ( socket . error ( err ) ) <EOL> else : <EOL> self . _write_result = self . _write_result [ n : ] <EOL> if len ( self . _write_result ) == <NUM_LIT:0> : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _write_overlap . object = self . _write_result = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> if coro : <EOL> coro . _proceed_ ( <NUM_LIT:0> ) <EOL> else : <EOL> err , n = win32file . WSASend ( self . _fileno , self . _write_result , <EOL> self . _write_overlap , <NUM_LIT:0> ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _write_overlap . object = self . _write_result = None <EOL> coro , self . _write_coro = self . _write_coro , None <EOL> if coro : <EOL> coro . throw ( socket . error ( err ) ) <EOL> self . _write_result = buffer ( data ) <EOL> self . _write_overlap . object = functools . partial ( _sendall , self ) <EOL> self . _write_coro = AsynCoro . cur_coro ( ) <EOL> self . _write_coro . _await_ ( ) <EOL> if self . _timeout : <EOL> self . _notifier . _add_timeout ( self ) <EOL> err , n = win32file . WSASend ( self . _fileno , self . _write_result , self . _write_overlap , <NUM_LIT:0> ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> self . _write_overlap . object = self . _write_result = self . _write_coro = None <EOL> raise socket . error ( err ) <EOL> def _iocp_connect ( self , host_port ) : <EOL> """<STR_LIT>""" <EOL> def _connect ( self , err , n ) : <EOL> def _ssl_handshake ( self , err , n ) : <EOL> try : <EOL> self . _rsock . do_handshake ( ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ : <EOL> err , n = win32file . WSARecv ( self . _fileno , self . _read_result , <EOL> self . _read_overlap , <NUM_LIT:0> ) <EOL> elif err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE : <EOL> err , n = win32file . WSASend ( self . _fileno , '<STR_LIT>' , self . _read_overlap , <NUM_LIT:0> ) <EOL> else : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro = self . _read_coro <EOL> self . close ( ) <EOL> if coro : <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro = self . _read_coro <EOL> self . close ( ) <EOL> if err != winerror . ERROR_OPERATION_ABORTED and coro : <EOL> coro . throw ( socket . error ( err ) ) <EOL> else : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> coro . _proceed_ ( <NUM_LIT:0> ) <EOL> if err : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> if err == winerror . ERROR_OPERATION_ABORTED : <EOL> self . _read_coro = None <EOL> else : <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> coro . throw ( socket . error ( err ) ) <EOL> else : <EOL> self . _rsock . setsockopt ( socket . SOL_SOCKET , <EOL> win32file . SO_UPDATE_CONNECT_CONTEXT , '<STR_LIT>' ) <EOL> if self . _certfile : <EOL> self . _rsock = ssl . wrap_socket ( self . _rsock , ca_certs = self . _certfile , <EOL> cert_reqs = ssl . CERT_REQUIRED , <EOL> server_side = False , <EOL> do_handshake_on_connect = False ) <EOL> self . _read_result = win32file . AllocateReadBuffer ( <NUM_LIT:0> ) <EOL> self . _read_overlap . object = functools . partial ( _ssl_handshake , self ) <EOL> self . _read_overlap . object ( None , <NUM_LIT:0> ) <EOL> else : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> coro . _proceed_ ( <NUM_LIT:0> ) <EOL> try : <EOL> self . _rsock . bind ( ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> except socket . error as exc : <EOL> if exc [ <NUM_LIT:0> ] != EINVAL : <EOL> raise <EOL> self . _read_overlap . object = functools . partial ( _connect , self ) <EOL> self . _read_coro = AsynCoro . cur_coro ( ) <EOL> self . _read_coro . _await_ ( ) <EOL> if self . _timeout : <EOL> self . _notifier . _add_timeout ( self ) <EOL> err , n = win32file . ConnectEx ( self . _rsock , host_port , self . _read_overlap ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> self . _read_overlap . object = self . _read_result = self . _read_coro = None <EOL> raise socket . error ( err ) <EOL> def _iocp_accept ( self ) : <EOL> """<STR_LIT>""" <EOL> def _accept ( self , conn , err , n ) : <EOL> def _ssl_handshake ( self , conn , addr , err , n ) : <EOL> try : <EOL> conn . _rsock . do_handshake ( ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_READ : <EOL> err , n = win32file . WSARecv ( conn . _fileno , self . _read_result , <EOL> self . _read_overlap , <NUM_LIT:0> ) <EOL> elif err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_WANT_WRITE : <EOL> err , n = win32file . WSASend ( conn . _fileno , '<STR_LIT>' , self . _read_overlap , <NUM_LIT:0> ) <EOL> else : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> conn . close ( ) <EOL> if coro : <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> except : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> conn . close ( ) <EOL> if err != winerror . ERROR_OPERATION_ABORTED and coro : <EOL> coro . throw ( socket . error ( err ) ) <EOL> else : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> coro . _proceed_ ( ( conn , addr ) ) <EOL> if err : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if err != winerror . ERROR_OPERATION_ABORTED and coro : <EOL> coro . throw ( socket . error ( err ) ) <EOL> else : <EOL> family , laddr , raddr = win32file . GetAcceptExSockaddrs ( conn , self . _read_result ) <EOL> conn . _rsock . setsockopt ( socket . SOL_SOCKET , win32file . SO_UPDATE_ACCEPT_CONTEXT , <EOL> struct . pack ( '<STR_LIT:P>' , self . _fileno ) ) <EOL> if self . _certfile : <EOL> if not self . ssl_server_ctx and hasattr ( ssl , '<STR_LIT>' ) : <EOL> self . ssl_server_ctx = ssl . create_default_context ( ssl . Purpose . CLIENT_AUTH ) <EOL> self . ssl_server_ctx . load_cert_chain ( certfile = self . _certfile , <EOL> keyfile = self . _keyfile ) <EOL> if self . ssl_server_ctx : <EOL> conn . _rsock = self . ssl_server_ctx . wrap_socket ( conn . _rsock , <EOL> server_side = True , <EOL> do_handshake_on_connect = False ) <EOL> else : <EOL> conn . _rsock = ssl . wrap_socket ( conn . _rsock , certfile = self . _certfile , <EOL> keyfile = self . _keyfile , <EOL> server_side = True , <EOL> ssl_version = self . _ssl_version , <EOL> do_handshake_on_connect = False ) <EOL> self . _read_result = win32file . AllocateReadBuffer ( <NUM_LIT:0> ) <EOL> self . _read_overlap . object = functools . partial ( _ssl_handshake , self , <EOL> conn , raddr ) <EOL> self . _read_overlap . object ( None , <NUM_LIT:0> ) <EOL> else : <EOL> if self . _timeout and self . _notifier : <EOL> self . _notifier . _del_timeout ( self ) <EOL> self . _read_overlap . object = self . _read_result = None <EOL> coro , self . _read_coro = self . _read_coro , None <EOL> if coro : <EOL> coro . _proceed_ ( ( conn , raddr ) ) <EOL> sock = socket . socket ( self . _rsock . family , self . _rsock . type , self . _rsock . proto ) <EOL> conn = AsyncSocket ( sock , keyfile = self . _keyfile , certfile = self . _certfile , <EOL> ssl_version = self . _ssl_version ) <EOL> self . _read_result = win32file . AllocateReadBuffer ( win32file . CalculateSocketEndPointSize ( sock ) ) <EOL> self . _read_overlap . object = functools . partial ( _accept , self , conn ) <EOL> self . _read_coro = AsynCoro . cur_coro ( ) <EOL> self . _read_coro . _await_ ( ) <EOL> if self . _timeout : <EOL> self . _notifier . _add_timeout ( self ) <EOL> err = win32file . AcceptEx ( self . _fileno , conn . _fileno , self . _read_result , <EOL> self . _read_overlap ) <EOL> if err and err != winerror . ERROR_IO_PENDING : <EOL> self . _read_overlap . object = self . _read_result = self . _read_coro = None <EOL> raise socket . error ( err ) <EOL> if not isinstance ( getattr ( sys . modules [ __name__ ] , '<STR_LIT>' , None ) , MetaSingleton ) : <EOL> class _AsyncPoller ( object ) : <EOL> """<STR_LIT>""" <EOL> __metaclass__ = MetaSingleton <EOL> __instance = None <EOL> _Read = None <EOL> _Write = None <EOL> _Hangup = None <EOL> _Error = None <EOL> _Block = None <EOL> @ classmethod <EOL> def instance ( cls , * args , ** kwargs ) : <EOL> if cls . __instance is None : <EOL> cls . __instance = cls ( * args , ** kwargs ) <EOL> return cls . __instance <EOL> def __init__ ( self ) : <EOL> if self . __class__ . __instance is None : <EOL> self . __class__ . __instance = self <EOL> self . timeout_multiplier = <NUM_LIT:1> <EOL> if hasattr ( select , '<STR_LIT>' ) : <EOL> self . _poller_name = '<STR_LIT>' <EOL> self . _poller = select . epoll ( ) <EOL> _AsyncPoller . _Read = select . EPOLLIN | select . EPOLLPRI <EOL> _AsyncPoller . _Write = select . EPOLLOUT <EOL> _AsyncPoller . _Hangup = select . EPOLLHUP <EOL> _AsyncPoller . _Error = select . EPOLLERR <EOL> _AsyncPoller . _Block = - <NUM_LIT:1> <EOL> elif hasattr ( select , '<STR_LIT>' ) : <EOL> self . _poller_name = '<STR_LIT>' <EOL> self . _poller = _KQueueNotifier ( ) <EOL> _AsyncPoller . _Read = <NUM_LIT> <EOL> _AsyncPoller . _Write = <NUM_LIT> <EOL> _AsyncPoller . _Hangup = <NUM_LIT> <EOL> _AsyncPoller . _Error = <NUM_LIT> <EOL> _AsyncPoller . _Block = None <EOL> elif hasattr ( select , '<STR_LIT>' ) : <EOL> self . _poller_name = '<STR_LIT>' <EOL> self . _poller = select . devpoll ( ) <EOL> _AsyncPoller . _Read = select . POLLIN | select . POLLPRI <EOL> _AsyncPoller . _Write = select . POLLOUT <EOL> _AsyncPoller . _Hangup = select . POLLHUP <EOL> _AsyncPoller . _Error = select . POLLERR <EOL> _AsyncPoller . _Block = - <NUM_LIT:1> <EOL> self . timeout_multiplier = <NUM_LIT:1000> <EOL> elif hasattr ( select , '<STR_LIT>' ) : <EOL> self . _poller_name = '<STR_LIT>' <EOL> self . _poller = select . poll ( ) <EOL> _AsyncPoller . _Read = select . POLLIN | select . POLLPRI <EOL> _AsyncPoller . _Write = select . POLLOUT <EOL> _AsyncPoller . _Hangup = select . POLLHUP <EOL> _AsyncPoller . _Error = select . POLLERR <EOL> _AsyncPoller . _Block = - <NUM_LIT:1> <EOL> self . timeout_multiplier = <NUM_LIT:1000> <EOL> else : <EOL> self . _poller_name = '<STR_LIT>' <EOL> self . _poller = _SelectNotifier ( ) <EOL> _AsyncPoller . _Read = <NUM_LIT> <EOL> _AsyncPoller . _Write = <NUM_LIT> <EOL> _AsyncPoller . _Hangup = <NUM_LIT> <EOL> _AsyncPoller . _Error = <NUM_LIT> <EOL> _AsyncPoller . _Block = None <EOL> self . _fds = { } <EOL> self . _events = { } <EOL> self . _timeouts = [ ] <EOL> self . cmd_rsock , self . cmd_wsock = _AsyncPoller . _socketpair ( ) <EOL> self . cmd_wsock . setblocking ( <NUM_LIT:0> ) <EOL> self . cmd_rsock = AsyncSocket ( self . cmd_rsock ) <EOL> setattr ( self . cmd_rsock , '<STR_LIT>' , lambda : self . cmd_rsock . _rsock . recv ( <NUM_LIT> ) ) <EOL> self . add ( self . cmd_rsock , _AsyncPoller . _Read ) <EOL> def interrupt ( self ) : <EOL> self . cmd_wsock . send ( '<STR_LIT:I>' ) <EOL> def poll ( self , timeout ) : <EOL> """<STR_LIT>""" <EOL> if timeout == <NUM_LIT:0> : <EOL> poll_timeout = timeout <EOL> elif self . _timeouts : <EOL> poll_timeout = self . _timeouts [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] - _time ( ) <EOL> if timeout is not None : <EOL> poll_timeout = min ( timeout , poll_timeout ) <EOL> poll_timeout *= self . timeout_multiplier <EOL> elif timeout is None : <EOL> poll_timeout = _AsyncPoller . _Block <EOL> else : <EOL> poll_timeout = timeout * self . timeout_multiplier <EOL> try : <EOL> events = self . _poller . poll ( poll_timeout ) <EOL> except : <EOL> logger . debug ( '<STR_LIT>' ) <EOL> logger . debug ( traceback . format_exc ( ) ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> return <EOL> try : <EOL> for fileno , event in events : <EOL> fd = self . _fds . get ( fileno , None ) <EOL> if fd is None : <EOL> if event != _AsyncPoller . _Hangup : <EOL> logger . debug ( '<STR_LIT>' , event ) <EOL> continue <EOL> if event & _AsyncPoller . _Read : <EOL> if fd . _read_task is None : <EOL> logger . debug ( '<STR_LIT>' , <EOL> fd . _fileno ) <EOL> self . unregister ( fd ) <EOL> else : <EOL> fd . _read_task ( ) <EOL> elif event & _AsyncPoller . _Write : <EOL> if fd . _write_task is None : <EOL> logger . debug ( '<STR_LIT>' , <EOL> fd . _fileno ) <EOL> self . unregister ( fd ) <EOL> else : <EOL> fd . _write_task ( ) <EOL> elif event & _AsyncPoller . _Hangup : <EOL> fd . _eof ( ) <EOL> elif event & _AsyncPoller . _Error : <EOL> logger . warning ( '<STR_LIT>' , fd . _fileno ) <EOL> self . unregister ( fd ) <EOL> except : <EOL> logger . debug ( traceback . format_exc ( ) ) <EOL> if self . _timeouts : <EOL> now = _time ( ) <EOL> while self . _timeouts and self . _timeouts [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <= now : <EOL> fd_timeout , fd = self . _timeouts . pop ( <NUM_LIT:0> ) <EOL> if fd . _timeout_id == fd_timeout : <EOL> fd . _timeout_id = None <EOL> fd . _timed_out ( ) <EOL> def terminate ( self ) : <EOL> self . cmd_wsock . close ( ) <EOL> self . cmd_rsock . close ( ) <EOL> if len ( self . _timeouts ) : <EOL> logger . warning ( '<STR_LIT>' , len ( self . _timeouts ) ) <EOL> if hasattr ( self . _poller , '<STR_LIT>' ) : <EOL> self . _poller . terminate ( ) <EOL> for fd in self . _fds . itervalues ( ) : <EOL> try : <EOL> self . _poller . unregister ( fd . _fileno ) <EOL> except : <EOL> logger . warning ( '<STR_LIT>' , <EOL> fd . _fileno , traceback . format_exc ( ) ) <EOL> fd . _notifier = None <EOL> self . _poller = None <EOL> self . _timeouts = [ ] <EOL> self . _fds = { } <EOL> self . __class__ . __instance = None <EOL> def _add_timeout ( self , fd ) : <EOL> if fd . _timeout : <EOL> fd . _timeout_id = _time ( ) + fd . _timeout <EOL> i = bisect_left ( self . _timeouts , ( fd . _timeout_id , fd ) ) <EOL> self . _timeouts . insert ( i , ( fd . _timeout_id , fd ) ) <EOL> else : <EOL> fd . _timeout_id = None <EOL> def _del_timeout ( self , fd ) : <EOL> if fd . _timeout_id : <EOL> i = bisect_left ( self . _timeouts , ( fd . _timeout_id , fd ) ) <EOL> while i < len ( self . _timeouts ) : <EOL> if self . _timeouts [ i ] == ( fd . _timeout_id , fd ) : <EOL> del self . _timeouts [ i ] <EOL> fd . _timeout_id = None <EOL> break <EOL> if fd . _timeout_id != self . _timeouts [ i ] [ <NUM_LIT:0> ] : <EOL> logger . warning ( '<STR_LIT>' , fd . _fileno , fd . _timeout_id ) <EOL> break <EOL> i += <NUM_LIT:1> <EOL> def unregister ( self , fd ) : <EOL> if self . _fds . pop ( fd . _fileno , None ) is None : <EOL> return <EOL> self . _events . pop ( fd . _fileno , None ) <EOL> self . _poller . unregister ( fd . _fileno ) <EOL> self . _del_timeout ( fd ) <EOL> def add ( self , fd , event ) : <EOL> cur_event = self . _events . get ( fd . _fileno , None ) <EOL> if cur_event is None : <EOL> self . _fds [ fd . _fileno ] = fd <EOL> self . _events [ fd . _fileno ] = event <EOL> self . _poller . register ( fd . _fileno , event ) <EOL> else : <EOL> event |= cur_event <EOL> self . _events [ fd . _fileno ] = event <EOL> self . _poller . modify ( fd . _fileno , event ) <EOL> self . _add_timeout ( fd ) <EOL> def clear ( self , fd , event = <NUM_LIT:0> ) : <EOL> cur_event = self . _events . get ( fd . _fileno , None ) <EOL> if cur_event : <EOL> if event : <EOL> cur_event &= ~ event <EOL> else : <EOL> cur_event = <NUM_LIT:0> <EOL> self . _events [ fd . _fileno ] = cur_event <EOL> self . _poller . modify ( fd . _fileno , cur_event ) <EOL> if not cur_event : <EOL> self . _del_timeout ( fd ) <EOL> @ staticmethod <EOL> def _socketpair ( ) : <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> return socket . socketpair ( ) <EOL> srv_sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> srv_sock . bind ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) ) <EOL> srv_sock . listen ( <NUM_LIT:1> ) <EOL> sock1 = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> conn_thread = threading . Thread ( target = lambda sock , addr_port : sock . connect ( addr_port ) , <EOL> args = ( sock1 , srv_sock . getsockname ( ) ) ) <EOL> conn_thread . daemon = True <EOL> conn_thread . start ( ) <EOL> sock2 , caddr = srv_sock . accept ( ) <EOL> srv_sock . close ( ) <EOL> return ( sock1 , sock2 ) <EOL> class _KQueueNotifier ( object ) : <EOL> """<STR_LIT>""" <EOL> __metaclass__ = MetaSingleton <EOL> def __init__ ( self ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . poller = select . kqueue ( ) <EOL> self . events = { } <EOL> def register ( self , fid , event ) : <EOL> self . events [ fid ] = event <EOL> self . update ( fid , event , select . KQ_EV_ADD ) <EOL> def unregister ( self , fid ) : <EOL> event = self . events . pop ( fid , None ) <EOL> if event is not None : <EOL> self . update ( fid , event , select . KQ_EV_DELETE ) <EOL> def modify ( self , fid , event ) : <EOL> self . unregister ( fid ) <EOL> self . register ( fid , event ) <EOL> def update ( self , fid , event , flags ) : <EOL> if event & _AsyncPoller . _Read : <EOL> self . poller . control ( [ select . kevent ( fid , filter = select . KQ_FILTER_READ , flags = flags ) ] , <NUM_LIT:0> ) <EOL> if event & _AsyncPoller . _Write : <EOL> self . poller . control ( [ select . kevent ( fid , filter = select . KQ_FILTER_WRITE , flags = flags ) ] , <NUM_LIT:0> ) <EOL> def poll ( self , timeout ) : <EOL> kevents = self . poller . control ( None , <NUM_LIT> , timeout ) <EOL> events = [ ( kevent . ident , <EOL> _AsyncPoller . _Read if kevent . filter == select . KQ_FILTER_READ else <EOL> _AsyncPoller . _Write if kevent . filter == select . KQ_FILTER_WRITE else <NUM_LIT:0> | <EOL> _AsyncPoller . _Hangup if kevent . flags == select . KQ_EV_EOF else <EOL> _AsyncPoller . _Error if kevent . flags == select . KQ_EV_ERROR else <NUM_LIT:0> ) <EOL> for kevent in kevents ] <EOL> return events <EOL> class _SelectNotifier ( object ) : <EOL> """<STR_LIT>""" <EOL> __metaclass__ = MetaSingleton <EOL> def __init__ ( self ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . poller = select . select <EOL> self . rset = set ( ) <EOL> self . wset = set ( ) <EOL> self . xset = set ( ) <EOL> def register ( self , fid , event ) : <EOL> if event : <EOL> if event & _AsyncPoller . _Read : <EOL> self . rset . add ( fid ) <EOL> if event & _AsyncPoller . _Write : <EOL> self . wset . add ( fid ) <EOL> if event & _AsyncPoller . _Error : <EOL> self . xset . add ( fid ) <EOL> def unregister ( self , fid ) : <EOL> self . rset . discard ( fid ) <EOL> self . wset . discard ( fid ) <EOL> self . xset . discard ( fid ) <EOL> def modify ( self , fid , event ) : <EOL> self . unregister ( fid ) <EOL> self . register ( fid , event ) <EOL> def poll ( self , timeout ) : <EOL> rlist , wlist , xlist = self . poller ( self . rset , self . wset , self . xset , timeout ) <EOL> events = { } <EOL> for fid in rlist : <EOL> events [ fid ] = _AsyncPoller . _Read <EOL> for fid in wlist : <EOL> events [ fid ] = events . get ( fid , <NUM_LIT:0> ) | _AsyncPoller . _Write <EOL> for fid in xlist : <EOL> events [ fid ] = events . get ( fid , <NUM_LIT:0> ) | _AsyncPoller . _Error <EOL> return events . iteritems ( ) <EOL> def terminate ( self ) : <EOL> self . rset = set ( ) <EOL> self . wset = set ( ) <EOL> self . xset = set ( ) <EOL> AsyncSocket = _AsyncSocket <EOL> _AsyncNotifier = _AsyncPoller <EOL> AsynCoroSocket = AsyncSocket <EOL> class Lock ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _owner = None <EOL> self . _waitlist = [ ] <EOL> def acquire ( self , blocking = True ) : <EOL> """<STR_LIT>""" <EOL> if not blocking and self . _owner is not None : <EOL> raise StopIteration ( False ) <EOL> coro = AsynCoro . cur_coro ( ) <EOL> while self . _owner is not None : <EOL> self . _waitlist . append ( coro ) <EOL> yield coro . _await_ ( ) <EOL> self . _owner = coro <EOL> raise StopIteration ( True ) <EOL> def release ( self ) : <EOL> """<STR_LIT>""" <EOL> coro = AsynCoro . cur_coro ( ) <EOL> if self . _owner != coro : <EOL> raise RuntimeError ( '<STR_LIT>' % ( coro . _name , coro . _id ) ) <EOL> self . _owner = None <EOL> if self . _waitlist : <EOL> wake = self . _waitlist . pop ( <NUM_LIT:0> ) <EOL> wake . _proceed_ ( True ) <EOL> class RLock ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _owner = None <EOL> self . _depth = <NUM_LIT:0> <EOL> self . _waitlist = [ ] <EOL> def acquire ( self , blocking = True ) : <EOL> """<STR_LIT>""" <EOL> coro = AsynCoro . cur_coro ( ) <EOL> if self . _owner == coro : <EOL> assert self . _depth > <NUM_LIT:0> <EOL> self . _depth += <NUM_LIT:1> <EOL> raise StopIteration ( True ) <EOL> if not blocking and self . _owner is not None : <EOL> raise StopIteration ( False ) <EOL> while self . _owner is not None : <EOL> self . _waitlist . append ( coro ) <EOL> yield coro . _await_ ( ) <EOL> assert self . _depth == <NUM_LIT:0> <EOL> self . _owner = coro <EOL> self . _depth = <NUM_LIT:1> <EOL> raise StopIteration ( True ) <EOL> def release ( self ) : <EOL> """<STR_LIT>""" <EOL> coro = AsynCoro . cur_coro ( ) <EOL> if self . _owner != coro : <EOL> raise RuntimeError ( '<STR_LIT>' % <EOL> ( coro . _name , coro . _id , self . _owner . _name , self . _owner . _id ) ) <EOL> self . _depth -= <NUM_LIT:1> <EOL> if self . _depth == <NUM_LIT:0> : <EOL> self . _owner = None <EOL> if self . _waitlist : <EOL> wake = self . _waitlist . pop ( <NUM_LIT:0> ) <EOL> wake . _proceed_ ( True ) <EOL> class Condition ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _owner = None <EOL> self . _depth = <NUM_LIT:0> <EOL> self . _waitlist = [ ] <EOL> self . _notifylist = [ ] <EOL> def acquire ( self , blocking = True ) : <EOL> """<STR_LIT>""" <EOL> coro = AsynCoro . cur_coro ( ) <EOL> if self . _owner == coro : <EOL> self . _depth += <NUM_LIT:1> <EOL> raise StopIteration ( True ) <EOL> if not blocking and self . _owner is not None : <EOL> raise StopIteration ( False ) <EOL> while self . _owner is not None : <EOL> self . _waitlist . append ( coro ) <EOL> yield coro . _await_ ( ) <EOL> assert self . _depth == <NUM_LIT:0> <EOL> self . _owner = coro <EOL> self . _depth = <NUM_LIT:1> <EOL> raise StopIteration ( True ) <EOL> def release ( self ) : <EOL> """<STR_LIT>""" <EOL> coro = AsynCoro . cur_coro ( ) <EOL> if self . _owner != coro : <EOL> raise RuntimeError ( '<STR_LIT>' % <EOL> ( coro . _name , coro . _id , self . _owner . _name , self . _owner . _id ) ) <EOL> self . _depth -= <NUM_LIT:1> <EOL> if self . _depth == <NUM_LIT:0> : <EOL> self . _owner = None <EOL> if self . _waitlist : <EOL> wake = self . _waitlist . pop ( <NUM_LIT:0> ) <EOL> wake . _proceed_ ( True ) <EOL> def notify ( self , n = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> while self . _notifylist and n : <EOL> wake = self . _notifylist . pop ( <NUM_LIT:0> ) <EOL> wake . _proceed_ ( True ) <EOL> n -= <NUM_LIT:1> <EOL> def notify_all ( self ) : <EOL> self . notify ( len ( self . _notifylist ) ) <EOL> notifyAll = notify_all <EOL> def wait ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> coro = AsynCoro . cur_coro ( ) <EOL> if self . _owner != coro : <EOL> raise RuntimeError ( '<STR_LIT>' % <EOL> ( coro . _name , coro . _id , self . _owner . _name , self . _owner . _id ) ) <EOL> assert self . _depth > <NUM_LIT:0> <EOL> depth = self . _depth <EOL> self . _depth = <NUM_LIT:0> <EOL> self . _owner = None <EOL> if self . _waitlist : <EOL> wake = self . _waitlist . pop ( <NUM_LIT:0> ) <EOL> wake . _proceed_ ( True ) <EOL> self . _notifylist . append ( coro ) <EOL> start = _time ( ) <EOL> if ( yield coro . _await_ ( timeout ) ) is None : <EOL> try : <EOL> self . _notifylist . remove ( coro ) <EOL> except ValueError : <EOL> pass <EOL> raise StopIteration ( False ) <EOL> while self . _owner is not None : <EOL> self . _waitlist . insert ( <NUM_LIT:0> , coro ) <EOL> if timeout is not None : <EOL> timeout -= ( _time ( ) - start ) <EOL> if timeout <= <NUM_LIT:0> : <EOL> raise StopIteration ( False ) <EOL> start = _time ( ) <EOL> if ( yield coro . _await_ ( timeout ) ) is None : <EOL> try : <EOL> self . _waitlist . remove ( coro ) <EOL> except ValueError : <EOL> pass <EOL> raise StopIteration ( False ) <EOL> assert self . _depth == <NUM_LIT:0> <EOL> self . _owner = coro <EOL> self . _depth = depth <EOL> raise StopIteration ( True ) <EOL> class Event ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _flag = False <EOL> self . _waitlist = [ ] <EOL> def set ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _flag = True <EOL> for coro in self . _waitlist : <EOL> coro . _proceed_ ( True ) <EOL> self . _waitlist = [ ] <EOL> def is_set ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _flag <EOL> isSet = is_set <EOL> def clear ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _flag = False <EOL> def wait ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if self . _flag : <EOL> raise StopIteration ( True ) <EOL> coro = AsynCoro . cur_coro ( ) <EOL> if timeout is not None : <EOL> if timeout <= <NUM_LIT:0> : <EOL> raise StopIteration ( False ) <EOL> self . _waitlist . append ( coro ) <EOL> if ( yield coro . _await_ ( timeout ) ) is None : <EOL> try : <EOL> self . _waitlist . remove ( coro ) <EOL> except ValueError : <EOL> pass <EOL> raise StopIteration ( False ) <EOL> else : <EOL> raise StopIteration ( True ) <EOL> class Semaphore ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , value = <NUM_LIT:1> ) : <EOL> assert value >= <NUM_LIT:1> <EOL> self . _waitlist = [ ] <EOL> self . _counter = value <EOL> def acquire ( self , blocking = True ) : <EOL> """<STR_LIT>""" <EOL> if blocking : <EOL> coro = AsynCoro . cur_coro ( ) <EOL> while self . _counter == <NUM_LIT:0> : <EOL> self . _waitlist . append ( coro ) <EOL> yield coro . _await_ ( ) <EOL> elif self . _counter == <NUM_LIT:0> : <EOL> raise StopIteration ( False ) <EOL> self . _counter -= <NUM_LIT:1> <EOL> raise StopIteration ( True ) <EOL> def release ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _counter += <NUM_LIT:1> <EOL> assert self . _counter > <NUM_LIT:0> <EOL> if self . _waitlist : <EOL> wake = self . _waitlist . pop ( <NUM_LIT:0> ) <EOL> wake . _proceed_ ( True ) <EOL> class HotSwapException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class MonitorException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Coro ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> _asyncoro = None <EOL> def __init__ ( self , target , * args , ** kwargs ) : <EOL> self . _generator = Coro . __get_generator ( self , target , * args , ** kwargs ) <EOL> self . _name = target . __name__ <EOL> self . _id = id ( self ) <EOL> self . _state = None <EOL> self . _value = None <EOL> self . _exceptions = [ ] <EOL> self . _callers = [ ] <EOL> self . _timeout = None <EOL> self . _daemon = False <EOL> self . _complete = None <EOL> self . _msgs = collections . deque ( ) <EOL> self . _monitors = set ( ) <EOL> self . _swap_generator = None <EOL> self . _hot_swappable = False <EOL> if not Coro . _asyncoro : <EOL> Coro . _asyncoro = AsynCoro . instance ( ) <EOL> self . _location = Coro . _asyncoro . _location <EOL> Coro . _asyncoro . _add ( self ) <EOL> @ property <EOL> def location ( self ) : <EOL> """<STR_LIT>""" <EOL> return copy . copy ( self . _location ) <EOL> @ property <EOL> def name ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _name <EOL> @ staticmethod <EOL> def locate ( name , location = None , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if not Coro . _asyncoro : <EOL> Coro . _asyncoro = AsynCoro . instance ( ) <EOL> if not location or location == Coro . _asyncoro . _location : <EOL> rcoro = Coro . _asyncoro . _rcoros . get ( name , None ) <EOL> if rcoro or location == Coro . _asyncoro . _location : <EOL> raise StopIteration ( rcoro ) <EOL> req = _NetRequest ( '<STR_LIT>' , kwargs = { '<STR_LIT:name>' : name } , dst = location , timeout = timeout ) <EOL> req_id = id ( req ) <EOL> req . event = Event ( ) <EOL> Coro . _asyncoro . _pending_reqs [ req_id ] = req <EOL> _Peer . send_req_to ( req , location ) <EOL> if ( yield req . event . wait ( timeout ) ) is False : <EOL> req . reply = None <EOL> Coro . _asyncoro . _pending_reqs . pop ( req_id , None ) <EOL> rcoro = req . reply <EOL> raise StopIteration ( rcoro ) <EOL> def register ( self , name = None ) : <EOL> """<STR_LIT>""" <EOL> if self . _location != Coro . _asyncoro . _location : <EOL> return - <NUM_LIT:1> <EOL> if not name : <EOL> name = self . _name <EOL> return Coro . _asyncoro . _register_coro ( self , name ) <EOL> def unregister ( self , name = None ) : <EOL> """<STR_LIT>""" <EOL> if self . _location != Coro . _asyncoro . _location : <EOL> return - <NUM_LIT:1> <EOL> if not name : <EOL> name = self . _name <EOL> return Coro . _asyncoro . _unregister_coro ( self , name ) <EOL> def set_daemon ( self , flag = True ) : <EOL> """<STR_LIT>""" <EOL> return Coro . _asyncoro . _set_daemon ( self , bool ( flag ) ) <EOL> def suspend ( self , timeout = None , alarm_value = None ) : <EOL> """<STR_LIT>""" <EOL> return Coro . _asyncoro . _suspend ( self , timeout , alarm_value , AsynCoro . _Suspended ) <EOL> sleep = suspend <EOL> def resume ( self , update = None ) : <EOL> """<STR_LIT>""" <EOL> return Coro . _asyncoro . _resume ( self , update , AsynCoro . _Suspended ) <EOL> wakeup = resume <EOL> def send ( self , message ) : <EOL> """<STR_LIT>""" <EOL> if self . _location == Coro . _asyncoro . _location : <EOL> return Coro . _asyncoro . _resume ( self , message , AsynCoro . _AwaitMsg_ ) <EOL> else : <EOL> request = _NetRequest ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : self . _id , '<STR_LIT:message>' : message } , <EOL> dst = self . _location , timeout = <NUM_LIT:5> ) <EOL> if _Peer . send_req ( request ) != <NUM_LIT:0> : <EOL> logger . warning ( '<STR_LIT>' , self . _location ) <EOL> return - <NUM_LIT:1> <EOL> else : <EOL> return <NUM_LIT:0> <EOL> def deliver ( self , message , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if self . _location == Coro . _asyncoro . _location : <EOL> reply = Coro . _asyncoro . _resume ( self , message , AsynCoro . _AwaitMsg_ ) <EOL> if reply == <NUM_LIT:0> : <EOL> reply = <NUM_LIT:1> <EOL> else : <EOL> request = _NetRequest ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : self . _id , '<STR_LIT:message>' : message } , <EOL> dst = self . _location , timeout = timeout ) <EOL> request . reply = - <NUM_LIT:1> <EOL> reply = yield Coro . _asyncoro . _sync_reply ( request , alarm_value = <NUM_LIT:0> ) <EOL> if reply is None : <EOL> reply = - <NUM_LIT:1> <EOL> raise StopIteration ( reply ) <EOL> def receive ( self , timeout = None , alarm_value = None ) : <EOL> """<STR_LIT>""" <EOL> return Coro . _asyncoro . _suspend ( self , timeout , alarm_value , AsynCoro . _AwaitMsg_ ) <EOL> recv = receive <EOL> def throw ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> if len ( args ) == <NUM_LIT:0> : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> return - <NUM_LIT:1> <EOL> if len ( args ) == <NUM_LIT:1> : <EOL> if isinstance ( args [ <NUM_LIT:0> ] , tuple ) and len ( args [ <NUM_LIT:0> ] ) > <NUM_LIT:1> : <EOL> args = args [ <NUM_LIT:0> ] <EOL> else : <EOL> args = ( type ( args [ <NUM_LIT:0> ] ) , args [ <NUM_LIT:0> ] ) <EOL> return Coro . _asyncoro . _throw ( self , * args ) <EOL> def value ( self ) : <EOL> """<STR_LIT>""" <EOL> Coro . _asyncoro . _lock . acquire ( ) <EOL> if self . _complete is None : <EOL> self . _complete = threading . Event ( ) <EOL> Coro . _asyncoro . _lock . release ( ) <EOL> self . _complete . wait ( ) <EOL> elif self . _complete == <NUM_LIT:0> : <EOL> Coro . _asyncoro . _lock . release ( ) <EOL> else : <EOL> Coro . _asyncoro . _lock . release ( ) <EOL> self . _complete . wait ( ) <EOL> return self . _value <EOL> def finish ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _complete is None : <EOL> self . _complete = Event ( ) <EOL> yield self . _complete . wait ( ) <EOL> elif self . _complete == <NUM_LIT:0> : <EOL> pass <EOL> elif isinstance ( self . _complete , Event ) : <EOL> yield self . _complete . wait ( ) <EOL> else : <EOL> raise RuntimeError ( '<STR_LIT>' % <EOL> ( self . _name , self . _id , type ( self . _complete ) ) ) <EOL> raise StopIteration ( self . _value ) <EOL> def terminate ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _location == Coro . _asyncoro . _location : <EOL> return Coro . _asyncoro . _terminate_coro ( self ) <EOL> else : <EOL> request = _NetRequest ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : self . _id } , <EOL> dst = self . _location , timeout = <NUM_LIT:5> ) <EOL> if _Peer . send_req ( request ) != <NUM_LIT:0> : <EOL> logger . warning ( '<STR_LIT>' , self . _location ) <EOL> return - <NUM_LIT:1> <EOL> return <NUM_LIT:0> <EOL> def hot_swappable ( self , flag ) : <EOL> if Coro . _asyncoro . cur_coro ( ) == self : <EOL> if flag : <EOL> self . _hot_swappable = True <EOL> if self . _swap_generator : <EOL> return Coro . _asyncoro . _swap_generator ( self ) <EOL> else : <EOL> self . _hot_swappable = False <EOL> return <NUM_LIT:0> <EOL> else : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> return - <NUM_LIT:1> <EOL> def hot_swap ( self , target , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> generator = Coro . __get_generator ( self , target , * args , ** kwargs ) <EOL> except : <EOL> logger . warning ( '<STR_LIT>' , target . __name__ ) <EOL> return - <NUM_LIT:1> <EOL> self . _swap_generator = generator <EOL> return Coro . _asyncoro . _swap_generator ( self ) <EOL> def monitor ( self , observe ) : <EOL> """<STR_LIT>""" <EOL> if observe . _location == Coro . _asyncoro . location : <EOL> reply = Coro . _asyncoro . _monitor ( self , observe ) <EOL> else : <EOL> request = _NetRequest ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : self , '<STR_LIT>' : observe . _id } , <EOL> dst = observe . _location , timeout = <NUM_LIT:5> ) <EOL> reply = yield Coro . _asyncoro . _sync_reply ( request ) <EOL> raise StopIteration ( reply ) <EOL> def notify ( self , monitor ) : <EOL> """<STR_LIT>""" <EOL> if Coro . _asyncoro . location == self . _location : <EOL> return Coro . _asyncoro . _monitor ( monitor , self ) <EOL> else : <EOL> return - <NUM_LIT:1> <EOL> def _await_ ( self , timeout = None , alarm_value = None ) : <EOL> """<STR_LIT>""" <EOL> return Coro . _asyncoro . _suspend ( self , timeout , alarm_value , AsynCoro . _AwaitIO_ ) <EOL> def _proceed_ ( self , update = None ) : <EOL> """<STR_LIT>""" <EOL> return Coro . _asyncoro . _resume ( self , update , AsynCoro . _AwaitIO_ ) <EOL> @ staticmethod <EOL> def __get_generator ( coro , target , * args , ** kwargs ) : <EOL> if not inspect . isgeneratorfunction ( target ) : <EOL> raise Exception ( '<STR_LIT>' % target . __name__ ) <EOL> if not args and kwargs : <EOL> args = kwargs . pop ( '<STR_LIT:args>' , ( ) ) <EOL> kwargs = kwargs . pop ( '<STR_LIT>' , kwargs ) <EOL> if target . func_defaults and '<STR_LIT>' in target . func_code . co_varnames [ : target . func_code . co_argcount ] [ - len ( target . func_defaults ) : ] : <EOL> kwargs [ '<STR_LIT>' ] = coro <EOL> return target ( * args , ** kwargs ) <EOL> def __getstate__ ( self ) : <EOL> state = { '<STR_LIT>' : self . _name , '<STR_LIT>' : str ( self . _id ) , '<STR_LIT>' : self . _location } <EOL> return state <EOL> def __setstate__ ( self , state ) : <EOL> self . _name = state [ '<STR_LIT>' ] <EOL> self . _location = state [ '<STR_LIT>' ] <EOL> self . _id = state [ '<STR_LIT>' ] <EOL> def __repr__ ( self ) : <EOL> s = '<STR_LIT>' % ( self . _name , self . _id ) <EOL> if self . _location : <EOL> s = '<STR_LIT>' % ( s , self . _location ) <EOL> return s <EOL> def __eq__ ( self , other ) : <EOL> return isinstance ( other , Coro ) and self . _location == other . _location and self . _id == other . _id <EOL> def __hash__ ( self ) : <EOL> if self . _location : <EOL> return hash ( str ( self ) ) <EOL> else : <EOL> return hash ( id ( self ) ) <EOL> class Location ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT:port>' ) <EOL> def __init__ ( self , host , tcp_port ) : <EOL> if re . match ( r'<STR_LIT>' , host ) : <EOL> self . addr = host <EOL> else : <EOL> self . addr = socket . gethostbyname ( host ) <EOL> self . port = int ( tcp_port ) <EOL> def __eq__ ( self , other ) : <EOL> return isinstance ( other , type ( self ) ) and self . addr == other . addr and self . port == other . port <EOL> def __ne__ ( self , other ) : <EOL> return ( not isinstance ( other , type ( self ) ) ) or self . addr != other . addr or self . port != other . port <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . addr , self . port ) <EOL> def __hash__ ( self ) : <EOL> return hash ( str ( self ) ) <EOL> class Channel ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> _asyncoro = None <EOL> def __init__ ( self , name , transform = None ) : <EOL> """<STR_LIT>""" <EOL> if not Channel . _asyncoro : <EOL> Channel . _asyncoro = AsynCoro . instance ( ) <EOL> self . _name = name <EOL> self . _location = Channel . _asyncoro . _location <EOL> if transform is not None : <EOL> try : <EOL> argspec = inspect . getargspec ( transform ) <EOL> assert len ( argspec . args ) == <NUM_LIT:2> <EOL> except : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> transform = None <EOL> self . _transform = transform <EOL> self . _subscribers = set ( ) <EOL> self . _subscribe_event = Event ( ) <EOL> Channel . _asyncoro . _lock . acquire ( ) <EOL> if name in Channel . _asyncoro . _channels : <EOL> Channel . _asyncoro . _lock . release ( ) <EOL> logger . warning ( '<STR_LIT>' , name ) <EOL> else : <EOL> Channel . _asyncoro . _channels [ name ] = self <EOL> Channel . _asyncoro . _lock . release ( ) <EOL> @ property <EOL> def location ( self ) : <EOL> """<STR_LIT>""" <EOL> return copy . copy ( self . _location ) <EOL> @ property <EOL> def name ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _name <EOL> @ staticmethod <EOL> def locate ( name , location = None , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if not Channel . _asyncoro : <EOL> Channel . _asyncoro = AsynCoro . instance ( ) <EOL> if not location or location == Channel . _asyncoro . _location : <EOL> rchannel = Channel . _asyncoro . _channels . get ( name , None ) <EOL> if rchannel or location == Channel . _asyncoro . _location : <EOL> raise StopIteration ( rchannel ) <EOL> req = _NetRequest ( '<STR_LIT>' , kwargs = { '<STR_LIT:name>' : name } , dst = location , timeout = timeout ) <EOL> req . event = Event ( ) <EOL> req_id = id ( req ) <EOL> Channel . _asyncoro . _pending_reqs [ req_id ] = req <EOL> _Peer . send_req_to ( req , location ) <EOL> if ( yield req . event . wait ( timeout ) ) is False : <EOL> req . reply = None <EOL> Channel . _asyncoro . _pending_reqs . pop ( req_id , None ) <EOL> rchannel = req . reply <EOL> raise StopIteration ( rchannel ) <EOL> def register ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _location != Channel . _asyncoro . _location : <EOL> return - <NUM_LIT:1> <EOL> return Channel . _asyncoro . _register_channel ( self , self . _name ) <EOL> def unregister ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _location != Channel . _asyncoro . _location : <EOL> return - <NUM_LIT:1> <EOL> return Channel . _asyncoro . _unregister_channel ( self , self . _name ) <EOL> def set_transform ( self , transform ) : <EOL> if self . _location != Channel . _asyncoro . _location : <EOL> return - <NUM_LIT:1> <EOL> try : <EOL> argspec = inspect . getargspec ( transform ) <EOL> assert len ( argspec . args ) == <NUM_LIT:2> <EOL> except : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> return - <NUM_LIT:1> <EOL> self . _transform = transform <EOL> return <NUM_LIT:0> <EOL> def subscribe ( self , subscriber , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( subscriber , Coro ) and not isinstance ( subscriber , Channel ) : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> raise StopIteration ( - <NUM_LIT:1> ) <EOL> if self . _location == Channel . _asyncoro . _location : <EOL> if subscriber . _location != self . _location : <EOL> if isinstance ( subscriber , Coro ) : <EOL> subscriber . _id = int ( subscriber . _id ) <EOL> for s in self . _subscribers : <EOL> if isinstance ( s , Coro ) and s . _id == subscriber . _id and s . _location == subscriber . _location : <EOL> subscriber = s <EOL> break <EOL> elif isinstance ( subscriber , Channel ) : <EOL> for s in self . _subscribers : <EOL> if isinstance ( s , Channel ) and s . _name == subscriber . _name and s . _location == subscriber . _location : <EOL> subscriber = s <EOL> break <EOL> self . _subscribers . add ( subscriber ) <EOL> yield self . _subscribe_event . set ( ) <EOL> reply = <NUM_LIT:0> <EOL> else : <EOL> kwargs = { '<STR_LIT>' : self . _name } <EOL> kwargs [ '<STR_LIT>' ] = subscriber <EOL> request = _NetRequest ( '<STR_LIT>' , kwargs = kwargs , dst = self . _location , timeout = timeout ) <EOL> reply = yield Channel . _asyncoro . _sync_reply ( request ) <EOL> raise StopIteration ( reply ) <EOL> def unsubscribe ( self , subscriber , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( subscriber , Coro ) and not isinstance ( subscriber , Channel ) : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> raise StopIteration ( - <NUM_LIT:1> ) <EOL> if self . _location == Channel . _asyncoro . _location : <EOL> if subscriber . _location != self . _location : <EOL> if isinstance ( subscriber , Coro ) : <EOL> subscriber . _id = int ( subscriber . _id ) <EOL> for s in self . _subscribers : <EOL> if isinstance ( s , Coro ) and s . _id == subscriber . _id and s . _location == subscriber . _location : <EOL> subscriber = s <EOL> break <EOL> elif isinstance ( subscriber , Channel ) : <EOL> for s in self . _subscribers : <EOL> if isinstance ( s , Channel ) and s . _name == subscriber . _name and s . _location == subscriber . _location : <EOL> subscriber = s <EOL> break <EOL> try : <EOL> self . _subscribers . remove ( subscriber ) <EOL> except KeyError : <EOL> reply = - <NUM_LIT:1> <EOL> else : <EOL> reply = <NUM_LIT:0> <EOL> else : <EOL> kwargs = { '<STR_LIT>' : self . _name } <EOL> kwargs [ '<STR_LIT>' ] = subscriber <EOL> request = _NetRequest ( '<STR_LIT>' , kwargs = kwargs , dst = self . _location , timeout = timeout ) <EOL> reply = yield Channel . _asyncoro . _sync_reply ( request ) <EOL> raise StopIteration ( reply ) <EOL> def send ( self , message ) : <EOL> """<STR_LIT>""" <EOL> if self . _location == Channel . _asyncoro . _location : <EOL> if self . _transform : <EOL> try : <EOL> message = self . _transform ( self . _name , message ) <EOL> except : <EOL> message = None <EOL> if message is None : <EOL> return <NUM_LIT:0> <EOL> invalid = [ ] <EOL> for subscriber in self . _subscribers : <EOL> if subscriber . send ( message ) != <NUM_LIT:0> : <EOL> invalid . append ( subscriber ) <EOL> if invalid : <EOL> def _unsub ( self , subscriber , coro = None ) : <EOL> logger . debug ( '<STR_LIT>' , subscriber ) <EOL> yield self . unsubscribe ( subscriber ) <EOL> for subscriber in invalid : <EOL> Coro ( _unsub , self , subscriber ) <EOL> else : <EOL> request = _NetRequest ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : self . _name , '<STR_LIT:message>' : message } , <EOL> dst = self . _location , timeout = <NUM_LIT:5> ) <EOL> if _Peer . send_req ( request ) != <NUM_LIT:0> : <EOL> logger . warning ( '<STR_LIT>' , self . _location ) <EOL> return - <NUM_LIT:1> <EOL> return <NUM_LIT:0> <EOL> def deliver ( self , message , timeout = None , n = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( n , int ) or n < <NUM_LIT:0> : <EOL> raise StopIteration ( - <NUM_LIT:1> ) <EOL> if self . _location == Channel . _asyncoro . _location : <EOL> if self . _transform : <EOL> try : <EOL> message = self . _transform ( self . _name , message ) <EOL> except : <EOL> message = None <EOL> if message is None : <EOL> raise StopIteration ( <NUM_LIT:0> ) <EOL> if n : <EOL> while len ( self . _subscribers ) < n : <EOL> start = _time ( ) <EOL> self . _subscribe_event . clear ( ) <EOL> if ( yield self . _subscribe_event . wait ( timeout ) ) is False : <EOL> raise StopIteration ( <NUM_LIT:0> ) <EOL> if timeout is not None : <EOL> timeout -= _time ( ) - start <EOL> if timeout <= <NUM_LIT:0> : <EOL> raise StopIteration ( <NUM_LIT:0> ) <EOL> subscribers = list ( self . _subscribers ) <EOL> info = { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : len ( subscribers ) , '<STR_LIT:success>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : Event ( ) , '<STR_LIT>' : [ ] } <EOL> def _deliver ( subscriber , info , timeout , n , coro = None ) : <EOL> try : <EOL> reply = yield subscriber . deliver ( message , timeout = timeout ) <EOL> if reply > <NUM_LIT:0> : <EOL> info [ '<STR_LIT>' ] += reply <EOL> info [ '<STR_LIT:success>' ] += <NUM_LIT:1> <EOL> if n > <NUM_LIT:0> and info [ '<STR_LIT:success>' ] >= n : <EOL> info [ '<STR_LIT>' ] . set ( ) <EOL> elif reply < <NUM_LIT:0> : <EOL> info [ '<STR_LIT>' ] . append ( subscriber ) <EOL> except : <EOL> pass <EOL> info [ '<STR_LIT>' ] -= <NUM_LIT:1> <EOL> if info [ '<STR_LIT>' ] == <NUM_LIT:0> : <EOL> info [ '<STR_LIT>' ] . set ( ) <EOL> for subscriber in subscribers : <EOL> if isinstance ( subscriber , Coro ) and self . _location == subscriber . _location : <EOL> if subscriber . send ( message ) == <NUM_LIT:0> : <EOL> info [ '<STR_LIT>' ] += <NUM_LIT:1> <EOL> info [ '<STR_LIT:success>' ] += <NUM_LIT:1> <EOL> info [ '<STR_LIT>' ] -= <NUM_LIT:1> <EOL> else : <EOL> Coro ( _deliver , subscriber , info , timeout , n ) <EOL> if info [ '<STR_LIT>' ] == <NUM_LIT:0> : <EOL> info [ '<STR_LIT>' ] . set ( ) <EOL> if n == <NUM_LIT:0> or info [ '<STR_LIT:success>' ] < n : <EOL> yield info [ '<STR_LIT>' ] . wait ( timeout ) <EOL> if info [ '<STR_LIT>' ] : <EOL> def _unsub ( self , subscriber , coro = None ) : <EOL> logger . debug ( '<STR_LIT>' , subscriber ) <EOL> yield self . unsubscribe ( subscriber ) <EOL> for subscriber in info [ '<STR_LIT>' ] : <EOL> Coro ( _unsub , self , subscriber ) <EOL> raise StopIteration ( info [ '<STR_LIT>' ] ) <EOL> else : <EOL> request = _NetRequest ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : self . _name , '<STR_LIT:message>' : message , <EOL> '<STR_LIT:n>' : n } , <EOL> dst = self . _location , timeout = timeout ) <EOL> request . reply = - <NUM_LIT:1> <EOL> reply = yield Channel . _asyncoro . _sync_reply ( request , alarm_value = <NUM_LIT:0> ) <EOL> if reply is None : <EOL> reply = - <NUM_LIT:1> <EOL> raise StopIteration ( reply ) <EOL> def close ( self ) : <EOL> if self . _location == Channel . _asyncoro . _location : <EOL> self . unregister ( ) <EOL> self . _subscribers = set ( ) <EOL> Channel . _asyncoro . _lock . acquire ( ) <EOL> Channel . _asyncoro . _channels . pop ( self . _name , None ) <EOL> Channel . _asyncoro . _lock . release ( ) <EOL> def __getstate__ ( self ) : <EOL> state = { '<STR_LIT>' : self . _name , '<STR_LIT>' : self . _location } <EOL> return state <EOL> def __setstate__ ( self , state ) : <EOL> self . _name = state [ '<STR_LIT>' ] <EOL> self . _location = state [ '<STR_LIT>' ] <EOL> self . _transform = None <EOL> def __repr__ ( self ) : <EOL> s = '<STR_LIT:%s>' % ( self . _name ) <EOL> if self . _location : <EOL> s = '<STR_LIT>' % ( s , self . _location ) <EOL> return s <EOL> class CategorizeMessages ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , coro ) : <EOL> """<STR_LIT>""" <EOL> self . _coro = coro <EOL> self . _categories = { None : collections . deque ( ) } <EOL> self . _categorize = [ ] <EOL> def add ( self , categorize ) : <EOL> """<STR_LIT>""" <EOL> if inspect . isfunction ( categorize ) : <EOL> argspec = inspect . getargspec ( categorize ) <EOL> if len ( argspec . args ) != <NUM_LIT:1> : <EOL> categorize = None <EOL> elif type ( categorize ) != functools . partial : <EOL> categorize = None <EOL> if categorize : <EOL> self . _categorize . append ( categorize ) <EOL> else : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> def remove ( self , categorize ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _categorize . remove ( categorize ) <EOL> except ValueError : <EOL> logger . warning ( '<STR_LIT>' ) <EOL> def receive ( self , category = None , timeout = None , alarm_value = None ) : <EOL> """<STR_LIT>""" <EOL> c = self . _categories . get ( category , None ) <EOL> if c : <EOL> msg = c . popleft ( ) <EOL> raise StopIteration ( msg ) <EOL> if timeout : <EOL> start = _time ( ) <EOL> while <NUM_LIT:1> : <EOL> msg = yield self . _coro . receive ( timeout = timeout , alarm_value = alarm_value ) <EOL> if msg == alarm_value : <EOL> raise StopIteration ( msg ) <EOL> for categorize in reversed ( self . _categorize ) : <EOL> c = categorize ( msg ) <EOL> if c == category : <EOL> raise StopIteration ( msg ) <EOL> if c is not None : <EOL> bucket = self . _categories . get ( c , None ) <EOL> if not bucket : <EOL> bucket = self . _categories [ c ] = collections . deque ( ) <EOL> bucket . append ( msg ) <EOL> break <EOL> else : <EOL> self . _categories [ None ] . append ( msg ) <EOL> if timeout : <EOL> now = _time ( ) <EOL> timeout -= now - start <EOL> start = now <EOL> recv = receive <EOL> class AsynCoro ( object ) : <EOL> """<STR_LIT>""" <EOL> __metaclass__ = MetaSingleton <EOL> __instance = None <EOL> __cur_coro = None <EOL> _Scheduled = <NUM_LIT:1> <EOL> _Running = <NUM_LIT:2> <EOL> _Suspended = <NUM_LIT:3> <EOL> _AwaitIO_ = <NUM_LIT:4> <EOL> _AwaitMsg_ = <NUM_LIT:5> <EOL> def __init__ ( self , notifier = None ) : <EOL> if self . __class__ . __instance is None : <EOL> self . __class__ . __instance = self <EOL> if notifier is None : <EOL> self . _notifier = _AsyncNotifier ( ) <EOL> else : <EOL> self . _notifier = notifier <EOL> self . _location = None <EOL> self . _name = None <EOL> self . _coros = { } <EOL> self . _scheduled = set ( ) <EOL> self . _suspended = set ( ) <EOL> self . _timeouts = [ ] <EOL> self . _lock = threading . RLock ( ) <EOL> self . _quit = False <EOL> self . _complete = threading . Event ( ) <EOL> self . _daemons = <NUM_LIT:0> <EOL> self . _polling = False <EOL> self . _channels = { } <EOL> self . _atexit = [ ] <EOL> Coro . _asyncoro = Channel . _asyncoro = self <EOL> self . _scheduler = threading . Thread ( target = self . _schedule ) <EOL> self . _scheduler . daemon = True <EOL> self . _scheduler . start ( ) <EOL> atexit . register ( self . finish ) <EOL> logger . info ( '<STR_LIT>' , <EOL> __version__ , self . _notifier . _poller_name ) <EOL> @ property <EOL> def location ( self ) : <EOL> """<STR_LIT>""" <EOL> return copy . copy ( self . _location ) <EOL> @ property <EOL> def name ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _name <EOL> @ classmethod <EOL> def instance ( cls , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if cls . __instance is None : <EOL> cls . __instance = cls ( * args , ** kwargs ) <EOL> return cls . __instance <EOL> @ staticmethod <EOL> def cur_coro ( ) : <EOL> """<STR_LIT>""" <EOL> return AsynCoro . __cur_coro <EOL> def _add ( self , coro ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> self . _coros [ coro . _id ] = coro <EOL> self . _complete . clear ( ) <EOL> coro . _state = AsynCoro . _Scheduled <EOL> self . _scheduled . add ( coro . _id ) <EOL> if self . _polling and len ( self . _scheduled ) == <NUM_LIT:1> : <EOL> self . _notifier . interrupt ( ) <EOL> self . _lock . release ( ) <EOL> def _remove ( self , coro ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> try : <EOL> self . _scheduled . remove ( coro . _id ) <EOL> except KeyError : <EOL> ret = - <NUM_LIT:1> <EOL> else : <EOL> self . _coros . pop ( coro . _id , None ) <EOL> ret = <NUM_LIT:0> <EOL> self . _lock . release ( ) <EOL> return ret <EOL> def _set_daemon ( self , coro , flag ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> if AsynCoro . __cur_coro != coro : <EOL> self . _lock . release ( ) <EOL> logger . warning ( '<STR_LIT>' , coro , AsynCoro . __cur_coro ) <EOL> return - <NUM_LIT:1> <EOL> if coro . _daemon != flag : <EOL> coro . _daemon = flag <EOL> if flag : <EOL> self . _daemons += <NUM_LIT:1> <EOL> if len ( self . _coros ) == self . _daemons : <EOL> self . _complete . set ( ) <EOL> else : <EOL> self . _daemons -= <NUM_LIT:1> <EOL> self . _complete . clear ( ) <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> def _monitor ( self , monitor , coro ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> cid = coro . _id <EOL> coro = self . _coros . get ( cid , None ) <EOL> if coro is None or not isinstance ( monitor , Coro ) : <EOL> self . _lock . release ( ) <EOL> logger . warning ( '<STR_LIT>' ) <EOL> return - <NUM_LIT:1> <EOL> coro . _monitors . add ( monitor ) <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> def _suspend ( self , coro , timeout , alarm_value , state ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> if AsynCoro . __cur_coro != coro : <EOL> self . _lock . release ( ) <EOL> logger . warning ( '<STR_LIT>' , coro , AsynCoro . __cur_coro ) <EOL> return - <NUM_LIT:1> <EOL> cid = coro . _id <EOL> if state == AsynCoro . _AwaitMsg_ and coro . _msgs : <EOL> s , update = coro . _msgs [ <NUM_LIT:0> ] <EOL> if s == state : <EOL> coro . _msgs . popleft ( ) <EOL> self . _lock . release ( ) <EOL> return update <EOL> if timeout is None : <EOL> coro . _timeout = None <EOL> else : <EOL> if not isinstance ( timeout , ( float , int ) ) : <EOL> logger . warning ( '<STR_LIT>' , timeout ) <EOL> self . _lock . release ( ) <EOL> return - <NUM_LIT:1> <EOL> if timeout <= <NUM_LIT:0> : <EOL> self . _lock . release ( ) <EOL> return alarm_value <EOL> else : <EOL> coro . _timeout = _time ( ) + timeout <EOL> heappush ( self . _timeouts , ( coro . _timeout , cid , alarm_value ) ) <EOL> self . _scheduled . discard ( cid ) <EOL> self . _suspended . add ( cid ) <EOL> coro . _state = state <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> def _resume ( self , coro , update , state ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> cid = coro . _id <EOL> coro = self . _coros . get ( cid , None ) <EOL> if coro is None : <EOL> self . _lock . release ( ) <EOL> logger . warning ( '<STR_LIT>' , cid ) <EOL> return - <NUM_LIT:1> <EOL> if state == AsynCoro . _AwaitMsg_ and coro . _state != state : <EOL> coro . _msgs . append ( ( state , update ) ) <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> if coro . _state == state : <EOL> coro . _timeout = None <EOL> coro . _value = update <EOL> self . _suspended . discard ( cid ) <EOL> self . _scheduled . add ( cid ) <EOL> coro . _state = AsynCoro . _Scheduled <EOL> if self . _polling and len ( self . _scheduled ) == <NUM_LIT:1> : <EOL> self . _notifier . interrupt ( ) <EOL> else : <EOL> logger . warning ( '<STR_LIT>' , coro . _name , cid , coro . _state ) <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> def _throw ( self , coro , * args ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> cid = coro . _id <EOL> coro = self . _coros . get ( cid , None ) <EOL> if coro is None or coro . _state not in ( AsynCoro . _Scheduled , AsynCoro . _Suspended , <EOL> AsynCoro . _AwaitIO_ , AsynCoro . _AwaitMsg_ ) : <EOL> logger . warning ( '<STR_LIT>' , cid ) <EOL> self . _lock . release ( ) <EOL> return - <NUM_LIT:1> <EOL> coro . _timeout = None <EOL> coro . _exceptions . append ( args ) <EOL> if coro . _state in ( AsynCoro . _AwaitIO_ , AsynCoro . _Suspended , AsynCoro . _AwaitMsg_ ) : <EOL> self . _suspended . discard ( cid ) <EOL> self . _scheduled . add ( cid ) <EOL> coro . _state = AsynCoro . _Scheduled <EOL> if self . _polling and len ( self . _scheduled ) == <NUM_LIT:1> : <EOL> self . _notifier . interrupt ( ) <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> def _terminate_coro ( self , coro ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> cid = coro . _id <EOL> coro = self . _coros . get ( cid , None ) <EOL> if coro is None : <EOL> logger . warning ( '<STR_LIT>' , cid ) <EOL> self . _lock . release ( ) <EOL> return - <NUM_LIT:1> <EOL> if coro . _state == AsynCoro . _Running : <EOL> logger . warning ( '<STR_LIT>' , coro . _name , cid ) <EOL> else : <EOL> self . _suspended . discard ( cid ) <EOL> self . _scheduled . add ( cid ) <EOL> coro . _exceptions . append ( ( GeneratorExit , GeneratorExit ( '<STR_LIT>' ) ) ) <EOL> coro . _timeout = None <EOL> coro . _state = AsynCoro . _Scheduled <EOL> coro . _callers = [ ] <EOL> if self . _polling and len ( self . _scheduled ) == <NUM_LIT:1> : <EOL> self . _notifier . interrupt ( ) <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> def _swap_generator ( self , coro ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( ) <EOL> cid = coro . _id <EOL> coro = self . _coros . get ( cid , None ) <EOL> if coro is None : <EOL> logger . warning ( '<STR_LIT>' , cid ) <EOL> self . _lock . release ( ) <EOL> return - <NUM_LIT:1> <EOL> if coro . _callers or not coro . _hot_swappable : <EOL> logger . debug ( '<STR_LIT>' , str ( coro ) ) <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> else : <EOL> coro . _timeout = None <EOL> if coro . _state is None : <EOL> coro . _generator = coro . _swap_generator <EOL> coro . _value = None <EOL> if coro . _complete == <NUM_LIT:0> : <EOL> coro . _complete = None <EOL> elif isinstance ( coro . _complete , Event ) : <EOL> coro . _complete . clear ( ) <EOL> self . _scheduled . add ( cid ) <EOL> coro . _state = AsynCoro . _Scheduled <EOL> coro . _hot_swappable = False <EOL> else : <EOL> coro . _exceptions . append ( ( HotSwapException , HotSwapException ( coro . _swap_generator ) ) ) <EOL> if coro . _state in ( AsynCoro . _Suspended , AsynCoro . _AwaitMsg_ ) : <EOL> self . _suspended . discard ( cid ) <EOL> self . _scheduled . add ( cid ) <EOL> coro . _state = AsynCoro . _Scheduled <EOL> coro . _swap_generator = None <EOL> if self . _polling and len ( self . _scheduled ) == <NUM_LIT:1> : <EOL> self . _notifier . interrupt ( ) <EOL> self . _lock . release ( ) <EOL> return <NUM_LIT:0> <EOL> def _schedule ( self ) : <EOL> """<STR_LIT>""" <EOL> while not self . _quit : <EOL> self . _notifier . poll ( <NUM_LIT:0> ) <EOL> self . _lock . acquire ( ) <EOL> if not self . _scheduled : <EOL> if self . _timeouts : <EOL> timeout = self . _timeouts [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] - _time ( ) <EOL> else : <EOL> timeout = None <EOL> self . _polling = True <EOL> self . _lock . release ( ) <EOL> self . _notifier . poll ( timeout ) <EOL> self . _lock . acquire ( ) <EOL> self . _polling = False <EOL> if self . _timeouts : <EOL> now = _time ( ) + <NUM_LIT> <EOL> while self . _timeouts and self . _timeouts [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <= now : <EOL> timeout , cid , alarm_value = heappop ( self . _timeouts ) <EOL> assert timeout <= now <EOL> coro = self . _coros . get ( cid , None ) <EOL> if not coro or coro . _timeout != timeout : <EOL> continue <EOL> if coro . _state not in ( AsynCoro . _AwaitIO_ , AsynCoro . _Suspended , <EOL> AsynCoro . _AwaitMsg_ ) : <EOL> logger . warning ( '<STR_LIT>' , <EOL> coro . _name , coro . _id , coro . _state ) <EOL> continue <EOL> coro . _timeout = None <EOL> self . _suspended . discard ( cid ) <EOL> self . _scheduled . add ( cid ) <EOL> coro . _state = AsynCoro . _Scheduled <EOL> coro . _value = alarm_value <EOL> scheduled = [ self . _coros [ cid ] for cid in self . _scheduled ] <EOL> self . _lock . release ( ) <EOL> for coro in scheduled : <EOL> coro . _state = AsynCoro . _Running <EOL> AsynCoro . __cur_coro = coro <EOL> try : <EOL> if coro . _exceptions : <EOL> exc = coro . _exceptions . pop ( <NUM_LIT:0> ) <EOL> if exc [ <NUM_LIT:0> ] == GeneratorExit : <EOL> coro . _generator . close ( ) <EOL> retval = coro . _value <EOL> else : <EOL> retval = coro . _generator . throw ( * exc ) <EOL> else : <EOL> retval = coro . _generator . send ( coro . _value ) <EOL> except : <EOL> self . _lock . acquire ( ) <EOL> exc = sys . exc_info ( ) <EOL> if exc [ <NUM_LIT:0> ] == StopIteration : <EOL> v = exc [ <NUM_LIT:1> ] . args <EOL> if v : <EOL> if len ( v ) == <NUM_LIT:1> : <EOL> coro . _value = v [ <NUM_LIT:0> ] <EOL> else : <EOL> coro . _value = v <EOL> coro . _exceptions = [ ] <EOL> elif exc [ <NUM_LIT:0> ] == HotSwapException : <EOL> v = exc [ <NUM_LIT:1> ] . args <EOL> if isinstance ( v , tuple ) and len ( v ) == <NUM_LIT:1> and inspect . isgenerator ( v [ <NUM_LIT:0> ] ) and coro . _hot_swappable and not coro . _callers : <EOL> try : <EOL> coro . _generator . close ( ) <EOL> except : <EOL> logger . warning ( '<STR_LIT>' , <EOL> coro . _name , coro . _id , traceback . format_exc ( ) ) <EOL> coro . _generator = v [ <NUM_LIT:0> ] <EOL> coro . _name = coro . _generator . __name__ <EOL> coro . _exceptions = [ ] <EOL> coro . _value = None <EOL> coro . _state = AsynCoro . _Scheduled <EOL> else : <EOL> logger . warning ( '<STR_LIT>' , <EOL> coro . _name , coro . _id ) <EOL> self . _lock . release ( ) <EOL> continue <EOL> else : <EOL> coro . _exceptions . append ( exc ) <EOL> if coro . _callers : <EOL> caller = coro . _callers . pop ( - <NUM_LIT:1> ) <EOL> coro . _generator = caller [ <NUM_LIT:0> ] <EOL> if coro . _swap_generator and not coro . _callers and coro . _hot_swappable : <EOL> coro . _exceptions . append ( ( HotSwapException , <EOL> HotSwapException ( coro . _swap_generator ) ) ) <EOL> coro . _swap_generator = None <EOL> coro . _state = AsynCoro . _Scheduled <EOL> elif coro . _exceptions : <EOL> coro . _value = caller [ <NUM_LIT:1> ] <EOL> self . _suspended . discard ( coro . _id ) <EOL> self . _scheduled . add ( coro . _id ) <EOL> coro . _state = AsynCoro . _Scheduled <EOL> elif coro . _state == AsynCoro . _Running : <EOL> coro . _state = AsynCoro . _Scheduled <EOL> else : <EOL> if coro . _exceptions : <EOL> exc = coro . _exceptions [ <NUM_LIT:0> ] <EOL> assert isinstance ( exc , tuple ) <EOL> if len ( exc ) == <NUM_LIT:2> : <EOL> exc = '<STR_LIT>' . join ( traceback . format_exception_only ( * exc ) ) <EOL> else : <EOL> exc = '<STR_LIT>' . join ( traceback . format_exception ( * exc ) ) <EOL> logger . warning ( '<STR_LIT>' , coro , exc ) <EOL> try : <EOL> coro . _generator . close ( ) <EOL> except : <EOL> logger . warning ( '<STR_LIT>' , <EOL> coro . _name , traceback . format_exc ( ) ) <EOL> if coro . _state not in ( AsynCoro . _Scheduled , AsynCoro . _Running ) : <EOL> logger . warning ( '<STR_LIT>' , coro . _name , coro . _state ) <EOL> monitors = list ( coro . _monitors ) <EOL> for monitor in monitors : <EOL> if monitor . _location == self . _location : <EOL> if coro . _exceptions : <EOL> exc = MonitorException ( coro , coro . _exceptions [ <NUM_LIT:0> ] ) <EOL> else : <EOL> exc = MonitorException ( coro , ( StopIteration , coro . _value ) ) <EOL> if self . _coros . get ( monitor . _id , None ) == monitor : <EOL> monitor . send ( exc ) <EOL> else : <EOL> logger . warning ( '<STR_LIT>' , <EOL> coro . _name , coro . _id ) <EOL> coro . _monitors . discard ( monitor ) <EOL> else : <EOL> if coro . _exceptions : <EOL> exc = coro . _exceptions [ <NUM_LIT:0> ] [ : <NUM_LIT:2> ] <EOL> try : <EOL> serialize ( exc [ <NUM_LIT:1> ] ) <EOL> except pickle . PicklingError : <EOL> exc = ( exc [ <NUM_LIT:0> ] , type ( exc [ <NUM_LIT:1> ] . args [ <NUM_LIT:0> ] ) ) <EOL> exc = MonitorException ( coro , exc ) <EOL> coro . _exceptions = [ ] <EOL> else : <EOL> exc = coro . _value <EOL> try : <EOL> serialize ( exc ) <EOL> except pickle . PicklingError : <EOL> exc = type ( exc ) <EOL> exc = MonitorException ( coro , ( StopIteration , exc ) ) <EOL> monitor . send ( exc ) <EOL> if not coro . _monitors or not coro . _exceptions : <EOL> coro . _msgs . clear ( ) <EOL> coro . _monitors . clear ( ) <EOL> coro . _exceptions = [ ] <EOL> if self . _coros . pop ( coro . _id , None ) != coro : <EOL> logger . warning ( '<STR_LIT>' , coro . _id , coro . _state ) <EOL> if coro . _daemon is True : <EOL> self . _daemons -= <NUM_LIT:1> <EOL> elif coro . _monitors : <EOL> coro . _hot_swappable = True <EOL> coro . _exceptions = [ ] <EOL> coro . _state = None <EOL> coro . _generator = None <EOL> if coro . _complete : <EOL> coro . _complete . set ( ) <EOL> else : <EOL> coro . _complete = <NUM_LIT:0> <EOL> self . _scheduled . discard ( coro . _id ) <EOL> if len ( self . _coros ) == self . _daemons : <EOL> self . _complete . set ( ) <EOL> self . _lock . release ( ) <EOL> else : <EOL> self . _lock . acquire ( ) <EOL> if coro . _state == AsynCoro . _Running : <EOL> coro . _state = AsynCoro . _Scheduled <EOL> coro . _value = retval <EOL> if isinstance ( retval , types . GeneratorType ) : <EOL> coro . _callers . append ( ( coro . _generator , coro . _value ) ) <EOL> coro . _generator = retval <EOL> coro . _value = None <EOL> self . _lock . release ( ) <EOL> AsynCoro . __cur_coro = None <EOL> self . _lock . acquire ( ) <EOL> for coro in self . _coros . itervalues ( ) : <EOL> logger . debug ( '<STR_LIT>' , coro . _name , coro . _id , <EOL> '<STR_LIT>' if coro . _daemon else '<STR_LIT>' ) <EOL> AsynCoro . __cur_coro = coro <EOL> coro . _state = AsynCoro . _Running <EOL> while coro . _generator : <EOL> try : <EOL> coro . _generator . close ( ) <EOL> except : <EOL> logger . warning ( '<STR_LIT>' , <EOL> coro . _generator . __name__ , traceback . format_exc ( ) ) <EOL> if coro . _callers : <EOL> coro . _generator , coro . _value = coro . _callers . pop ( - <NUM_LIT:1> ) <EOL> else : <EOL> coro . _generator = None <EOL> if coro . _complete : <EOL> coro . _complete . set ( ) <EOL> else : <EOL> coro . _complete = <NUM_LIT:0> <EOL> self . _scheduled = set ( ) <EOL> self . _suspended = set ( ) <EOL> self . _timeouts = [ ] <EOL> self . _coros = { } <EOL> self . _channels = { } <EOL> self . __class__ . __instance = None <EOL> self . _quit = True <EOL> self . _notifier . terminate ( ) <EOL> self . _lock . release ( ) <EOL> logger . debug ( '<STR_LIT>' , ( '<STR_LIT>' % self . _location ) if self . _location else '<STR_LIT>' ) <EOL> while self . _atexit : <EOL> func , fargs , fkwargs = self . _atexit . pop ( ) <EOL> try : <EOL> func ( * fargs , ** fkwargs ) <EOL> except : <EOL> logger . warning ( '<STR_LIT>' , func . __name__ ) <EOL> logger . warning ( traceback . format_exc ( ) ) <EOL> self . _complete . set ( ) <EOL> logger . shutdown ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> def _exit ( self , await_non_daemons ) : <EOL> """<STR_LIT>""" <EOL> if not self . _quit : <EOL> if await_non_daemons : <EOL> if len ( self . _coros ) > self . _daemons : <EOL> logger . debug ( '<STR_LIT>' , <EOL> ( len ( self . _coros ) - self . _daemons ) ) <EOL> else : <EOL> self . _lock . acquire ( ) <EOL> for coro in self . _coros . itervalues ( ) : <EOL> if not coro . _daemon : <EOL> coro . _daemon = True <EOL> self . _daemons += <NUM_LIT:1> <EOL> if len ( self . _coros ) != self . _daemons : <EOL> logger . warning ( '<STR_LIT>' , len ( self . _coros ) , self . _daemons ) <EOL> self . _complete . set ( ) <EOL> self . _lock . release ( ) <EOL> self . _complete . wait ( ) <EOL> if self . _location : <EOL> Coro ( _Peer . shutdown , timeout = <NUM_LIT:5> ) <EOL> self . _complete . wait ( ) <EOL> self . _lock . acquire ( ) <EOL> if not self . _quit : <EOL> self . _complete . clear ( ) <EOL> self . _quit = True <EOL> heappush ( self . _timeouts , ( _time ( ) + <NUM_LIT:0.1> , None , None ) ) <EOL> self . _lock . release ( ) <EOL> self . _notifier . interrupt ( ) <EOL> self . _complete . wait ( ) <EOL> else : <EOL> self . _lock . release ( ) <EOL> def finish ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _exit ( True ) <EOL> def terminate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _exit ( False ) <EOL> def join ( self , show_running = False ) : <EOL> """<STR_LIT>""" <EOL> if show_running : <EOL> self . _lock . acquire ( ) <EOL> for coro in self . _coros . itervalues ( ) : <EOL> logger . info ( '<STR_LIT>' , coro . _name , coro . _id , <EOL> '<STR_LIT>' if coro . _daemon else '<STR_LIT>' ) <EOL> self . _lock . release ( ) <EOL> self . _complete . wait ( ) <EOL> def atexit ( self , func , * fargs , ** fkwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _atexit . append ( ( func , fargs , fkwargs ) ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' <EOL> class AsyncThreadPool ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , num_threads ) : <EOL> self . _num_threads = num_threads <EOL> self . _task_queue = queue . Queue ( ) <EOL> for n in xrange ( num_threads ) : <EOL> tasklet = threading . Thread ( target = self . _tasklet ) <EOL> tasklet . daemon = True <EOL> tasklet . start ( ) <EOL> def _tasklet ( self ) : <EOL> while <NUM_LIT:1> : <EOL> item = self . _task_queue . get ( block = True ) <EOL> if item is None : <EOL> self . _task_queue . task_done ( ) <EOL> break <EOL> coro , target , args , kwargs = item <EOL> try : <EOL> val = target ( * args , ** kwargs ) <EOL> coro . _proceed_ ( val ) <EOL> except : <EOL> coro . throw ( * sys . exc_info ( ) ) <EOL> finally : <EOL> self . _task_queue . task_done ( ) <EOL> def async_task ( self , target , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> coro = AsynCoro . cur_coro ( ) <EOL> if not args and kwargs : <EOL> args = kwargs . pop ( '<STR_LIT:args>' , ( ) ) <EOL> kwargs = kwargs . pop ( '<STR_LIT>' , kwargs ) <EOL> coro . _await_ ( ) <EOL> self . _task_queue . put ( ( coro , target , args , kwargs ) ) <EOL> def join ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _task_queue . join ( ) <EOL> def terminate ( self ) : <EOL> """<STR_LIT>""" <EOL> for n in xrange ( self . _num_threads ) : <EOL> self . _task_queue . put ( None ) <EOL> self . _task_queue . join ( ) <EOL> class AsyncDBCursor ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , thread_pool , cursor ) : <EOL> self . _thread_pool = thread_pool <EOL> self . _cursor = cursor <EOL> self . _sem = Semaphore ( ) <EOL> def __getattr__ ( self , name ) : <EOL> return getattr ( self . _cursor , name ) <EOL> def _exec_task ( self , func ) : <EOL> try : <EOL> return func ( ) <EOL> finally : <EOL> self . _sem . release ( ) <EOL> def execute ( self , query , args = None ) : <EOL> """<STR_LIT>""" <EOL> yield self . _sem . acquire ( ) <EOL> self . _thread_pool . async_task ( self . _exec_task , <EOL> functools . partial ( self . _cursor . execute , query , args ) ) <EOL> def executemany ( self , query , args ) : <EOL> """<STR_LIT>""" <EOL> yield self . _sem . acquire ( ) <EOL> self . _thread_pool . async_task ( self . _exec_task , <EOL> functools . partial ( self . _cursor . executemany , query , args ) ) <EOL> def callproc ( self , proc , args = ( ) ) : <EOL> """<STR_LIT>""" <EOL> yield self . _sem . acquire ( ) <EOL> self . _thread_pool . async_task ( self . _exec_task , <EOL> functools . partial ( self . _cursor . callproc , proc , args ) ) </s>
