Execute the function in a multiprocess and return the event tag to use
        to watch for the return
346761
def asynchronous(self, fun, low, user='UNKNOWN', pub=None):
        '''
        
        '''
        async_pub = pub if pub is not None else self._gen_async_pub()

        proc = salt.utils.process.SignalHandlingMultiprocessingProcess(
                target=self._proc_function,
                args=(fun, low, user, async_pub['tag'], async_pub['jid']))
        with salt.utils.process.default_signals(signal.SIGINT, signal.SIGTERM):
            # Reset current signals before starting the process in
            # order not to inherit the current signal handlers
            proc.start()
        proc.join()  # MUST join, otherwise we leave zombies all over
        return async_pub
1069642
def cmd_full_return(
            self,
            tgt,
            fun,
            arg=(),
            timeout=None,
            tgt_type='glob',
            ret='',
            verbose=False,
            kwarg=None,
            **kwargs):
        '''
        
        '''
        was_listening = self.event.cpub

        try:
            pub_data = self.run_job(
                tgt,
                fun,
                arg,
                tgt_type,
                ret,
                timeout,
                kwarg=kwarg,
                listen=True,
                **kwargs)

            if not pub_data:
                return pub_data

            return (self.get_cli_static_event_returns(pub_data['jid'],
                                                      pub_data['minions'],
                                                      timeout,
                                                      tgt,
                                                      tgt_type,
                                                      verbose))
        finally:
            if not was_listening:
                self.event.close_pub()
1074549
def wait_for_event(self, event, timeout=10):
        
        return self.__handler.wait_for_event(event, timeout=timeout)
699640
def execute(self):
        
        func_params = []
        exec_str = self.func.__name__ + '(' 
        for p in self.params:
            if p[0][0:2] != '__':   # ignore custom param names
                exec_str += p[0] + '="' + self._force_str(p[1]) + '", '
                func_params.append(p[1])
        exec_str = exec_str[:-2]
        exec_str += ')  # task' + str(self.task_id) + ': ' + self.name
        
        
        self.result = self.func(*func_params)
        print(exec_str + ' loaded ', self.result)
1175796
def run(self):
        '''
        
        '''
        def receive():
            if self.with_communicate:
                self.stdout, self.stderr = self.process.communicate(input=self.stdin)
            elif self.wait:
                self.process.wait()

        if not self.timeout:
            receive()
        else:
            rt = threading.Thread(target=receive)
            rt.start()
            rt.join(self.timeout)
            if rt.isAlive():
                # Subprocess cleanup (best effort)
                self.process.kill()

                def terminate():
                    if rt.isAlive():
                        self.process.terminate()
                threading.Timer(10, terminate).start()
                raise salt.exceptions.TimedProcTimeoutError(
                    '{0} : Timed out after {1} seconds'.format(
                        self.command,
                        six.text_type(self.timeout),
                    )
                )
        return self.process.returncode
202380
def process(self):
        ''''''
        for tag_proc in self.tag_procs:
            before_count = self.entry_count
            self.run_tag_processor(tag_proc)
            after_count = self.entry_count

            if self.verbose:
                print('Inserted %d entries for "%s" tag processor' % (
                    after_count - before_count, tag_proc), file=sys.stderr)

        if self.verbose:
            print('Inserted %d entries overall' % self.entry_count,
                  file=sys.stderr)
726564
async def _thread_coro(self, *args):
        
        return await self._loop.run_in_executor(
            self._executor, self._function, *args)
356910
def event_return(events):
    '''
    
    '''

    for event in events:
        ret = event.get('data', False)

        if ret:
            returner(ret)
480418
def event_return(events):
    '''
    
    '''
    conn = _get_conn()
    if conn is None:
        return None
    cur = conn.cursor()
    for event in events:
        tag = event.get('tag', '')
        data = event.get('data', '')
        sql = '''INSERT INTO salt_events
                (tag, data, master_id)
                VALUES (%s, %s, %s)'''
        cur.execute(sql, (tag, salt.utils.json.dumps(data), __opts__['id']))
    _close_conn(conn)
265473
def wait_for_event(event):
    
    f = Future()
    def ready():
        get_event_loop().remove_win32_handle(event)
        f.set_result(None)
    get_event_loop().add_win32_handle(event, ready)
    return f
