TASKS_TTL = 120
    QUEUED = SuperEnum.E
    RUNNING = SuperEnum.E
    SUCCESS = SuperEnum.E
    ERROR = SuperEnum.E
    CANCELLED = SuperEnum.E
    SUCCESS_CONDITION = 'tasks.result = \'success\''
    ERROR_CONDITION = 'tasks.result = \'error\''
    CANCELLED_CONDITION = 'tasks.result = \'cancelled\''
    FINISHED_CONDITION = 'tasks.finished IS NOT NULL'
    QUEUED_CONDITION = 'tasks.finished IS NULL AND (tasks.execution_id IS NULL OR tasks.last_contact <= datetime(:now, "unixepoch", "-%s second"))' % TASKS_TTL
    PROJECTION = re.sub(r'\s+', ' ', """
        (CASE
            WHEN (%s) THEN UPPER(tasks.result)
            WHEN (%s) THEN 'QUEUED'
            ELSE 'RUNNING'
        END)
    """ % (FINISHED_CONDITION, QUEUED_CONDITION)).strip()
    QUEUED = SuperEnum.E
    RUNNING = SuperEnum.E
    FINISHED = SuperEnum.E
    CANCELLED = SuperEnum.E
    PROJECTION = re.sub(r'\s+', ' ', '''
        (CASE
            WHEN (
                (job_tasks.tasks_total - job_tasks.tasks_finished) = 0
                AND job_tasks.tasks_cancelled > 0) THEN 'CANCELLED'
            WHEN (
                job_tasks.tasks_total IS NULL
                OR job_tasks.tasks_finished = job_tasks.tasks_total) THEN 'FINISHED'
            WHEN (job_tasks.tasks_queued = job_tasks.tasks_total) THEN 'QUEUED'
            ELSE 'RUNNING'
        END)
    ''').strip()
    DESC = SuperEnum.E
    ASC = SuperEnum.E
class TaskState(SuperEnum):
class JobState(SuperEnum):
class SortDirection(SuperEnum):
    def projection_params():
def task_load_row(row):
def job_load_row(row): class ConnectionWrapperException(IOError):
class ConnectionWrapper(object):
    def __init__(self, errno, message, host, port=None, database=None,
                 user=None, password=None):
    def __init__(self, host, port=3306, database="information_schema",
                 user=None, password=None, max_idle_time=7 * 3600,
                 options=None):
    def connection_info(self):
    def __enter__(self):
    def __exit__(self, exc_type, exc_value, traceback):
    def __wrap_errors(self, fn, *args, **kwargs):
        def wrapped(*args, **kwargs):
    def __potential_connection_failure(self, e):
    def __handle_connection_failure(self, e):
    def connect(self):
    def __getattr__(self, key): _POOL = connection_pool.ConnectionPool()
    _POOL = connection_pool.ConnectionPool()
