docs: 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