891217
def mp_check_impl(self, process_count):
        

        from multiprocessing import Process, Queue

        options = self.reporter.options

        # this is the function that will be run in a separate process,
        # which will handle the tasks queue and feed into the results
        # queue
        func = _mp_run_check

        # normally this would happen lazily, but since we'll have
        # multiple processes all running reports at the same time, we
        # need to make sure the setup is done before-hand. This is
        # hackish, but in particular this keeps the HTML reports from
        # trying to perform the default data copy over and over.
        self.reporter.setup()

        # enqueue the sub-reports for multi-processing. Other types of
        # changes can happen sync.
        changes = list(self.collect_impl())

        task_count = 0
        tasks = Queue()
        results = Queue()

        try:
            # as soon as we start using the tasks queue, we need to be
            # catching the KeyboardInterrupt event so that we can
            # drain the queue and lets its underlying thread terminate
            # happily.

            # TODO: is there a better way to handle this shutdown
            # gracefully?

            # feed any sub-reports to the tasks queue
            for index in range(0, len(changes)):
                change = changes[index]
                if isinstance(change, (DistJarReport, DistClassReport)):
                    changes[index] = None
                    tasks.put((index, change))
                    task_count += 1

            # infrequent edge case, but don't bother starting more
            # helpers than we'll ever use
            process_count = min(process_count, task_count)

            # start the number of helper processes, and make sure
            # there are that many stop sentinels at the end of the
            # tasks queue
            for _i in range(0, process_count):
                tasks.put(None)
                process = Process(target=func, args=(tasks, results, options))
                process.daemon = False
                process.start()

            # while the helpers are running, perform our checks
            for change in changes:
                if change:
                    change.check()

            # get all of the results and feed them back into our change
            for _i in range(0, task_count):
                index, change = results.get()
                changes[index] = change

        except KeyboardInterrupt:
            # drain the tasks queue so it will exit gracefully
            for _change in iter(tasks.get, None):
                pass
            raise

        # complete the check by setting our internal collection of
        # child changes and returning our overall status
        c = False
        for change in changes:
            c = c or change.is_change()
        self.changes = changes
        return c, None
833305
def _call_event(self, event_name, cmd, pargs, kwargs, **kws):
        
        def get_result_params(res):
            """return the right list of params"""
            if not isinstance(res, (list, tuple)):
                return res, pargs, kwargs
            elif len(res) == 2:
                return res, pargs, kwargs
            return res[0], (pargs[0], ) + tuple(res[1]), kwargs
        if hasattr(cmd, event_name):
            return get_result_params(
                getattr(cmd, event_name)(pargs[0], *pargs[1:], **kwargs))
        elif hasattr(cmd.__cls__, event_name):
            return get_result_params(
                getattr(cmd.__cls__, event_name)(
                    pargs[0], cmd.__cmd_name__ or cmd.__name__, *pargs[1:],
                    **kwargs))
        return None, pargs, kwargs
479757
def _main(self):
        
        self.set_proctitle(self.name)
        self.set_signal_handler()

        logger.info("process for module %s is now running (pid=%d)", self.name, os.getpid())

        # Will block here!
        try:
            self.main()
        except (IOError, EOFError):
            pass
            # logger.warning('[%s] EOF exception: %s', self.name, traceback.format_exc())
        except Exception as exp:  # pylint: disable=broad-except
            logger.exception('main function exception: %s', exp)

        self.do_stop()

        logger.info("process for module %s is now exiting (pid=%d)", self.name, os.getpid())
        exit()
