'Determines the overall state of the DagRun based on the state
of its TaskInstances.
:return: State'

@provide_session
def update_state(self, session=None):

    dag = self.get_dag()
    tis = self.get_task_instances(session=session)
    logging.info(u'Updating    state    for    {}    considering    {}    task(s)'.format(self, len(tis)))
    for ti in list(tis):
        if (ti.state == State.REMOVED):
            tis.remove(ti)
        else:
            ti.task = dag.get_task(ti.task_id)
    start_dttm = datetime.now()
    unfinished_tasks = self.get_task_instances(state=State.unfinished(), session=session)
    none_depends_on_past = all(((not t.task.depends_on_past) for t in unfinished_tasks))
    if (unfinished_tasks and none_depends_on_past):
        no_dependencies_met = all(((not t.are_dependencies_met(dep_context=self.DEADLOCK_CHECK_DEP_CONTEXT, session=session)) for t in unfinished_tasks))
    duration = ((datetime.now() - start_dttm).total_seconds() * 1000)
    Stats.timing(u'dagrun.dependency-check.{}.{}'.format(self.dag_id, self.execution_date), duration)
    if (len(tis) == len(dag.active_tasks)):
        root_ids = [t.task_id for t in dag.roots]
        roots = [t for t in tis if (t.task_id in root_ids)]
        if ((not unfinished_tasks) and any(((r.state in (State.FAILED, State.UPSTREAM_FAILED)) for r in roots))):
            logging.info(u'Marking    run    {}    failed'.format(self))
            self.state = State.FAILED
        elif ((not unfinished_tasks) and all(((r.state in (State.SUCCESS, State.SKIPPED)) for r in roots))):
            logging.info(u'Marking    run    {}    successful'.format(self))
            self.state = State.SUCCESS
        elif (unfinished_tasks and none_depends_on_past and no_dependencies_met):
            logging.info(u'Deadlock;    marking    run    {}    failed'.format(self))
            self.state = State.FAILED
        else:
            self.state = State.RUNNING
    session.merge(self)
    session.commit()
    return self.state
