Core recursion function for :func:`check_completion`, see there for more documentation

    Args:
        task (luigi.Task): task instance

        mark_incomplete (bool): see :func:`check_completion`

        clear (bool): see :func:`check_completion`

        stats (dict): task checking statistics (counts of incomplete, to be cleared, ...)

        visited (dict): cache for visited tasks: key = task name + parameter string, value = completion status

        to_clear (dict): dict of dicts of tasks to be cleared, key = task id,
            value =  {task: task instance object, required_by: set of task IDs that this task is required by}

def _check_completion(task, mark_incomplete, clear, stats, visited, to_clear):
    """Core recursion function for :func:`check_completion`, see there for more documentation

    Args:
        task (luigi.Task): task instance

        mark_incomplete (bool): see :func:`check_completion`

        clear (bool): see :func:`check_completion`

        stats (dict): task checking statistics (counts of incomplete, to be cleared, ...)

        visited (dict): cache for visited tasks: key = task name + parameter string, value = completion status

        to_clear (dict): dict of dicts of tasks to be cleared, key = task id,
            value =  {task: task instance object, required_by: set of task IDs that this task is required by}

    """
    # is this task (recursively) complete?
    task_complete = task.complete()
    is_complete = task_complete

    # task identification: task name plus parameters
    task_id = get_task_name(task) + ' ' + get_task_param_string(task)

    # check any requirements
    for req in task.requires():

        # task/parameter ID string to identify this task instance
        req_id = get_task_name(req) + ' ' + get_task_param_string(req)

        # skip recursion on already visited tasks, get completion status from cache
        if req_id in visited:
            req_complete = visited[req_id]
        else:
            req_complete, _ = _check_completion(task=req,
                                                mark_incomplete=mark_incomplete,
                                                clear=clear,
                                                stats=stats,
                                                visited=visited,
                                                to_clear=to_clear)
            visited[req_id] = req_complete

        # add any incomplete requirements to the list of tasks to clear, noting the current task as parent (required by)
        if clear and not req_complete:
            clear_entry = to_clear.setdefault(req_id, dict(task=req, required_by=set()))
            clear_entry['required_by'].add(task_id)

        is_complete &= req_complete

    if not is_complete:
        if task_complete:
            config.logger.info("Task complete but requirements incomplete: " + task_id)
        else:
            config.logger.info("Task incomplete: " + task_id)
        _increment_stats(stats, 'Incomplete tasks')

        if mark_incomplete:
            if isinstance(task, ORMTask):
                task.mark_incomplete()
                _increment_stats(stats, 'Marked incomplete')
                config.logger.info("Marked task incomplete: " + task_id)
            else:
                config.logger.info('Cannot mark task incomplete, not an ORMTask: ' + task_id)

    else:
        _increment_stats(stats, 'Complete tasks')
        config.logger.debug("Task complete: " + task_id)

    # if we want to clear and the current task is not in the dict of tasks to clear,
    # it is the root task, add it with no parent (required by) tasks
    if clear and not is_complete and task_id not in to_clear:
        to_clear[task_id] = dict(task=task, required_by=set())

    return is_complete, stats