def recreate_pool():
def close_connections():
def get_connection(host, port, database, user, password, pooled=True, **kwargs): class Tasks(apsw_sql_step_queue.APSWSQLStepQueue):
    def __init__(self):
        storage = LoaderStorage()
        super(Tasks, self).__init__('tasks', storage, execution_ttl=api.shared.TASKS_TTL, task_handler_class=TaskHandler)

    # NOTE: This method overrides bulk_finish on APSWSQLStepQueue so that it
    # finishes tasks even if they are currently running.
    def bulk_finish(self, result='cancelled', extra_predicate=None):
        extra_predicate_sql, extra_predicate_args = (
            self._build_extra_predicate(extra_predicate))

        with self.storage.transaction() as cursor:
            now = unix_timestamp(datetime.utcnow())
            affected_rows = apsw_helpers.query(cursor, '''
                SELECT * from %s
                WHERE
                    finished IS NULL
                    %s
            ''' % (self.table_name, extra_predicate_sql),
                **extra_predicate_args)
            apsw_helpers.query(cursor, '''
                UPDATE %s
                SET
                    execution_id = 0,
                    last_contact = datetime(:now, 'unixepoch'),
                    update_count = update_count + 1,
                    steps = '[]',
                    started = datetime(:now, 'unixepoch'),
                    finished = datetime(:now, 'unixepoch'),
                    result = :result
                WHERE
                    finished IS NULL
                    %s
            ''' % (self.table_name, extra_predicate_sql),
                now=now,
                result=result,
                **extra_predicate_args)

        return len(affected_rows)

    def get_tasks_in_state(self, state, extra_predicate=None):
        extra_predicate_sql, extra_predicate_args = (
            self._build_extra_predicate(extra_predicate))

        query_params = api.shared.TaskState.projection_params()
        if len(state) == 1:
            state_list = "('" + str(state[0]) + "')"
        else:
            state_list = str(tuple(str(v) for v in state ))
        query_params.update(extra_predicate_args)
        with self.storage.cursor() as cursor:
            rows = apsw_helpers.query(cursor, '''
                SELECT *
                FROM %s
                WHERE
                    %s IN %s
                    %s
                ORDER BY id ASC
            ''' % (self.table_name, api.shared.TaskState.PROJECTION, state_list, extra_predicate_sql),
                **query_params)

        return [ api.shared.task_load_row(row) for row in rows ] class Jobs(apsw_sql_utility.APSWSQLUtility):
    def __init__(self):
        super(Jobs, self).__init__(LoaderStorage())

        self._define_table(PRIMARY_TABLE)

    def save(self, job):
        assert isinstance(job, Job), 'job must be of type Job'
        with self.storage.transaction() as cursor:
            cursor.execute('''
                REPLACE INTO jobs (id, created, spec)
                VALUES (?, DATETIME(?, 'unixepoch'), ?)
            ''', (job.id, unix_timestamp(datetime.datetime.utcnow()), job.json_spec()))

    def delete(self, job):
        assert isinstance(job, Job), 'job must be of type Job'
        with self.storage.transaction() as cursor:
            cursor.execute('DELETE FROM jobs WHERE id = ?', (job.id,))

    def get(self, job_id):
        with self.storage.transaction() as cursor:
            job = apsw_helpers.get(
                cursor, 'SELECT id, spec FROM jobs WHERE id = ?', job_id)

        if job is not None:
            job['spec'] = json.loads(job.spec)
            return Job(job.spec, job.id)

    def all(self):
        with self.storage.cursor() as cursor:
            result = apsw_helpers.query(
                cursor, 'SELECT id, spec FROM jobs ORDER BY created ASC')

        return [Job(json.loads(job.spec), job.id) for job in result]

    def query_target(self, host, port, database, table):
        with self.storage.cursor() as cursor:
            result = apsw_helpers.query(cursor, 'SELECT id, spec FROM jobs')

        ret = []
        for job in result:
            spec = json.loads(job.spec)
            if spec['connection']['host'] != host:
                continue
            if spec['connection']['port'] != port:
                continue
            if spec['target']['database'] != database:
                continue
            if spec['target']['table'] != table:
                continue
            ret.append(Job(spec, job.id))
        return ret class WorkerException(Exception):
    def __init__(self, *args, **kwargs):
        super(Exception, self).__init__(*args, **kwargs)
        self.time = time.time() class ConnectionException(WorkerException):
    pass class RequeueTask(Exception):
    pass class Loader(Thread):
    def load(self, job, task, fifo, db_connection):
        self._job = job
        self._task = task
        self._fifo = fifo
        self._conn = db_connection
        load_data = LoadDataStmt(job, task.file_id, fifo.path)
        self._sql, self._params = load_data.build()
        self._error = None
        self._tb = None

        self._active_conn_id = None
        self._conn_lock = Lock()
        self._fifo.attach_reader(self.abort)

    @property
    def error(self):
        return self._error

    @property
    def traceback(self):
        return self._tb

    def abort(self):
        with self._conn_lock:
            if self._active_conn_id is not None:
                try:
                    with pool.get_connection(database='', **self._job.spec.connection) as conn:
                        db_utils.try_kill_connection(conn, self._active_conn_id)
                except pool.PoolConnectionException:
                    # If we couldn't connect, then its likely that we lost
                    # connection to the database and that the query is dead
                    # because of that anyways.
                    pass

                return True
            return False

    def run(self):
        self.logger = log.get_logger('loader')

        # Because self._conn was passed in from the worker thread, we need
        # to call the mysql_thread_init() C function to make sure that
        # everything is initialized properly.  However, _mysql doesn't expose
        # that function, so we call it implicitly by creating a MySQL
        # connection with a socket that's guaranteed to be invalid.
        try:
            _mysql.connect(unix_socket='.')
        except _mysql.MySQLError:
            pass

        try:
            self.logger.info('Starting loader')

            try:
                with self._conn_lock:
                    self._active_conn_id = self._conn.thread_id()
                    with self._task.protect():
                        self._task.data['conn_id'] = self._active_conn_id
                        self._task.save()
                row_count = self._conn.query(self._sql, *self._params)
            finally:
                with self._conn_lock:
                    self._active_conn_id = None

            with self._task.protect():
                self._task.data['row_count'] = row_count
                self._task.save()

        except connection_wrapper.ConnectionWrapperException as e:
            self.logger.error('LOAD DATA connection error: %s', str(e))
            self._set_error(ConnectionException(str(e)))

        except pool.MySQLError as e:
            errno, msg = e.args
            msg = "LOAD DATA error (%d): %s" % (errno, msg)
            self.logger.error(msg)
            self._set_error(WorkerException(msg))

        except Exception as e:
            self._set_error(e)

        except KeyboardInterrupt:
            self.logger.info('Received KeyboardInterrupt, exiting...')

        finally:
            self._fifo.detach_reader()
            self.logger.info('Finished LOAD_DATA')

    def _set_error(self, err):
        self._error = err
        self._tb = sys.exc_info()[2] class Downloader(threading.Thread):
    def __init__(self):
        super(Downloader, self).__init__()
        self.logger = log.get_logger('downloader')
        self._error = None
        self._tb = None
        self._should_exit = False

        self._last_size = -1
        self._last_download_time = 0

    def terminate(self):
        self._should_exit = True

    @property
    def error(self):
        return self._error

    @property
    def traceback(self):
        return self._tb

    def load(self, job, task, fifo):
        self.job = job
        self.task = task
        self.fifo = fifo
        self.key = None
        self.script_proc = None
        self.decompress_obj = None
        self.pycurl_callback_exception = None

        if task.data['scheme'] == 's3':
            self.is_anonymous = job.spec.source.aws_access_key is None or job.spec.source.aws_secret_key is None
            if self.is_anonymous:
                s3_conn = S3Connection(anon=True)
            else:
                s3_conn = S3Connection(job.spec.source.aws_access_key, job.spec.source.aws_secret_key)
            bucket = s3_conn.get_bucket(task.data['bucket'])

            try:
                self.key = bucket.get_key(task.data['key_name'])
            except S3ResponseError as e:
                raise WorkerException("Received %s %s accessing `%s`, aborting" % (e.status, e.reason, task.data['key_name']))
        elif task.data['scheme'] == 'hdfs':
            fname = task.data['key_name']
            client = PyWebHdfsClient(
                job.spec.source.hdfs_host,
                job.spec.source.webhdfs_port,
                user_name=job.spec.source.hdfs_user)
            try:
                filesize = client.get_file_dir_status(fname)['FileStatus']['length']
            except pywebhdfs.errors.FileNotFound:
                raise WorkerException("File '%s' does not exist on HDFS" % fname)
            self.key = AttrDict({'name': fname, 'size': filesize})
        elif task.data['scheme'] == 'file':
            globber = glob2.Globber()
            fname = globber._normalize_string(task.data['key_name'])

            if not os.path.exists(fname):
                raise WorkerException("File '%s' does not exist on this filesystem" % fname)
            elif not os.path.isfile(fname):
                raise WorkerException("File '%s' exists, but is not a file" % fname)

            self.key = AttrDict({'name': fname, 'size': os.path.getsize(fname)})
        else:
            raise WorkerException('Unsupported job with paths: %s' % [ str(p) for p in self.job.paths ])

        if self.key is None:
            raise WorkerException('Failed to find key associated with task ID %s' % task.task_id)

        self.metrics = DownloadMetrics(self.key.size)

    def run(self):
        try:
            try:
                # This is at the top so that any exceptions that occur will
                # emit a KILL QUERY due to fifo.open()

                # if we are piping through a script, the fifo should block
                # because the downloader is polling the script's stdin instead
                # of the fifo
                blocking = self.job.spec.options.script is not None
                with self.fifo.open(blocking=blocking) as target_file:
                    # allocate an URL for the target file
                    if self.task.data['scheme'] == 's3':
                        if self.is_anonymous:
                            key_url = 'http://%(bucket)s.s3.amazonaws.com/%(path)s' % {
                                'bucket': self.key.bucket.name,
                                'path': self.key.name.encode('utf-8')
                            }
                        else:
                            key_url = self.key.generate_url(expires_in=3600)
                    elif self.task.data['scheme'] == 'hdfs':
                        host = self.job.spec.source.hdfs_host
                        port = self.job.spec.source.webhdfs_port
                        hdfs_user = self.job.spec.source.hdfs_user
                        key_name = self.key.name
                        key_url = webhdfs.get_webhdfs_url(
                            host, port, hdfs_user, 'OPEN', key_name)
                    elif self.task.data['scheme'] == 'file':
                        key_url = 'file://%(path)s' % {'path': self.key.name}
                    else:
                        assert False, 'Unsupported job with paths: %s' % [ str(p) for p in self.job.paths ]

                    self._curl = curl = pycurl.Curl()
                    curl.setopt(pycurl.URL, key_url)
                    curl.setopt(pycurl.NOPROGRESS, 0)
                    curl.setopt(pycurl.PROGRESSFUNCTION, self._progress)
                    curl.setopt(pycurl.SSL_VERIFYPEER, 0)
                    curl.setopt(pycurl.SSL_VERIFYHOST, 0)
                    curl.setopt(pycurl.CONNECTTIMEOUT, 30)

                    if self.job.spec.options.script is not None:
                        self.script_proc = subprocess.Popen(
                            ["/bin/bash", "-c", self.job.spec.options.script],
                            stdout=target_file.fileno(),
                            stdin=subprocess.PIPE)

                        # check that script hasn't errored before downloading
                        # NOTE: we wait here so that we can check if a script exits prematurely
                        # if this is the case, we fail the job without requeueing
                        time.sleep(1)
                        if self.script_proc.poll() is not None:
                            self.logger.error('Script `%s` exited prematurely with return code %d' % (self.job.spec.options.script, self.script_proc.returncode))
                            raise WorkerException('Script `%s` exited prematurely with return code %d' % (self.job.spec.options.script, self.script_proc.returncode))

                        # If we're piping data into a script and this file is
                        # a gzipped file, we'll decompress the data ourselves
                        # before piping it into the script.
                        if self.task.data['key_name'].endswith('.gz'):
                            # Set the window bits during decompression to
                            # zlib.MAX_WBITS | 32 tells the zlib library to
                            # automatically detect gzip headers.
                            self.decompress_obj = zlib.decompressobj(zlib.MAX_WBITS | 32)

                        curl.setopt(pycurl.WRITEFUNCTION, self._write_to_fifo(self.script_proc.stdin))
                    else:
                        curl.setopt(pycurl.WRITEFUNCTION, self._write_to_fifo(target_file))

                    if self.task.data['scheme'] == 'hdfs':
                        curl.setopt(pycurl.FOLLOWLOCATION, True)

                    self.logger.info('Starting download')
                    with self.task.protect():
                        self.task.start_step('download')

                    try:
                        curl.perform()
                        status_code = curl.getinfo(pycurl.HTTP_CODE)
                        # HTTP client errors will cause task failure (no retry)
                        if status_code >= 400 and status_code < 500:
                            raise WorkerException('HTTP status code %s for file %s' % (status_code, self.key.name))
                        # HTTP server errors will cause task retry
                        elif status_code >= 500:
                            self.logger.warn('Received HTTP status code %s for file %s, requeueing' % (status_code, self.key.name))
                            raise RequeueTask()

                        # If we're piping data through a script, catch timeouts and return codes
                        if self.script_proc is not None:
                            self.script_proc.stdin.close()
                            for i in range(SCRIPT_EXIT_TIMEOUT):
                                if self.script_proc.poll() is not None:
                                    break

                                time.sleep(1)
                            else:
                                self.logger.error('Script `%s` failed to exit...killing' % self.job.spec.options.script)
                                self.script_proc.kill()
                                raise WorkerException('Script `%s` failed to exit after %d seconds' % (self.job.spec.options.script, SCRIPT_EXIT_TIMEOUT))

                            if self.script_proc.returncode != 0:
                                self.logger.error('Script `%s` exited with return code %d' % (self.job.spec.options.script, self.script_proc.returncode))
                                raise WorkerException('Script `%s` exited with return code %d' % (self.job.spec.options.script, self.script_proc.returncode))
                    finally:
                        with self.task.protect():
                            self.task.stop_step('download')

                            if self.script_proc is not None and self.script_proc.returncode is None:
                                try:
                                    self.script_proc.kill()
                                except OSError as e:
                                    self.logger.warn("Failed to kill script `%s`: %s" % (self.job.spec.options.script, str(e)))
            except pycurl.error as e:
                errno = e.args[0]
                if errno in (pycurl.E_WRITE_ERROR, pycurl.E_ABORTED_BY_CALLBACK):
                    if self.pycurl_callback_exception is not None:
                        raise self.pycurl_callback_exception
                    elif self._should_exit:
                        self.logger.warn('Download failed...requeueing')
                        # Caught by the outer `except Exception as e`
                        raise RequeueTask()

                # Caught by the outer `except pycurl.error as e`
                raise
        except pycurl.error as e:
            errno = e.args[0]
            self._set_error(ConnectionException('libcurl error #%d. Lookup error here: http://curl.haxx.se/libcurl/c/libcurl-errors.html' % errno))
        except IOError as e:
            # This is raised sometimes instead of a pycurl error
            self._set_error(ConnectionException('IOError: %s (%d)' % (e.args[1], e.args[0])))
        except Exception as e:
            self._set_error(e)
        except KeyboardInterrupt:
            pass
        finally:
            self.logger.info('Finished downloading')

    def _set_error(self, err):
        self._error = err
        self._tb = sys.exc_info()[2]
        self.logger.debug("Downloader failed: %s." % (err), exc_info=True)

    def _progress(self, dltotal, dlnow, ultotal, ulnow):
        self.metrics.accumulate_bytes(dlnow)

        if self._should_exit or time.time() > self.metrics.last_change + DOWNLOAD_TIMEOUT:
            return 1

    def _write_to_fifo(self, target_file):
        def _write_to_fifo_helper(data):
            to_write = data

            try:
                if self.decompress_obj is not None:
                    to_write = self.decompress_obj.decompress(to_write)
                while len(to_write) > 0:
                    # First step is to wait until we can write to the FIFO.
                    #
                    # Wait for half of the download timeout for the FIFO to become open
                    # for writing.  While we're doing this, ping the download metrics
                    # so that the worker doesn't assume this download has hung.
                    is_writable = False
                    while not is_writable:
                        self.metrics.ping()
                        timeout = DOWNLOAD_TIMEOUT / 2
                        _, writable_objects, _ = select.select(
                            [ ], [ target_file ], [ ], timeout)
                        is_writable = bool(writable_objects)

                    # Then, we write as much as we can within this opportunity to write
                    written_bytes = os.write(target_file.fileno(), to_write)
                    assert written_bytes >= 0, "Expect os.write() to return non-negative numbers"
                    to_write = to_write[written_bytes:]
            except zlib.error as e:
                self.terminate()
                # pycurl will just raise pycurl.error if this function
                # raises an exception, so we also need to set the exception
                # on the Downloader object so that we can check it and
                # re-raise it above.
                self.pycurl_callback_exception = WorkerException(
                    'Could not decompress data: %s' % str(e))
                raise self.pycurl_callback_exception
            except OSError as e:
                if self.script_proc is not None and self.script_proc.poll() is not None:
                    self.terminate()
                    self.pycurl_callback_exception = WorkerException(
                        'Script `%s` exited during download with return code %d' %
                        (self.job.spec.options.script, self.script_proc.returncode))
                    raise self.pycurl_callback_exception
                else:
                    raise

        return _write_to_fifo_helper def validate_file_id_column(conn, database, table, col_name):
