<s> import _mysql <EOL> import sys <EOL> from threading import Thread , Lock <EOL> from memsql_loader . db import connection_wrapper , pool <EOL> from memsql_loader . db . load_data import LoadDataStmt <EOL> from memsql_loader . util import db_utils , log <EOL> from memsql_loader . execution . errors import WorkerException , ConnectionException <EOL> class Loader ( Thread ) : <EOL> def load ( self , job , task , fifo , db_connection ) : <EOL> self . _job = job <EOL> self . _task = task <EOL> self . _fifo = fifo <EOL> self . _conn = db_connection <EOL> load_data = LoadDataStmt ( job , task . file_id , fifo . path ) <EOL> self . _sql , self . _params = load_data . build ( ) <EOL> self . _error = None <EOL> self . _tb = None <EOL> self . _active_conn_id = None <EOL> self . _conn_lock = Lock ( ) <EOL> self . _fifo . attach_reader ( self . abort ) <EOL> @ property <EOL> def error ( self ) : <EOL> return self . _error <EOL> @ property <EOL> def traceback ( self ) : <EOL> return self . _tb <EOL> def abort ( self ) : <EOL> with self . _conn_lock : <EOL> if self . _active_conn_id is not None : <EOL> try : <EOL> with pool . get_connection ( database = '<STR_LIT>' , ** self . _job . spec . connection ) as conn : <EOL> db_utils . try_kill_connection ( conn , self . _active_conn_id ) <EOL> except pool . PoolConnectionException : <EOL> pass <EOL> return True <EOL> return False <EOL> def run ( self ) : <EOL> self . logger = log . get_logger ( '<STR_LIT>' ) <EOL> try : <EOL> _mysql . connect ( unix_socket = '<STR_LIT:.>' ) <EOL> except _mysql . MySQLError : <EOL> pass <EOL> try : <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> try : <EOL> with self . _conn_lock : <EOL> self . _active_conn_id = self . _conn . thread_id ( ) <EOL> with self . _task . protect ( ) : <EOL> self . _task . data [ '<STR_LIT>' ] = self . _active_conn_id <EOL> self . _task . save ( ) <EOL> row_count = self . _conn . query ( self . _sql , * self . _params ) <EOL> finally : <EOL> with self . _conn_lock : <EOL> self . _active_conn_id = None <EOL> with self . _task . protect ( ) : <EOL> self . _task . data [ '<STR_LIT>' ] = row_count <EOL> self . _task . save ( ) <EOL> except connection_wrapper . ConnectionWrapperException as e : <EOL> self . logger . error ( '<STR_LIT>' , str ( e ) ) <EOL> self . _set_error ( ConnectionException ( str ( e ) ) ) <EOL> except pool . MySQLError as e : <EOL> errno , msg = e . args <EOL> msg = "<STR_LIT>" % ( errno , msg ) <EOL> self . logger . error ( msg ) <EOL> self . _set_error ( WorkerException ( msg ) ) <EOL> except Exception as e : <EOL> self . _set_error ( e ) <EOL> except KeyboardInterrupt : <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> finally : <EOL> self . _fifo . detach_reader ( ) <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> def _set_error ( self , err ) : <EOL> self . _error = err <EOL> self . _tb = sys . exc_info ( ) [ <NUM_LIT:2> ] </s>
