<s> import os <EOL> import threading <EOL> import traceback <EOL> from ovs . db import idl <EOL> from ovs import poller <EOL> import retrying <EOL> from six . moves import queue as Queue <EOL> from neutron . agent . ovsdb . native import helpers <EOL> from neutron . agent . ovsdb . native import idlutils <EOL> class TransactionQueue ( Queue . Queue , object ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( TransactionQueue , self ) . __init__ ( * args , ** kwargs ) <EOL> alertpipe = os . pipe ( ) <EOL> self . alertin = os . fdopen ( alertpipe [ <NUM_LIT:0> ] , '<STR_LIT:r>' , <NUM_LIT:0> ) <EOL> self . alertout = os . fdopen ( alertpipe [ <NUM_LIT:1> ] , '<STR_LIT:w>' , <NUM_LIT:0> ) <EOL> def get_nowait ( self , * args , ** kwargs ) : <EOL> try : <EOL> result = super ( TransactionQueue , self ) . get_nowait ( * args , ** kwargs ) <EOL> except Queue . Empty : <EOL> return None <EOL> self . alertin . read ( <NUM_LIT:1> ) <EOL> return result <EOL> def put ( self , * args , ** kwargs ) : <EOL> super ( TransactionQueue , self ) . put ( * args , ** kwargs ) <EOL> self . alertout . write ( '<STR_LIT:X>' ) <EOL> self . alertout . flush ( ) <EOL> @ property <EOL> def alert_fileno ( self ) : <EOL> return self . alertin . fileno ( ) <EOL> class Connection ( object ) : <EOL> def __init__ ( self , connection , timeout , schema_name ) : <EOL> self . idl = None <EOL> self . connection = connection <EOL> self . timeout = timeout <EOL> self . txns = TransactionQueue ( <NUM_LIT:1> ) <EOL> self . lock = threading . Lock ( ) <EOL> self . schema_name = schema_name <EOL> def start ( self , table_name_list = None ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> if self . idl is not None : <EOL> return <EOL> try : <EOL> helper = idlutils . get_schema_helper ( self . connection , <EOL> self . schema_name ) <EOL> except Exception : <EOL> helpers . enable_connection_uri ( self . connection ) <EOL> @ retrying . retry ( wait_exponential_multiplier = <NUM_LIT:10> , <EOL> stop_max_delay = <NUM_LIT:1000> ) <EOL> def do_get_schema_helper ( ) : <EOL> return idlutils . get_schema_helper ( self . connection , <EOL> self . schema_name ) <EOL> helper = do_get_schema_helper ( ) <EOL> if table_name_list is None : <EOL> helper . register_all ( ) <EOL> else : <EOL> for table_name in table_name_list : <EOL> helper . register_table ( table_name ) <EOL> self . idl = idl . Idl ( self . connection , helper ) <EOL> idlutils . wait_for_change ( self . idl , self . timeout ) <EOL> self . poller = poller . Poller ( ) <EOL> self . thread = threading . Thread ( target = self . run ) <EOL> self . thread . setDaemon ( True ) <EOL> self . thread . start ( ) <EOL> def run ( self ) : <EOL> while True : <EOL> self . idl . wait ( self . poller ) <EOL> self . poller . fd_wait ( self . txns . alert_fileno , poller . POLLIN ) <EOL> self . poller . block ( ) <EOL> self . idl . run ( ) <EOL> txn = self . txns . get_nowait ( ) <EOL> if txn is not None : <EOL> try : <EOL> txn . results . put ( txn . do_commit ( ) ) <EOL> except Exception as ex : <EOL> er = idlutils . ExceptionResult ( ex = ex , <EOL> tb = traceback . format_exc ( ) ) <EOL> txn . results . put ( er ) <EOL> self . txns . task_done ( ) <EOL> def queue_txn ( self , txn ) : <EOL> self . txns . put ( txn ) </s>
