<s> from twisted . internet import defer <EOL> from . persistence import TransactionActions <EOL> from . units import Transaction <EOL> from synapse . api . errors import HttpResponseException <EOL> from synapse . util . logutils import log_function <EOL> from synapse . util . logcontext import PreserveLoggingContext <EOL> from synapse . util . retryutils import ( <EOL> get_retry_limiter , NotRetryingDestination , <EOL> ) <EOL> import synapse . metrics <EOL> import logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> metrics = synapse . metrics . get_metrics_for ( __name__ ) <EOL> class TransactionQueue ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , hs , transport_layer ) : <EOL> self . server_name = hs . hostname <EOL> self . store = hs . get_datastore ( ) <EOL> self . transaction_actions = TransactionActions ( self . store ) <EOL> self . transport_layer = transport_layer <EOL> self . _clock = hs . get_clock ( ) <EOL> self . pending_transactions = { } <EOL> metrics . register_callback ( <EOL> "<STR_LIT>" , <EOL> lambda : len ( self . pending_transactions ) , <EOL> ) <EOL> self . pending_pdus_by_dest = pdus = { } <EOL> self . pending_edus_by_dest = edus = { } <EOL> metrics . register_callback ( <EOL> "<STR_LIT>" , <EOL> lambda : sum ( map ( len , pdus . values ( ) ) ) , <EOL> ) <EOL> metrics . register_callback ( <EOL> "<STR_LIT>" , <EOL> lambda : sum ( map ( len , edus . values ( ) ) ) , <EOL> ) <EOL> self . pending_failures_by_dest = { } <EOL> self . _next_txn_id = int ( self . _clock . time_msec ( ) ) <EOL> def can_send_to ( self , destination ) : <EOL> """<STR_LIT>""" <EOL> if destination == self . server_name : <EOL> return False <EOL> if self . server_name . startswith ( "<STR_LIT:localhost>" ) : <EOL> return destination . startswith ( "<STR_LIT:localhost>" ) <EOL> else : <EOL> return not destination . startswith ( "<STR_LIT:localhost>" ) <EOL> def enqueue_pdu ( self , pdu , destinations , order ) : <EOL> destinations = set ( destinations ) <EOL> destinations = set ( <EOL> dest for dest in destinations if self . can_send_to ( dest ) <EOL> ) <EOL> logger . debug ( "<STR_LIT>" , str ( destinations ) ) <EOL> if not destinations : <EOL> return <EOL> deferreds = [ ] <EOL> for destination in destinations : <EOL> deferred = defer . Deferred ( ) <EOL> self . pending_pdus_by_dest . setdefault ( destination , [ ] ) . append ( <EOL> ( pdu , deferred , order ) <EOL> ) <EOL> def chain ( failure ) : <EOL> if not deferred . called : <EOL> deferred . errback ( failure ) <EOL> def log_failure ( f ) : <EOL> logger . warn ( "<STR_LIT>" , destination , f . value ) <EOL> deferred . addErrback ( log_failure ) <EOL> with PreserveLoggingContext ( ) : <EOL> self . _attempt_new_transaction ( destination ) . addErrback ( chain ) <EOL> deferreds . append ( deferred ) <EOL> def enqueue_edu ( self , edu ) : <EOL> destination = edu . destination <EOL> if not self . can_send_to ( destination ) : <EOL> return <EOL> deferred = defer . Deferred ( ) <EOL> self . pending_edus_by_dest . setdefault ( destination , [ ] ) . append ( <EOL> ( edu , deferred ) <EOL> ) <EOL> def chain ( failure ) : <EOL> if not deferred . called : <EOL> deferred . errback ( failure ) <EOL> def log_failure ( f ) : <EOL> logger . warn ( "<STR_LIT>" , destination , f . value ) <EOL> deferred . addErrback ( log_failure ) <EOL> with PreserveLoggingContext ( ) : <EOL> self . _attempt_new_transaction ( destination ) . addErrback ( chain ) <EOL> return deferred <EOL> @ defer . inlineCallbacks <EOL> def enqueue_failure ( self , failure , destination ) : <EOL> if destination == self . server_name or destination == "<STR_LIT:localhost>" : <EOL> return <EOL> deferred = defer . Deferred ( ) <EOL> if not self . can_send_to ( destination ) : <EOL> return <EOL> self . pending_failures_by_dest . setdefault ( <EOL> destination , [ ] <EOL> ) . append ( <EOL> ( failure , deferred ) <EOL> ) <EOL> def chain ( f ) : <EOL> if not deferred . called : <EOL> deferred . errback ( f ) <EOL> def log_failure ( f ) : <EOL> logger . warn ( "<STR_LIT>" , destination , f . value ) <EOL> deferred . addErrback ( log_failure ) <EOL> with PreserveLoggingContext ( ) : <EOL> self . _attempt_new_transaction ( destination ) . addErrback ( chain ) <EOL> yield deferred <EOL> @ defer . inlineCallbacks <EOL> @ log_function <EOL> def _attempt_new_transaction ( self , destination ) : <EOL> if destination in self . pending_transactions : <EOL> logger . debug ( <EOL> "<STR_LIT>" , <EOL> destination <EOL> ) <EOL> return <EOL> pending_pdus = self . pending_pdus_by_dest . pop ( destination , [ ] ) <EOL> pending_edus = self . pending_edus_by_dest . pop ( destination , [ ] ) <EOL> pending_failures = self . pending_failures_by_dest . pop ( destination , [ ] ) <EOL> if pending_pdus : <EOL> logger . debug ( "<STR_LIT>" , <EOL> destination , len ( pending_pdus ) ) <EOL> if not pending_pdus and not pending_edus and not pending_failures : <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> return <EOL> try : <EOL> self . pending_transactions [ destination ] = <NUM_LIT:1> <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> pending_pdus . sort ( key = lambda t : t [ <NUM_LIT:2> ] ) <EOL> pdus = [ x [ <NUM_LIT:0> ] for x in pending_pdus ] <EOL> edus = [ x [ <NUM_LIT:0> ] for x in pending_edus ] <EOL> failures = [ x [ <NUM_LIT:0> ] . get_dict ( ) for x in pending_failures ] <EOL> deferreds = [ <EOL> x [ <NUM_LIT:1> ] <EOL> for x in pending_pdus + pending_edus + pending_failures <EOL> ] <EOL> txn_id = str ( self . _next_txn_id ) <EOL> limiter = yield get_retry_limiter ( <EOL> destination , <EOL> self . _clock , <EOL> self . store , <EOL> ) <EOL> logger . debug ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> destination , txn_id , <EOL> len ( pending_pdus ) , <EOL> len ( pending_edus ) , <EOL> len ( pending_failures ) <EOL> ) <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> transaction = Transaction . create_new ( <EOL> origin_server_ts = int ( self . _clock . time_msec ( ) ) , <EOL> transaction_id = txn_id , <EOL> origin = self . server_name , <EOL> destination = destination , <EOL> pdus = pdus , <EOL> edus = edus , <EOL> pdu_failures = failures , <EOL> ) <EOL> self . _next_txn_id += <NUM_LIT:1> <EOL> yield self . transaction_actions . prepare_to_send ( transaction ) <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> logger . info ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> destination , txn_id , <EOL> transaction . transaction_id , <EOL> len ( pending_pdus ) , <EOL> len ( pending_edus ) , <EOL> len ( pending_failures ) , <EOL> ) <EOL> with limiter : <EOL> def json_data_cb ( ) : <EOL> data = transaction . get_dict ( ) <EOL> now = int ( self . _clock . time_msec ( ) ) <EOL> if "<STR_LIT>" in data : <EOL> for p in data [ "<STR_LIT>" ] : <EOL> if "<STR_LIT>" in p : <EOL> unsigned = p . setdefault ( "<STR_LIT>" , { } ) <EOL> unsigned [ "<STR_LIT>" ] = now - int ( p [ "<STR_LIT>" ] ) <EOL> del p [ "<STR_LIT>" ] <EOL> return data <EOL> try : <EOL> response = yield self . transport_layer . send_transaction ( <EOL> transaction , json_data_cb <EOL> ) <EOL> code = <NUM_LIT:200> <EOL> if response : <EOL> for e_id , r in response . get ( "<STR_LIT>" , { } ) . items ( ) : <EOL> if "<STR_LIT:error>" in r : <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> e_id , r , <EOL> ) <EOL> except HttpResponseException as e : <EOL> code = e . code <EOL> response = e . response <EOL> logger . info ( <EOL> "<STR_LIT>" , <EOL> destination , txn_id , code <EOL> ) <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> yield self . transaction_actions . delivered ( <EOL> transaction , code , response <EOL> ) <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> for deferred in deferreds : <EOL> if code == <NUM_LIT:200> : <EOL> deferred . callback ( None ) <EOL> else : <EOL> deferred . errback ( RuntimeError ( "<STR_LIT>" % code ) ) <EOL> try : <EOL> yield deferred <EOL> except : <EOL> pass <EOL> logger . debug ( "<STR_LIT>" , destination ) <EOL> except NotRetryingDestination : <EOL> logger . info ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> destination , <EOL> ) <EOL> except RuntimeError as e : <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> destination , <EOL> e , <EOL> ) <EOL> except Exception as e : <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> destination , <EOL> e , <EOL> ) <EOL> for deferred in deferreds : <EOL> if not deferred . called : <EOL> deferred . errback ( e ) <EOL> finally : <EOL> self . pending_transactions . pop ( destination , None ) <EOL> self . _attempt_new_transaction ( destination ) </s>
