'Runs the task instance.
:param verbose: whether to turn on more verbose logging
:type verbose: boolean
:param ignore_all_deps: Ignore all of the non-critical dependencies, just runs
:type ignore_all_deps: boolean
:param ignore_depends_on_past: Ignore depends_on_past DAG attribute
:type ignore_depends_on_past: boolean
:param ignore_task_deps: Don\'t check the dependencies of this TI\'s task
:type ignore_task_deps: boolean
:param ignore_ti_state: Disregards previous task instance state
:type ignore_ti_state: boolean
:param mark_success: Don\'t run the task, mark its state as success
:type mark_success: boolean
:param test_mode: Doesn\'t record success or failure in the DB
:type test_mode: boolean
:param pool: specifies the pool to use to run the task instance
:type pool: str'

@provide_session
def run(self, verbose=True, ignore_all_deps=False, ignore_depends_on_past=False, ignore_task_deps=False, ignore_ti_state=False, mark_success=False, test_mode=False, job_id=None, pool=None, session=None):

    task = self.task
    self.pool = (pool or task.pool)
    self.test_mode = test_mode
    self.refresh_from_db(session=session, lock_for_update=True)
    self.job_id = job_id
    self.hostname = socket.getfqdn()
    self.operator = task.__class__.__name__
    if ((not ignore_all_deps) and (not ignore_ti_state) and (self.state == State.SUCCESS)):
        Stats.incr(u'previously_succeeded', 1, 1)
    queue_dep_context = DepContext(deps=QUEUE_DEPS, ignore_all_deps=ignore_all_deps, ignore_ti_state=ignore_ti_state, ignore_depends_on_past=ignore_depends_on_past, ignore_task_deps=ignore_task_deps)
    if (not self.are_dependencies_met(dep_context=queue_dep_context, session=session, verbose=True)):
        session.commit()
        return
    hr = ((u'\n' + (u'-' * 80)) + u'\n')
    msg = u'Starting    attempt    {attempt}    of    {total}'.format(attempt=(self.try_number + 1), total=(self.max_tries + 1))
    self.start_date = datetime.now()
    dep_context = DepContext(deps=(RUN_DEPS - QUEUE_DEPS), ignore_all_deps=ignore_all_deps, ignore_depends_on_past=ignore_depends_on_past, ignore_task_deps=ignore_task_deps, ignore_ti_state=ignore_ti_state)
    runnable = self.are_dependencies_met(dep_context=dep_context, session=session, verbose=True)
    if ((not runnable) and (not mark_success)):
        self.state = State.NONE
        msg = u'FIXME:    Rescheduling    due    to    concurrency    limits    reached    at    task    runtime.    Attempt    {attempt}    of    {total}.    State    set    to    NONE.'.format(attempt=(self.try_number + 1), total=(self.max_tries + 1))
        logging.warning(((hr + msg) + hr))
        self.queued_dttm = datetime.now()
        msg = u'Queuing    into    pool    {}'.format(self.pool)
        logging.info(msg)
        session.merge(self)
        session.commit()
        return
    if (self.state == State.RUNNING):
        msg = u'Task    Instance    already    running    {}'.format(self)
        logging.warning(msg)
        session.commit()
        return
    logging.info(((hr + msg) + hr))
    self.try_number += 1
    if (not test_mode):
        session.add(Log(State.RUNNING, self))
    self.state = State.RUNNING
    self.pid = os.getpid()
    self.end_date = None
    if (not test_mode):
        session.merge(self)
    session.commit()
    settings.engine.dispose()
    if verbose:
        if mark_success:
            msg = u'Marking    success    for    '
        else:
            msg = u'Executing    '
        msg += u'{self.task}    on    {self.execution_date}'
    context = {}
    try:
        logging.info(msg.format(self=self))
        if (not mark_success):
            context = self.get_template_context()
            task_copy = copy.copy(task)
            self.task = task_copy
            def signal_handler(signum, frame):
                u'Setting    kill    signal    handler'
                logging.error(u'Killing    subprocess')
                task_copy.on_kill()
                raise AirflowException(u'Task    received    SIGTERM    signal')
            signal.signal(signal.SIGTERM, signal_handler)
            self.clear_xcom_data()
            self.render_templates()
            task_copy.pre_execute(context=context)
            result = None
            if task_copy.execution_timeout:
                try:
                    with timeout(int(task_copy.execution_timeout.total_seconds())):
                        result = task_copy.execute(context=context)
                except AirflowTaskTimeout:
                    task_copy.on_kill()
                    raise
            else:
                result = task_copy.execute(context=context)
            if (result is not None):
                self.xcom_push(key=XCOM_RETURN_KEY, value=result)
            try:
                task_copy.post_execute(context=context, result=result)
            except TypeError as e:
                if (u'unexpected    keyword    argument' in str(e)):
                    warnings.warn(u'BaseOperator.post_execute()    now    takes    two    arguments,    `context`    and    `result`,    but    "{}"    only    expected    one.    This    behavior    is    deprecated    and    will    be    removed    in    a    future    version    of    Airflow.'.format(self.task_id), category=DeprecationWarning)
                    task_copy.post_execute(context=context)
                else:
                    raise
            Stats.incr(u'operator_successes_{}'.format(self.task.__class__.__name__), 1, 1)
            Stats.incr(u'ti_successes')
        self.state = State.SUCCESS
    except AirflowSkipException:
        self.state = State.SKIPPED
    except (Exception, KeyboardInterrupt) as e:
        self.handle_failure(e, test_mode, context)
        raise
    self.end_date = datetime.now()
    self.set_duration()
    if (not test_mode):
        session.add(Log(self.state, self))
        session.merge(self)
    session.commit()
    try:
        if task.on_success_callback:
            task.on_success_callback(context)
    except Exception as e3:
        logging.error(u'Failed    when    executing    success    callback')
        logging.exception(e3)
    session.commit()