3626
def handle_func(self, multiprocessing_enabled, func, data):
        '''
        
        '''
        if salt.utils.platform.is_windows() \
                or self.opts.get('transport') == 'zeromq':
            # Since function references can't be pickled and pickling
            # is required when spawning new processes on Windows, regenerate
            # the functions and returners.
            # This also needed for ZeroMQ transport to reset all functions
            # context data that could keep paretns connections. ZeroMQ will
            # hang on polling parents connections from the child process.
            if self.opts['__role'] == 'master':
                self.functions = salt.loader.runner(self.opts, utils=self.utils)
            else:
                self.functions = salt.loader.minion_mods(self.opts, proxy=self.proxy, utils=self.utils)
            self.returners = salt.loader.returners(self.opts, self.functions, proxy=self.proxy)
        ret = {'id': self.opts.get('id', 'master'),
               'fun': func,
               'fun_args': [],
               'schedule': data['name'],
               'jid': salt.utils.jid.gen_jid(self.opts)}

        if 'metadata' in data:
            if isinstance(data['metadata'], dict):
                ret['metadata'] = data['metadata']
                ret['metadata']['_TOS'] = self.time_offset
                ret['metadata']['_TS'] = time.ctime()
                ret['metadata']['_TT'] = time.strftime('%Y %B %d %a %H %m', time.gmtime())
            else:
                log.warning('schedule: The metadata parameter must be '
                            'specified as a dictionary.  Ignoring.')

        if multiprocessing_enabled:
            # We just want to modify the process name if we're on a different process
            salt.utils.process.appendproctitle('{0} {1}'.format(self.__class__.__name__, ret['jid']))
        data_returner = data.get('returner', None)

        if not self.standalone:
            proc_fn = os.path.join(
                salt.minion.get_proc_dir(self.opts['cachedir']),
                ret['jid']
            )

        if multiprocessing_enabled and not salt.utils.platform.is_windows():
            # Reconfigure multiprocessing logging after daemonizing
            log_setup.setup_multiprocessing_logging()

        if multiprocessing_enabled:
            # Don't *BEFORE* to go into try to don't let it triple execute the finally section.
            salt.utils.process.daemonize_if(self.opts)

        # TODO: Make it readable! Splt to funcs, remove nested try-except-finally sections.
        try:

            minion_blackout_violation = False
            if self.opts.get('pillar', {}).get('minion_blackout', False):
                whitelist = self.opts.get('pillar', {}).get('minion_blackout_whitelist', [])
                # this minion is blacked out. Only allow saltutil.refresh_pillar and the whitelist
                if func != 'saltutil.refresh_pillar' and func not in whitelist:
                    minion_blackout_violation = True
            elif self.opts.get('grains', {}).get('minion_blackout', False):
                whitelist = self.opts.get('grains', {}).get('minion_blackout_whitelist', [])
                if func != 'saltutil.refresh_pillar' and func 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.')

            ret['pid'] = os.getpid()

            if not self.standalone:
                if 'jid_include' not in data or data['jid_include']:
                    log.debug(
                        'schedule.handle_func: adding this job to the '
                        'jobcache with data %s', ret
                    )
                    # write this to /var/cache/salt/minion/proc
                    with salt.utils.files.fopen(proc_fn, 'w+b') as fp_:
                        fp_.write(salt.payload.Serial(self.opts).dumps(ret))

            args = tuple()
            if 'args' in data:
                args = data['args']
                ret['fun_args'].extend(data['args'])

            kwargs = {}
            if 'kwargs' in data:
                kwargs = data['kwargs']
                ret['fun_args'].append(copy.deepcopy(kwargs))

            if func not in self.functions:
                ret['return'] = self.functions.missing_fun_string(func)
                salt.utils.error.raise_error(
                    message=self.functions.missing_fun_string(func))

            # if the func support **kwargs, lets pack in the pub data we have
            # TODO: pack the *same* pub data as a minion?
            argspec = salt.utils.args.get_function_argspec(self.functions[func])
            if argspec.keywords:
                # this function accepts **kwargs, pack in the publish data
                for key, val in six.iteritems(ret):
                    if key is not 'kwargs':
                        kwargs['__pub_{0}'.format(key)] = copy.deepcopy(val)

            # Only include these when running runner modules
            if self.opts['__role'] == 'master':
                jid = salt.utils.jid.gen_jid(self.opts)
                tag = salt.utils.event.tagify(jid, prefix='salt/scheduler/')

                event = salt.utils.event.get_event(
                        self.opts['__role'],
                        self.opts['sock_dir'],
                        self.opts['transport'],
                        opts=self.opts,
                        listen=False)

                namespaced_event = salt.utils.event.NamespacedEvent(
                    event,
                    tag,
                    print_func=None
                )

                func_globals = {
                    '__jid__': jid,
                    '__user__': salt.utils.user.get_user(),
                    '__tag__': tag,
                    '__jid_event__': weakref.proxy(namespaced_event),
                }
                self_functions = copy.copy(self.functions)
                salt.utils.lazy.verify_fun(self_functions, func)

                # Inject some useful globals to *all* the function's global
                # namespace only once per module-- not per func
                completed_funcs = []

                for mod_name in six.iterkeys(self_functions):
                    if '.' not in mod_name:
                        continue
                    mod, _ = mod_name.split('.', 1)
                    if mod in completed_funcs:
                        continue
                    completed_funcs.append(mod)
                    for global_key, value in six.iteritems(func_globals):
                        self.functions[mod_name].__globals__[global_key] = value

            self.functions.pack['__context__']['retcode'] = 0

            ret['return'] = self.functions[func](*args, **kwargs)

            if not self.standalone:
                # runners do not provide retcode
                if 'retcode' in self.functions.pack['__context__']:
                    ret['retcode'] = self.functions.pack['__context__']['retcode']

                ret['success'] = True

                if data_returner or self.schedule_returner:
                    if 'return_config' in data:
                        ret['ret_config'] = data['return_config']
                    if 'return_kwargs' in data:
                        ret['ret_kwargs'] = data['return_kwargs']
                    rets = []
                    for returner in [data_returner, self.schedule_returner]:
                        if isinstance(returner, six.string_types):
                            rets.append(returner)
                        elif isinstance(returner, list):
                            rets.extend(returner)
                    # simple de-duplication with order retained
                    for returner in OrderedDict.fromkeys(rets):
                        ret_str = '{0}.returner'.format(returner)
                        if ret_str in self.returners:
                            self.returners[ret_str](ret)
                        else:
                            log.info(
                                'Job %s using invalid returner: %s. Ignoring.',
                                func, returner
                            )

        except Exception:
            log.exception('Unhandled exception running %s', ret['fun'])
            # Although catch-all exception handlers are bad, the exception here
            # is to let the exception bubble up to the top of the thread context,
            # where the thread will die silently, which is worse.
            if 'return' not in ret:
                ret['return'] = "Unhandled exception running {0}".format(ret['fun'])
            ret['success'] = False
            ret['retcode'] = 254
        finally:
            # Only attempt to return data to the master if the scheduled job is running
            # on a master itself or a minion.
            if '__role' in self.opts and self.opts['__role'] in ('master', 'minion'):
                # The 'return_job' option is enabled by default even if not set
                if 'return_job' in data and not data['return_job']:
                    pass
                else:
                    # Send back to master so the job is included in the job list
                    mret = ret.copy()
                    # No returners defined, so we're only sending back to the master
                    if not data_returner and not self.schedule_returner:
                        mret['jid'] = 'req'
                        if data.get('return_job') == 'nocache':
                            # overwrite 'req' to signal to master that
                            # this job shouldn't be stored
                            mret['jid'] = 'nocache'
                    load = {'cmd': '_return', 'id': self.opts['id']}
                    for key, value in six.iteritems(mret):
                        load[key] = value

                    if '__role' in self.opts and self.opts['__role'] == 'minion':
                        event = salt.utils.event.get_event('minion',
                                                           opts=self.opts,
                                                           listen=False)
                    elif '__role' in self.opts and self.opts['__role'] == 'master':
                        event = salt.utils.event.get_master_event(self.opts,
                                                                  self.opts['sock_dir'])
                    try:
                        event.fire_event(load, '__schedule_return')
                    except Exception as exc:
                        log.exception('Unhandled exception firing __schedule_return event')

            if not self.standalone:
                log.debug('schedule.handle_func: Removing %s', proc_fn)

                try:
                    os.unlink(proc_fn)
                except OSError as exc:
                    if exc.errno == errno.EEXIST or exc.errno == errno.ENOENT:
                        # EEXIST and ENOENT are OK because the file is gone and that's what
                        # we wanted
                        pass
                    else:
                        log.error("Failed to delete '%s': %s", proc_fn, exc.errno)
                        # Otherwise, failing to delete this file is not something
                        # we can cleanly handle.
                        raise
                finally:
                    if multiprocessing_enabled:
                        # Let's make sure we exit the process!
                        sys.exit(salt.defaults.exitcodes.EX_GENERIC)
693840
def check_event_loop():
        
        loop = asyncio.get_event_loop()
        if loop.is_closed():
            asyncio.set_event_loop(asyncio.new_event_loop())
863291
def execute(function, *args, **kwargs):
    """"""
    try:
        return function(*args, **kwargs)
    except Exception as error:
        error.traceback = format_exc()
        return error
129350
def _execute_workflow(cls, process_args):
        
        workflow, input_args, log_path = process_args

        if log_path:
            logger = logging.getLogger()
            logger.setLevel(logging.DEBUG)
            handler = cls._get_log_handler(log_path)
            logger.addHandler(handler)

        stats = {'start_time': dt.datetime.now()}
        try:
            _ = workflow.execute(input_args, monitor=True)
        except BaseException:
            stats['error'] = traceback.format_exc()
        stats['end_time'] = dt.datetime.now()

        if log_path:
            handler.close()
            logger.removeHandler(handler)

        return stats