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')
dataset/ETHPy150Open memsql/memsql-loader/memsql_loader/execution/loader.py/Loader.run/original