This method should be used as a threading target, start the actual
        minion side execution.

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
                    )