@provide_session
    def run(
            self,
            verbose=True,
            ignore_dependencies=False,  # Doesn't check for deps, just runs
            ignore_depends_on_past=False,   # Ignore depends_on_past but respect
                                            # other deps
            force=False,  # Disregards previous successes
            mark_success=False,  # Don't run the task, act as if it succeeded
            test_mode=False,  # Doesn't record success or failure in the DB
            job_id=None,
            pool=None,
            session=None):
        """
        Runs the task instance.
        """
        task = self.task
        self.pool = pool or task.pool
        self.test_mode = test_mode
        self.force = force
        self.refresh_from_db()
        self.clear_xcom_data()
        self.job_id = job_id
        iso = datetime.now().isoformat()
        self.hostname = socket.gethostname()
        self.operator = task.__class__.__name__

        if self.state == State.RUNNING:
            logging.warning("Another instance is running, skipping.")
        elif not force and self.state == State.SUCCESS:
            logging.info(
                "Task {self} previously succeeded"
                " on {self.end_date}".format(**locals())
            )
            Stats.incr('previously_succeeded', 1, 1)
        elif (
                not ignore_dependencies and
                not self.are_dependencies_met(
                    session=session,
                    ignore_depends_on_past=ignore_depends_on_past,
                    verbose=True)):
            logging.warning("Dependencies not met yet")
        elif (
                self.state == State.UP_FOR_RETRY and
                not self.ready_for_retry()):
            next_run = (self.end_date + task.retry_delay).isoformat()
            logging.info(
                "Not ready for retry yet. " +
                "Next run after {0}".format(next_run)
            )
        elif force or self.state in State.runnable():
            HR = "\n" + ("-" * 80) + "\n"  # Line break

            # For reporting purposes, we report based on 1-indexed,
            # not 0-indexed lists (i.e. Attempt 1 instead of
            # Attempt 0 for the first attempt).
            msg = "Starting attempt {attempt} of {total}".format(
                attempt=self.try_number % (task.retries + 1) + 1,
                total=task.retries + 1)
            self.start_date = datetime.now()

            if not mark_success and self.state != State.QUEUED and (
                    self.pool or self.task.dag.concurrency_reached):
                # If a pool is set for this task, marking the task instance
                # as QUEUED
                self.state = State.QUEUED
                msg = "Queuing attempt {attempt} of {total}".format(
                    attempt=self.try_number % (task.retries + 1) + 1,
                    total=task.retries + 1)
                logging.info(HR + msg + HR)

                self.queued_dttm = datetime.now()
                session.merge(self)
                session.commit()
                logging.info("Queuing into pool {}".format(self.pool))
                return

            # print status message
            logging.info(HR + msg + HR)
            self.try_number += 1

            if not test_mode:
                session.add(Log(State.RUNNING, self))
            self.state = State.RUNNING
            self.end_date = None
            if not test_mode:
                session.merge(self)
            session.commit()

            # Closing all pooled connections to prevent
            # "max number of connections reached"
            settings.engine.dispose()
            if verbose:
                if mark_success:
                    msg = "Marking success for "
                else:
                    msg = "Executing "
                msg += "{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):
                        '''Setting kill signal handler'''
                        logging.error("Killing subprocess")
                        task_copy.on_kill()
                        raise AirflowException("Task received SIGTERM signal")
                    signal.signal(signal.SIGTERM, signal_handler)

                    self.render_templates()
                    task_copy.pre_execute(context=context)

                    # If a timout is specified for the task, make it fail
                    # if it goes beyond
                    result = None
                    if task_copy.execution_timeout:
                        with timeout(int(
                                task_copy.execution_timeout.total_seconds())):
                            result = task_copy.execute(context=context)

                    else:
                        result = task_copy.execute(context=context)

                    # If the task returns a result, push an XCom containing it
                    if result is not None:
                        self.xcom_push(key=XCOM_RETURN_KEY, value=result)

                    task_copy.post_execute(context=context)
                self.state = State.SUCCESS
            except AirflowSkipException:
                self.state = State.SKIPPED
            except (Exception, __HOLE__) as e:
                self.handle_failure(e, test_mode, context)
                raise

            # Recording SUCCESS
            self.end_date = datetime.now()
            self.set_duration()
            if not test_mode:
                session.add(Log(self.state, self))
                session.merge(self)
            session.commit()

            # Success callback
            try:
                if task.on_success_callback:
                    task.on_success_callback(context)
            except Exception as e3:
                logging.error("Failed when executing success callback")
                logging.exception(e3)

        session.commit()
dataset/ETHPy150Open airbnb/airflow/airflow/models.py/TaskInstance.run