docs: def _thread_return(cls, minion_instance, opts, data):
          '''
          This method should be used as a threading target, start the actual
          minion side execution.
          '''
          fn_ = os.path.join(minion_instance.proc_dir, data['jid'])
  
          if opts['multiprocessing'] and not salt.utils.platform.is_windows():
              # Shutdown the multiprocessing before daemonizing
              salt.log.setup.shutdown_multiprocessing_logging()
  
              salt.utils.process.daemonize_if(opts)
  
              # Reconfigure multiprocessing logging after daemonizing
              salt.log.setup.setup_multiprocessing_logging()
  
          salt.utils.process.appendproctitle('{0}._thread_return {1}'.format(cls.__name__, data['jid']))
  
          sdata = {'pid': os.getpid()}
          sdata.update(data)
          log.info('Starting a new job %s with PID %s', data['jid'], sdata['pid'])
          with salt.utils.files.fopen(fn_, 'w+b') as fp_:
              fp_.write(minion_instance.serial.dumps(sdata))
          ret = {'success': False}
          function_name = data['fun']
          executors = data.get('module_executors') or \
                      getattr(minion_instance, 'module_executors', []) or \
                      opts.get('module_executors', ['direct_call'])
          allow_missing_funcs = any([
              minion_instance.executors['{0}.allow_missing_func'.format(executor)](function_name)
              for executor in executors
              if '{0}.allow_missing_func' in minion_instance.executors
          ])
          if function_name in minion_instance.functions or allow_missing_funcs is True:
              try:
                  minion_blackout_violation = False
                  if minion_instance.connected and minion_instance.opts['pillar'].get('minion_blackout', False):
                      whitelist = minion_instance.opts['pillar'].get('minion_blackout_whitelist', [])
                      # this minion is blacked out. Only allow saltutil.refresh_pillar and the whitelist
                      if function_name != 'saltutil.refresh_pillar' and function_name not in whitelist:
                          minion_blackout_violation = True
                  # use minion_blackout_whitelist from grains if it exists
                  if minion_instance.opts['grains'].get('minion_blackout', False):
                      whitelist = minion_instance.opts['grains'].get('minion_blackout_whitelist', [])
                      if function_name != 'saltutil.refresh_pillar' and function_name not in whitelist:
                          minion_blackout_violation = True
                  if minion_blackout_violation:
                      raise SaltInvocationError('Minion in blackout mode. Set \'minion_blackout\' '
                                               'to False in pillar or grains to resume operations. Only '
                                               'saltutil.refresh_pillar allowed in blackout mode.')
  
                  if function_name in minion_instance.functions:
                      func = minion_instance.functions[function_name]
                      args, kwargs = load_args_and_kwargs(
                          func,
                          data['arg'],
                          data)
                  else:
                      # only run if function_name is not in minion_instance.functions and allow_missing_funcs is True
                      func = function_name
                      args, kwargs = data['arg'], data
                  minion_instance.functions.pack['__context__']['retcode'] = 0
                  if isinstance(executors, six.string_types):
                      executors = [executors]
                  elif not isinstance(executors, list) or not executors:
                      raise SaltInvocationError("Wrong executors specification: {0}. String or non-empty list expected".
                          format(executors))
                  if opts.get('sudo_user', '') and executors[-1] != 'sudo':
                      executors[-1] = 'sudo'  # replace the last one with sudo
                  log.trace('Executors list %s', executors)  # pylint: disable=no-member
  
                  for name in executors:
                      fname = '{0}.execute'.format(name)
                      if fname not in minion_instance.executors:
                          raise SaltInvocationError("Executor '{0}' is not available".format(name))
                      return_data = minion_instance.executors[fname](opts, data, func, args, kwargs)
                      if return_data is not None:
                          break
  
                  if isinstance(return_data, types.GeneratorType):
                      ind = 0
                      iret = {}
                      for single in return_data:
                          if isinstance(single, dict) and isinstance(iret, dict):
                              iret.update(single)
                          else:
                              if not iret:
                                  iret = []
                              iret.append(single)
                          tag = tagify([data['jid'], 'prog', opts['id'], six.text_type(ind)], 'job')
                          event_data = {'return': single}
                          minion_instance._fire_master(event_data, tag)
                          ind += 1
                      ret['return'] = iret
                  else:
                      ret['return'] = return_data
  
                  retcode = minion_instance.functions.pack['__context__'].get(
                      'retcode',
                      salt.defaults.exitcodes.EX_OK
                  )
                  if retcode == salt.defaults.exitcodes.EX_OK:
                      # No nonzero retcode in __context__ dunder. Check if return
                      # is a dictionary with a "result" or "success" key.
                      try:
                          func_result = all(return_data.get(x, True)
                                            for x in ('result', 'success'))
                      except Exception:
                          # return data is not a dict
                          func_result = True
                      if not func_result:
                          retcode = salt.defaults.exitcodes.EX_GENERIC
  
                  ret['retcode'] = retcode
                  ret['success'] = retcode == salt.defaults.exitcodes.EX_OK
              except CommandNotFoundError as exc:
                  msg = 'Command required for \'{0}\' not found'.format(
                      function_name
                  )
                  log.debug(msg, exc_info=True)
                  ret['return'] = '{0}: {1}'.format(msg, exc)
                  ret['out'] = 'nested'
                  ret['retcode'] = salt.defaults.exitcodes.EX_GENERIC
              except CommandExecutionError as exc:
                  log.error(
                      'A command in \'%s\' had a problem: %s',
                      function_name, exc,
                      exc_info_on_loglevel=logging.DEBUG
                  )
                  ret['return'] = 'ERROR: {0}'.format(exc)
                  ret['out'] = 'nested'
                  ret['retcode'] = salt.defaults.exitcodes.EX_GENERIC
              except SaltInvocationError as exc:
                  log.error(
                      'Problem executing \'%s\': %s',
                      function_name, exc,
                      exc_info_on_loglevel=logging.DEBUG
                  )
                  ret['return'] = 'ERROR executing \'{0}\': {1}'.format(
                      function_name, exc
                  )
                  ret['out'] = 'nested'
                  ret['retcode'] = salt.defaults.exitcodes.EX_GENERIC
              except TypeError as exc:
                  msg = 'Passed invalid arguments to {0}: {1}\n{2}'.format(
                      function_name, exc, func.__doc__ or ''
                  )
                  log.warning(msg, exc_info_on_loglevel=logging.DEBUG)
                  ret['return'] = msg
                  ret['out'] = 'nested'
                  ret['retcode'] = salt.defaults.exitcodes.EX_GENERIC
              except Exception:
                  msg = 'The minion function caused an exception'
                  log.warning(msg, exc_info_on_loglevel=True)
                  salt.utils.error.fire_exception(salt.exceptions.MinionError(msg), opts, job=data)
                  ret['return'] = '{0}: {1}'.format(msg, traceback.format_exc())
                  ret['out'] = 'nested'
                  ret['retcode'] = salt.defaults.exitcodes.EX_GENERIC
          else:
              docs = minion_instance.functions['sys.doc']('{0}*'.format(function_name))
              if docs:
                  docs[function_name] = minion_instance.functions.missing_fun_string(function_name)
                  ret['return'] = docs
              else:
                  ret['return'] = minion_instance.functions.missing_fun_string(function_name)
                  mod_name = function_name.split('.')[0]
                  if mod_name in minion_instance.function_errors:
                      ret['return'] += ' Possible reasons: \'{0}\''.format(
                          minion_instance.function_errors[mod_name]
                      )
              ret['success'] = False
              ret['retcode'] = salt.defaults.exitcodes.EX_GENERIC
              ret['out'] = 'nested'
  
          ret['jid'] = data['jid']
          ret['fun'] = data['fun']
          ret['fun_args'] = data['arg']
          if 'master_id' in data:
              ret['master_id'] = data['master_id']
          if 'metadata' in data:
              if isinstance(data['metadata'], dict):
                  ret['metadata'] = data['metadata']
              else:
                  log.warning('The metadata parameter must be a dictionary. Ignoring.')
          if minion_instance.connected:
              minion_instance._return_pub(
                  ret,
                  timeout=minion_instance._return_retry_timer()
              )
  
          # Add default returners from minion config
          # Should have been coverted to comma-delimited string already
          if isinstance(opts.get('return'), six.string_types):
              if data['ret']:
                  data['ret'] = ','.join((data['ret'], opts['return']))
              else:
                  data['ret'] = opts['return']
  
          log.debug('minion return: %s', ret)
          # TODO: make a list? Seems odd to split it this late :/
          if data['ret'] and isinstance(data['ret'], six.string_types):
              if 'ret_config' in data:
                  ret['ret_config'] = data['ret_config']
              if 'ret_kwargs' in data:
                  ret['ret_kwargs'] = data['ret_kwargs']
              ret['id'] = opts['id']
              for returner in set(data['ret'].split(',')):
                  try:
                      returner_str = '{0}.returner'.format(returner)
                      if returner_str in minion_instance.returners:
                          minion_instance.returners[returner_str](ret)
                      else:
                          returner_err = minion_instance.returners.missing_fun_string(returner_str)
                          log.error(
                              'Returner %s could not be loaded: %s',
                              returner_str, returner_err
                          )
                  except Exception as exc:
                      log.exception(
                          'The return failed for job %s: %s', data['jid'], exc
                      )