<s> """<STR_LIT>""" <EOL> import logging <EOL> import pycurl <EOL> import threading <EOL> from cStringIO import StringIO <EOL> from ganeti import http <EOL> from ganeti import compat <EOL> from ganeti import netutils <EOL> from ganeti import locking <EOL> class HttpClientRequest ( object ) : <EOL> def __init__ ( self , host , port , method , path , headers = None , post_data = None , <EOL> read_timeout = None , curl_config_fn = None , nicename = None , <EOL> completion_cb = None ) : <EOL> """<STR_LIT>""" <EOL> assert path . startswith ( "<STR_LIT:/>" ) , "<STR_LIT>" <EOL> assert curl_config_fn is None or callable ( curl_config_fn ) <EOL> assert completion_cb is None or callable ( completion_cb ) <EOL> self . host = host <EOL> self . port = port <EOL> self . method = method <EOL> self . path = path <EOL> self . read_timeout = read_timeout <EOL> self . curl_config_fn = curl_config_fn <EOL> self . nicename = nicename <EOL> self . completion_cb = completion_cb <EOL> if post_data is None : <EOL> self . post_data = "<STR_LIT>" <EOL> else : <EOL> self . post_data = post_data <EOL> if headers is None : <EOL> self . headers = [ ] <EOL> elif isinstance ( headers , dict ) : <EOL> self . headers = [ "<STR_LIT>" % ( name , value ) <EOL> for name , value in headers . items ( ) ] <EOL> else : <EOL> self . headers = headers <EOL> self . success = None <EOL> self . error = None <EOL> self . resp_status_code = None <EOL> self . resp_body = None <EOL> def __repr__ ( self ) : <EOL> status = [ "<STR_LIT>" % ( self . __class__ . __module__ , self . __class__ . __name__ ) , <EOL> "<STR_LIT>" % ( self . host , self . port ) , <EOL> self . method , <EOL> self . path ] <EOL> return "<STR_LIT>" % ( "<STR_LIT:U+0020>" . join ( status ) , id ( self ) ) <EOL> @ property <EOL> def url ( self ) : <EOL> """<STR_LIT>""" <EOL> if netutils . IPAddress . IsValid ( self . host ) : <EOL> address = netutils . FormatAddress ( ( self . host , self . port ) ) <EOL> else : <EOL> address = "<STR_LIT>" % ( self . host , self . port ) <EOL> return "<STR_LIT>" % ( address , self . path ) <EOL> def _StartRequest ( curl , req ) : <EOL> """<STR_LIT>""" <EOL> logging . debug ( "<STR_LIT>" , req ) <EOL> url = req . url <EOL> method = req . method <EOL> post_data = req . post_data <EOL> headers = req . headers <EOL> assert isinstance ( method , str ) <EOL> assert isinstance ( url , str ) <EOL> assert isinstance ( post_data , str ) <EOL> assert compat . all ( isinstance ( i , str ) for i in headers ) <EOL> resp_buffer = StringIO ( ) <EOL> curl . setopt ( pycurl . VERBOSE , False ) <EOL> curl . setopt ( pycurl . NOSIGNAL , True ) <EOL> curl . setopt ( pycurl . USERAGENT , http . HTTP_GANETI_VERSION ) <EOL> curl . setopt ( pycurl . PROXY , "<STR_LIT>" ) <EOL> curl . setopt ( pycurl . CUSTOMREQUEST , str ( method ) ) <EOL> curl . setopt ( pycurl . URL , url ) <EOL> curl . setopt ( pycurl . POSTFIELDS , post_data ) <EOL> curl . setopt ( pycurl . HTTPHEADER , headers ) <EOL> if req . read_timeout is None : <EOL> curl . setopt ( pycurl . TIMEOUT , <NUM_LIT:0> ) <EOL> else : <EOL> curl . setopt ( pycurl . TIMEOUT , int ( req . read_timeout ) ) <EOL> if hasattr ( pycurl , "<STR_LIT>" ) : <EOL> curl . setopt ( pycurl . SSL_SESSIONID_CACHE , False ) <EOL> curl . setopt ( pycurl . WRITEFUNCTION , resp_buffer . write ) <EOL> if req . curl_config_fn : <EOL> req . curl_config_fn ( curl ) <EOL> return _PendingRequest ( curl , req , resp_buffer . getvalue ) <EOL> class _PendingRequest ( object ) : <EOL> def __init__ ( self , curl , req , resp_buffer_read ) : <EOL> """<STR_LIT>""" <EOL> assert req . success is None <EOL> self . _curl = curl <EOL> self . _req = req <EOL> self . _resp_buffer_read = resp_buffer_read <EOL> def GetCurlHandle ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _curl <EOL> def GetCurrentRequest ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _req <EOL> def Done ( self , errmsg ) : <EOL> """<STR_LIT>""" <EOL> curl = self . _curl <EOL> req = self . _req <EOL> assert req . success is None , "<STR_LIT>" <EOL> try : <EOL> from_str = "<STR_LIT>" % ( <EOL> curl . getinfo ( pycurl . LOCAL_IP ) , <EOL> curl . getinfo ( pycurl . LOCAL_PORT ) <EOL> ) <EOL> except AttributeError : <EOL> from_str = "<STR_LIT>" <EOL> logging . debug ( "<STR_LIT>" , from_str , req , errmsg ) <EOL> req . success = not bool ( errmsg ) <EOL> req . error = errmsg <EOL> req . resp_status_code = curl . getinfo ( pycurl . RESPONSE_CODE ) <EOL> req . resp_body = self . _resp_buffer_read ( ) <EOL> curl . setopt ( pycurl . POSTFIELDS , "<STR_LIT>" ) <EOL> curl . setopt ( pycurl . WRITEFUNCTION , lambda _ : None ) <EOL> if req . completion_cb : <EOL> req . completion_cb ( req ) <EOL> class _NoOpRequestMonitor ( object ) : <EOL> """<STR_LIT>""" <EOL> @ staticmethod <EOL> def acquire ( * args , ** kwargs ) : <EOL> pass <EOL> release = acquire <EOL> Disable = acquire <EOL> class _PendingRequestMonitor ( object ) : <EOL> _LOCK = "<STR_LIT>" <EOL> def __init__ ( self , owner , pending_fn ) : <EOL> """<STR_LIT>""" <EOL> self . _owner = owner <EOL> self . _pending_fn = pending_fn <EOL> self . _lock = locking . SharedLock ( "<STR_LIT>" ) <EOL> self . acquire = self . _lock . acquire <EOL> self . release = self . _lock . release <EOL> @ locking . ssynchronized ( _LOCK ) <EOL> def Disable ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _pending_fn = None <EOL> @ locking . ssynchronized ( _LOCK , shared = <NUM_LIT:1> ) <EOL> def GetLockInfo ( self , requested ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> if self . _pending_fn : <EOL> owner_name = self . _owner . getName ( ) <EOL> for client in self . _pending_fn ( ) : <EOL> req = client . GetCurrentRequest ( ) <EOL> if req : <EOL> if req . nicename is None : <EOL> name = "<STR_LIT>" % ( req . host , req . path ) <EOL> else : <EOL> name = req . nicename <EOL> result . append ( ( "<STR_LIT>" % name , None , [ owner_name ] , None ) ) <EOL> return result <EOL> def _ProcessCurlRequests ( multi , requests ) : <EOL> """<STR_LIT>""" <EOL> for curl in requests : <EOL> multi . add_handle ( curl ) <EOL> while True : <EOL> ( ret , active ) = multi . perform ( ) <EOL> assert ret in ( pycurl . E_MULTI_OK , pycurl . E_CALL_MULTI_PERFORM ) <EOL> if ret == pycurl . E_CALL_MULTI_PERFORM : <EOL> continue <EOL> while True : <EOL> ( remaining_messages , successful , failed ) = multi . info_read ( ) <EOL> for curl in successful : <EOL> multi . remove_handle ( curl ) <EOL> yield ( curl , None ) <EOL> for curl , errnum , errmsg in failed : <EOL> multi . remove_handle ( curl ) <EOL> yield ( curl , "<STR_LIT>" % ( errnum , errmsg ) ) <EOL> if remaining_messages == <NUM_LIT:0> : <EOL> break <EOL> if active == <NUM_LIT:0> : <EOL> break <EOL> multi . select ( <NUM_LIT:1.0> ) <EOL> def ProcessRequests ( requests , lock_monitor_cb = None , _curl = pycurl . Curl , <EOL> _curl_multi = pycurl . CurlMulti , <EOL> _curl_process = _ProcessCurlRequests ) : <EOL> """<STR_LIT>""" <EOL> assert compat . all ( ( req . error is None and <EOL> req . success is None and <EOL> req . resp_status_code is None and <EOL> req . resp_body is None ) <EOL> for req in requests ) <EOL> curl_to_client = dict ( ( client . GetCurlHandle ( ) , client ) <EOL> for client in map ( lambda req : _StartRequest ( _curl ( ) , req ) , requests ) ) <EOL> assert len ( curl_to_client ) == len ( requests ) <EOL> if lock_monitor_cb : <EOL> monitor = _PendingRequestMonitor ( threading . currentThread ( ) , <EOL> curl_to_client . values ) <EOL> lock_monitor_cb ( monitor ) <EOL> else : <EOL> monitor = _NoOpRequestMonitor <EOL> for ( curl , msg ) in _curl_process ( _curl_multi ( ) , curl_to_client . keys ( ) ) : <EOL> monitor . acquire ( shared = <NUM_LIT:0> ) <EOL> try : <EOL> curl_to_client . pop ( curl ) . Done ( msg ) <EOL> finally : <EOL> monitor . release ( ) <EOL> assert not curl_to_client , "<STR_LIT>" <EOL> monitor . Disable ( ) <EOL> assert compat . all ( req . error is not None or <EOL> ( req . success and <EOL> req . resp_status_code is not None and <EOL> req . resp_body is not None ) <EOL> for req in requests ) </s>