def validate_database_table(conn, database, table):
def try_kill_query(conn, query_id):
def try_kill_connection(conn, conn_id): def setup(log_path=None, stdout_enabled=True):
def update_verbosity(debug=False, extra_verbose=False):
    def __call__(self, parser, namespace, value, option_string=None):
    def __call__(self, parser, namespace, value, option_string=None):
    def __call__(self, parser, namespace, value, option_string=None):
def configure(parser):
def get_logger(name, level=None):
    def format(self, record):
    def __init__(self, name, level=logging.DEBUG):
class _SetDebug(argparse.Action):
class _SetStdout(argparse.Action):
class _SetLogPath(argparse.Action):
class Formatter(logging.Formatter):
class Logger(logging.Logger): class FIFO(object):
    def __init__(self, gzip=False):
        self._closed = False
        self._reader_abort_fn = None
        self._tempdir = tempfile.mkdtemp(dir='/tmp')
        os.chmod(self._tempdir, S_IRWXU | S_IRWXG | S_IRWXO)

        self.path = os.path.join(self._tempdir, 'fifo')
        if gzip:
            self.path += '.gz'

        try:
            os.mkfifo(self.path)
            os.chmod(self.path, S_IRWXU | S_IRWXG | S_IRWXO)
        except OSError:
            os.rmdir(self._tempdir)
            raise

    def __del__(self):
        self.cleanup()

    @contextmanager
    def open(self, blocking=False):
        """ This returns a file descriptor, not a file object. The
        file descriptor must be written to with os.write()

        :param blocking: If True, the FIFO will block on write
        """

        assert not self._closed, 'FIFO has already been cleaned up'
        # Will hang until a reader attaches to the other end of the pipe,
        # which is what we want since we don't want to start writing to
        # the pipe until a reader attaches in case we finish writing
        # and close the file before the reader attaches, which would
        # drop all the data.
        with open(self.path, 'wb') as fifofile:
            try:
                fcntl.fcntl(fifofile.fileno(), fcntl.F_SETFL, 0 if blocking else os.O_NONBLOCK)
                yield fifofile
            except:
                self.abort_reader()
                raise

    def attach_reader(self, abort_function):
        assert self._reader_abort_fn is None, 'A reader is already attached to this FIFO'
        self._reader_abort_fn = abort_function

    def detach_reader(self):
        # Opening the pipe for writing above will hang until a reader attaches,
        # so we make sure that on detaching the reader, we always attempt to
        # read from the pipe to unblock it.
        try:
            os.close(os.open(self.path, os.O_RDONLY | os.O_NONBLOCK))
        except OSError:
            pass
        self._reader_abort_fn = None

    def abort_reader(self):
        if self._reader_abort_fn is not None:
            self._reader_abort_fn()
            self._reader_abort_fn = None

    def cleanup(self):
        if self._closed:
            return

        self._closed = True
        os.unlink(self.path)
        os.rmdir(self._tempdir) class APSWSQLStepQueueException(Exception):
    pass class TaskDoesNotExist(APSWSQLStepQueueException):
    pass
