<s> from twisted . internet import reactor <EOL> from errno import ECONNRESET , ENOTCONN , ESHUTDOWN , EWOULDBLOCK , ENOBUFS , EAGAIN , EINTR <EOL> from datetime import datetime <EOL> from time import sleep <EOL> from threading import Thread , Condition <EOL> import socket <EOL> import sys , traceback <EOL> MAX_WORKERS = <NUM_LIT:30> <EOL> class AsyncSender ( Thread ) : <EOL> userv = None <EOL> def __init__ ( self , userv ) : <EOL> Thread . __init__ ( self ) <EOL> self . userv = userv <EOL> self . setDaemon ( True ) <EOL> self . start ( ) <EOL> def run ( self ) : <EOL> while True : <EOL> self . userv . wi_available . acquire ( ) <EOL> while len ( self . userv . wi ) == <NUM_LIT:0> : <EOL> self . userv . wi_available . wait ( ) <EOL> wi = self . userv . wi . pop ( <NUM_LIT:0> ) <EOL> if wi == None : <EOL> self . userv . wi . append ( None ) <EOL> self . userv . wi_available . notify ( ) <EOL> self . userv . wi_available . release ( ) <EOL> if wi == None : <EOL> break <EOL> data , address = wi <EOL> try : <EOL> ai = socket . getaddrinfo ( address [ <NUM_LIT:0> ] , None , self . userv . family ) <EOL> except : <EOL> continue <EOL> if self . userv . family == socket . AF_INET : <EOL> address = ( ai [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] [ <NUM_LIT:0> ] , address [ <NUM_LIT:1> ] ) <EOL> else : <EOL> address = ( ai [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] [ <NUM_LIT:0> ] , address [ <NUM_LIT:1> ] , ai [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] [ <NUM_LIT:2> ] , ai [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] [ <NUM_LIT:3> ] ) <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:20> ) : <EOL> try : <EOL> if self . userv . skt . sendto ( data , address ) == len ( data ) : <EOL> break <EOL> except socket . error , why : <EOL> if why [ <NUM_LIT:0> ] not in ( EWOULDBLOCK , ENOBUFS , EAGAIN ) : <EOL> break <EOL> sleep ( <NUM_LIT> ) <EOL> self . userv = None <EOL> class AsyncReceiver ( Thread ) : <EOL> userv = None <EOL> def __init__ ( self , userv ) : <EOL> Thread . __init__ ( self ) <EOL> self . userv = userv <EOL> self . setDaemon ( True ) <EOL> self . start ( ) <EOL> def run ( self ) : <EOL> while True : <EOL> try : <EOL> data , address = self . userv . skt . recvfrom ( <NUM_LIT> ) <EOL> if not data : <EOL> break <EOL> except Exception , why : <EOL> if isinstance ( why , socket . error ) and why [ <NUM_LIT:0> ] in ( ECONNRESET , ENOTCONN , ESHUTDOWN ) : <EOL> break <EOL> if isinstance ( why , socket . error ) and why [ <NUM_LIT:0> ] in ( EINTR , ) : <EOL> continue <EOL> else : <EOL> print datetime . now ( ) , '<STR_LIT>' <EOL> print '<STR_LIT:->' * <NUM_LIT> <EOL> traceback . print_exc ( file = sys . stdout ) <EOL> print '<STR_LIT:->' * <NUM_LIT> <EOL> sys . stdout . flush ( ) <EOL> sleep ( <NUM_LIT:1> ) <EOL> continue <EOL> if self . userv . family == socket . AF_INET6 : <EOL> address = ( '<STR_LIT>' % address [ <NUM_LIT:0> ] , address [ <NUM_LIT:1> ] ) <EOL> reactor . callFromThread ( self . userv . handle_read , data , address ) <EOL> self . userv = None <EOL> class Udp_server ( object ) : <EOL> skt = None <EOL> family = None <EOL> data_callback = None <EOL> laddress = None <EOL> sendqueue = None <EOL> stats = None <EOL> wi_available = None <EOL> wi = None <EOL> def __init__ ( self , address , data_callback , family = None ) : <EOL> self . laddress = address <EOL> if family == None : <EOL> if address != None and address [ <NUM_LIT:0> ] . startswith ( '<STR_LIT:[>' ) : <EOL> family = socket . AF_INET6 <EOL> address = ( address [ <NUM_LIT:0> ] [ <NUM_LIT:1> : - <NUM_LIT:1> ] , address [ <NUM_LIT:1> ] ) <EOL> else : <EOL> family = socket . AF_INET <EOL> self . family = family <EOL> self . skt = socket . socket ( family , socket . SOCK_DGRAM ) <EOL> if address != None : <EOL> ai = socket . getaddrinfo ( address [ <NUM_LIT:0> ] , None , family ) <EOL> if family == socket . AF_INET : <EOL> address = ( ai [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] [ <NUM_LIT:0> ] , address [ <NUM_LIT:1> ] ) <EOL> else : <EOL> address = ( ai [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] [ <NUM_LIT:0> ] , address [ <NUM_LIT:1> ] , ai [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] [ <NUM_LIT:2> ] , ai [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] [ <NUM_LIT:3> ] ) <EOL> self . skt . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> self . skt . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEPORT , <NUM_LIT:1> ) <EOL> self . skt . bind ( address ) <EOL> self . data_callback = data_callback <EOL> self . sendqueue = [ ] <EOL> self . stats = [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] <EOL> self . wi_available = Condition ( ) <EOL> self . wi = [ ] <EOL> for i in range ( <NUM_LIT:0> , MAX_WORKERS ) : <EOL> AsyncSender ( self ) <EOL> AsyncReceiver ( self ) <EOL> def send_to ( self , data , address ) : <EOL> if self . family == socket . AF_INET6 : <EOL> if not address [ <NUM_LIT:0> ] . startswith ( '<STR_LIT:[>' ) : <EOL> raise Exception ( '<STR_LIT>' % address [ <NUM_LIT:0> ] ) <EOL> address = ( address [ <NUM_LIT:0> ] [ <NUM_LIT:1> : - <NUM_LIT:1> ] , address [ <NUM_LIT:1> ] ) <EOL> self . wi_available . acquire ( ) <EOL> self . wi . append ( ( data , address ) ) <EOL> self . wi_available . notify ( ) <EOL> self . wi_available . release ( ) <EOL> def handle_read ( self , data , address ) : <EOL> self . stats [ <NUM_LIT:2> ] += <NUM_LIT:1> <EOL> try : <EOL> self . data_callback ( data , address , self ) <EOL> except : <EOL> print datetime . now ( ) , '<STR_LIT>' <EOL> print '<STR_LIT:->' * <NUM_LIT> <EOL> traceback . print_exc ( file = sys . stdout ) <EOL> print '<STR_LIT:->' * <NUM_LIT> <EOL> sys . stdout . flush ( ) <EOL> def shutdown ( self ) : <EOL> self . skt . shutdown ( socket . SHUT_RDWR ) <EOL> self . wi_available . acquire ( ) <EOL> self . wi . append ( None ) <EOL> self . wi_available . notify ( ) <EOL> self . wi_available . release ( ) <EOL> self . data_callback = None <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> from sys import exit <EOL> npongs = <NUM_LIT:2> <EOL> def ping_received ( data , address , udp_server ) : <EOL> print '<STR_LIT>' <EOL> if not ( data == '<STR_LIT>' and address == ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) : <EOL> exit ( <NUM_LIT:1> ) <EOL> udp_server . send_to ( '<STR_LIT>' , address ) <EOL> def pong_received ( data , address , udp_server ) : <EOL> print '<STR_LIT>' <EOL> if not ( data == '<STR_LIT>' and address == ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) : <EOL> exit ( <NUM_LIT:1> ) <EOL> global npongs <EOL> npongs -= <NUM_LIT:1> <EOL> if npongs == <NUM_LIT:0> : <EOL> reactor . stop ( ) <EOL> def ping_received6 ( data , address , udp_server ) : <EOL> print '<STR_LIT>' , address <EOL> if not ( data == '<STR_LIT>' and address == ( '<STR_LIT>' , <NUM_LIT> ) ) : <EOL> exit ( <NUM_LIT:1> ) <EOL> udp_server . send_to ( '<STR_LIT>' , address ) <EOL> def pong_received6 ( data , address , udp_server ) : <EOL> print '<STR_LIT>' , address <EOL> if not ( data == '<STR_LIT>' and address == ( '<STR_LIT>' , <NUM_LIT> ) ) : <EOL> exit ( <NUM_LIT:1> ) <EOL> global npongs <EOL> npongs -= <NUM_LIT:1> <EOL> if npongs == <NUM_LIT:0> : <EOL> reactor . stop ( ) <EOL> udp_server_ping = Udp_server ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) , ping_received ) <EOL> udp_server_pong = Udp_server ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) , pong_received ) <EOL> udp_server_pong . send_to ( '<STR_LIT>' , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) <EOL> udp_server_ping6 = Udp_server ( ( '<STR_LIT>' , <NUM_LIT> ) , ping_received6 ) <EOL> udp_server_pong6 = Udp_server ( ( '<STR_LIT>' , <NUM_LIT> ) , pong_received6 , socket . AF_INET6 ) <EOL> udp_server_pong6 . send_to ( '<STR_LIT>' , ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> reactor . run ( ) </s>
