(app_name, job_id)
return parents_completed, consume_queue, parent_lock
Assume that given job_id is pulled from the app_name's queue.

    Check that the parent tasks for this (app_name, job_id) pair have completed
    If they haven't completed and aren't pending, maybe create the
    parent task in its appropriate queue.  Also decide whether the calling
    process should requeue given job_id or remove itself from queue.
    Because this needs to happen as one transaction, also return a list of
    execute locks that the calling code must release after it decides how to
    handle the current job_id.

    Returns a tuple:
        (are_parents_completed, should_job_id_be_consumed_from_queue,
         parent_execute_locks_to_release)
Assume that given job_id is pulled from the app_name's queue.
Assume that given job_id is pulled from the app_name s queue .
def ensure_parents_completed(app_name, job_id):
    """
    Assume that given job_id is pulled from the app_name's queue.

    Check that the parent tasks for this (app_name, job_id) pair have completed
    If they haven't completed and aren't pending, maybe create the
    parent task in its appropriate queue.  Also decide whether the calling
    process should requeue given job_id or remove itself from queue.
    Because this needs to happen as one transaction, also return a list of
    execute locks that the calling code must release after it decides how to
    handle the current job_id.

    Returns a tuple:
        (are_parents_completed, should_job_id_be_consumed_from_queue,
         parent_execute_locks_to_release)
    """
    parents_completed = True
    consume_queue = False
    parent_lock = None
    for parent, pjob_id, dep_grp in dt.get_parents(app_name, job_id, True):
        if check_state(app_name=parent, job_id=pjob_id, completed=True):
            continue
        parents_completed = False
        log.info(
            'My parent has not completed yet.',
            extra=dict(
                parent_app_name=parent, parent_job_id=pjob_id,
                app_name=app_name, job_id=job_id))

        # At this point, I need to be re-run
        # The question at this point is whether to requeue myself or assume the
        # parent will.

        # Assume the default is I requeue myself.  Sometimes, this might result
        # in me cycling through the queue a couple times until parent finishes.

        # If parent is running, it will be able to requeue me if I exit in
        # time.  If it doesn't, either I'll requeue myself by default or
        # another parent will.  So, do nothing in this case.

        # if parent is not running, I should try to maybe_add_subtask it.
        # - if can't add parent, then possibly something else is adding it, or
        # it ran once and is waiting on one of my grandparents.
        # - if I can maybe_add_subtask my parent, then it definitely wasn't
        # running before.

        # In both cases,
        # I should try to unqueue myself if I can guarantee that the parent
        # won't run by the time I unqueue myself.  Otherwise, I should just
        # default to assuming parent is running and requeue myself by default.

        added = maybe_add_subtask(parent, pjob_id)

        # if parent marked 'skipped' and then someone calls a maybe_add_subtask
        # on the child, child could requeue itself indefinitely.  to prevent,
        # child should unqueue itself and raise error complaint that for some
        # insane reason it's running but it's parent is "skipped"
        if not added and check_state(parent, pjob_id, skipped=True):
            consume_queue = True
            #  raise some sort of error
            log.warn(
                "My parent_job_id is marked as 'skipped',"
                " so should be impossible for me, the child, to exist!"
                " Requesting to unqueue myself.  This is odd.", extra=dict(
                    parent_app_name=parent, parent_job_id=pjob_id,
                    app_name=app_name, job_id=job_id))
            break

        if parent_lock is not None:
            continue  # we already found a parent that promises to requeue me

        elock = obtain_execute_lock(
            parent, pjob_id, raise_on_error=False, blocking=False)
        if elock:
            if not check_state(parent, pjob_id, pending=True):
                elock.release()  # race condition: parent just did something!
            else:
                consume_queue = True
                parent_lock = elock
                log.info(
                    "I will unqueue myself with the expectation that"
                    " my parent will requeue me", extra=dict(
                        app_name=app_name, job_id=job_id))
    return parents_completed, consume_queue, parent_lock
def ensure_parents_completed ( app_name , job_id ) : parents_completed = True consume_queue = False parent_lock = None for parent , pjob_id , dep_grp in dt . get_parents ( app_name , job_id , True ) : if check_state ( app_name = parent , job_id = pjob_id , completed = True ) : continue parents_completed = False log . info ( 'My parent has not completed yet.' , extra = dict ( parent_app_name = parent , parent_job_id = pjob_id , app_name = app_name , job_id = job_id ) ) # At this point, I need to be re-run # The question at this point is whether to requeue myself or assume the # parent will. # Assume the default is I requeue myself.  Sometimes, this might result # in me cycling through the queue a couple times until parent finishes. # If parent is running, it will be able to requeue me if I exit in # time.  If it doesn't, either I'll requeue myself by default or # another parent will.  So, do nothing in this case. # if parent is not running, I should try to maybe_add_subtask it. # - if can't add parent, then possibly something else is adding it, or # it ran once and is waiting on one of my grandparents. # - if I can maybe_add_subtask my parent, then it definitely wasn't # running before. # In both cases, # I should try to unqueue myself if I can guarantee that the parent # won't run by the time I unqueue myself.  Otherwise, I should just # default to assuming parent is running and requeue myself by default. added = maybe_add_subtask ( parent , pjob_id ) # if parent marked 'skipped' and then someone calls a maybe_add_subtask # on the child, child could requeue itself indefinitely.  to prevent, # child should unqueue itself and raise error complaint that for some # insane reason it's running but it's parent is "skipped" if not added and check_state ( parent , pjob_id , skipped = True ) : consume_queue = True #  raise some sort of error log . warn ( "My parent_job_id is marked as 'skipped'," " so should be impossible for me, the child, to exist!" " Requesting to unqueue myself.  This is odd." , extra = dict ( parent_app_name = parent , parent_job_id = pjob_id , app_name = app_name , job_id = job_id ) ) break if parent_lock is not None : continue # we already found a parent that promises to requeue me elock = obtain_execute_lock ( parent , pjob_id , raise_on_error = False , blocking = False ) if elock : if not check_state ( parent , pjob_id , pending = True ) : elock . release ( ) # race condition: parent just did something! else : consume_queue = True parent_lock = elock log . info ( "I will unqueue myself with the expectation that" " my parent will requeue me" , extra = dict ( app_name = app_name , job_id = job_id ) ) return parents_completed , consume_queue , parent_lock