import multiprocessing
import threading
import uuid
import random
import time
import os
import signal
import traceback
from memsql_loader.api import shared
from memsql_loader.db import connection_wrapper, pool
from memsql_loader.loader_db.tasks import Tasks
from memsql_loader.loader_db.jobs import Jobs
from memsql_loader.execution.errors import WorkerException, ConnectionException, RequeueTask
from memsql_loader.execution.loader import Loader
from memsql_loader.execution.downloader import Downloader
from memsql_loader.util import db_utils, log
from memsql_loader.util.fifo import FIFO
from memsql_loader.util.apsw_sql_step_queue.errors import APSWSQLStepQueueException, TaskDoesNotExist



HUNG_DOWNLOADER_TIMEOUT = 3600

class ExitingException(Exception):
    pass

class Worker(multiprocessing.Process):
    def __init__(self, worker_sleep, parent_pid, worker_lock):
        self.worker_id = uuid.uuid1().hex[:8]
        self.worker_sleep = worker_sleep
        self.worker_lock = worker_lock
        self.worker_working = multiprocessing.Value('i', 1)
        self.parent_pid = parent_pid
        self._exit_evt = multiprocessing.Event()
        self.logger = log.get_logger('worker[%s]' % self.worker_id)
        super(Worker, self).__init__(name=('worker-%s' % self.worker_id))

    def kill_query_if_exists(self, conn_args, conn_id):
        with pool.get_connection(database='information_schema', **conn_args) as conn:
            id_row = conn.query("SELECT id FROM processlist WHERE info LIKE '%%LOAD DATA%%' AND id=%s", conn_id)
            if len(id_row) > 0:
                # Since this is a LOAD DATA LOCAL query, we need to kill the
                # connection, not the query, since LOAD DATA LOCAL queries
                # don't end until the file is fully read, even if they're
                # killed.
                db_utils.try_kill_connection(conn, conn_id)

    def kill_delete_query_if_exists(self, conn_args, conn_id):
        with pool.get_connection(database='information_schema', **conn_args) as conn:
            id_row = conn.query("SELECT id FROM processlist WHERE info LIKE '%%DELETE%%' AND id=%s", conn_id)
            if len(id_row) > 0:
                db_utils.try_kill_query(conn, conn_id)

    def signal_exit(self):
        self._exit_evt.set()

    def is_working(self):
        return self.worker_working.value == 1

    def run(self):
        self.jobs = Jobs()
        self.tasks = Tasks()
        task = None

        ignore = lambda *args, **kwargs: None
        signal.signal(signal.SIGINT, ignore)
        signal.signal(signal.SIGQUIT, ignore)

        try:
            while not self.exiting():
                time.sleep(random.random() * 0.5)
                task = self.tasks.start()

                if task is None:
                    self.worker_working.value = 0
                else:
                    self.worker_working.value = 1

                    job_id = task.job_id
                    job = self.jobs.get(job_id)

                    old_conn_id = task.data.get('conn_id', None)
                    if old_conn_id is not None:
                        self.kill_query_if_exists(job.spec.connection, old_conn_id)

                    self.logger.info('Task %d: starting' % task.task_id)

                    try:
                        # can't use a pooled connection due to transactions staying open in the
                        # pool on failure
                        with pool.get_connection(database=job.spec.target.database, pooled=False, **job.spec.connection) as db_connection:
                            db_connection.execute("BEGIN")
                            self._process_task(task, db_connection)
                        self.logger.info('Task %d: finished with success', task.task_id)

                    except (RequeueTask, ConnectionException):