Do a function call on every worker with different arguments

        Parameters
        ----------
        fcn: funtion
            Function to call.
        args: tuple
            The arguments for Pool.map
772840
def allmap(self, fcn, args):
        
        results = self.map(_lockstep_fcn,
                           [(len(self), fcn, arg) for arg in args])
        _numdone.value = 0
        return results
1179627
def _fcall_out_type(fcall):
    
    if isinstance(fcall, FunctionSpaceElement):
        call_has_out = fcall._call_has_out
        call_out_optional = fcall._call_out_optional

    # Numpy Ufuncs and similar objects (e.g. Numba DUfuncs)
    elif hasattr(fcall, 'nin') and hasattr(fcall, 'nout'):
        if fcall.nin != 1:
            raise ValueError('ufunc {} has {} input parameter(s), '
                             'expected 1'
                             ''.format(fcall.__name__, fcall.nin))
        if fcall.nout > 1:
            raise ValueError('ufunc {} has {} output parameter(s), '
                             'expected at most 1'
                             ''.format(fcall.__name__, fcall.nout))
        call_has_out = call_out_optional = (fcall.nout == 1)
    elif inspect.isfunction(fcall):
        call_has_out, call_out_optional = _check_out_arg(fcall)
    elif callable(fcall):
        call_has_out, call_out_optional = _check_out_arg(fcall.__call__)
    else:
        raise TypeError('object {!r} not callable'.format(fcall))

    return call_has_out, call_out_optional
977859
def foreach_worker(self, fn):
        
        results = ray.get([w.foreach_worker.remote(fn) for w in self.workers])
        return results
110767
def send_if_client(fctn):
    """"""
    @functools.wraps(fctn)
    def _send_if_client(self, *args, **kwargs):
        fctn_map = {'set_quantity': 'set_value'}
        b = self._bundle
        if b is not None and b.is_client:
            # TODO: self._filter???
            # TODO: args???
            method = fctn_map.get(fctn.__name__, fctn.__name__)
            d = self._filter if hasattr(self, '_filter') \
                else {'twig': self.twig}
            d['bundleid'] = b._bundleid
            for k, v in kwargs.items():
                d[k] = v

            logger.info('emitting to {}({}) to server'.format(method, d))
            b._socketio.emit(method, d)

            if fctn.__name__ in ['run_compute', 'run_fitting']:
                # then we're expecting a quick response with an added jobparam
                # let's add that now
                self._bundle.client_update()
        else:
            return fctn(self, *args, **kwargs)
    return _send_if_client
936915
function allocifyPoolFn(fn, ResultCons) {
    return allocFn;

    function allocFn(arg1, arg2, arg3) {
        return fn(new ResultCons(), arg1, arg2, arg3);
    }
}
410711
def run_fn(self, name, *args, **kwds):
        
        fn = None
        to_check = [picardrun]
        for ns in to_check:
            try:
                fn = getattr(ns, name)
                break
            except AttributeError:
                pass
        assert fn is not None, "Could not find function %s in %s" % (name, to_check)
        return fn(self, *args, **kwds)
605119
def bind(self, fn: Callable[[Any], 'List']) -> 'List':
        
        return List.concat(self.map(fn))
106575
def lambda_tuple_converter(func):
    
    if func is not None and func.__code__.co_argcount == 1:
        return lambda *args: func(args[0] if len(args) == 1 else args)
    else:
        return func
364763
def coroutine(f):
    
    @wraps(f)
    def g(*args, **kwargs):
        sink = f(*args, **kwargs)
        sink.send(None)
        return sink

    return g
647771
def _multi_call(function, contentkey, *args, **kwargs):
    '''
    
    '''
    ret = function(*args, **kwargs)
    position = ret.get('position')

    while position:
        more = function(*args, position=position, **kwargs)
        ret[contentkey].extend(more[contentkey])
        position = more.get('position')
    return ret.get(contentkey)
832248
def bind(self, fn: Callable[[Any], 'Cont']) -> 'Cont':
        
        """
        return Cont(lambda c: self.run(lambda a: fn(a).run(c)))
1076284
def schedule(self, function, args=(), kwargs={}):
        
        self._check_pool_state()

        future = Future()
        payload = TaskPayload(function, args, kwargs)
        task = Task(next(self._task_counter), future, None, payload)

        self._context.task_queue.put(task)

        return future
128785
def duel_command(f):
    

    @functools.wraps(f)
    def func_wrapper(self, *argv, **kwargs):
        commands = _get_commands(f, self, *argv, **kwargs)
        return self.execute_dual(commands)

    return func_wrapper
1309704
def driveFunction(self):
        

        if self.function == 'HELP':
            # General help for all functions.
            self.printLn("N", "")
            self.printLn("N", "Usage:")
            self.printLn("N", "  python " + self.cmdName + " --help")
            for key in sorted(ReqHandle.funcHandler):
                ReqHandle.funcHandler[key][0](self)
            self.printLn("N", "")
            self.printLn("N", "Operand(s):")
            for key in sorted(ReqHandle.funcHandler):
                ReqHandle.funcHandler[key][1](self)
            self.printLn("N", "")
            self.updateResults({}, reset=1)
        elif self.function == 'VERSION':
            # Version of ReqHandle.
            self.printLn("N", "Version: " + version)
            self.updateResults({}, reset=1)
        else:
            # Some type of function/subfunction invocation.
            if self.function in self.funcHandler:
                # Invoke the functions doIt routine to route to the
                # appropriate subfunction.
                self.funcHandler[self.function][3](self)
            else:
                # Unrecognized function
                msg = msgs.msg['0007'][1] % (modId, self.function)
                self.printLn("ES", msg)
                self.updateResults(msgs.msg['0007'][0])

        return self.results
150403
def multi_ops(data_stream, *funcs):
    

    assert all(callable(func) for func in funcs), 'multi_ops can only apply functions to the first argument'
    assert len(funcs), 'multi_ops needs at least one function to apply to data_stream'

    for i in data_stream:
        if len(funcs) > 1:
            yield tuple(func(i) for func in funcs)
        elif len(funcs) == 1:
            yield funcs[0](i)
682332
def send(func, *args, **kwargs):
    '''
    
    '''
    kwargs = salt.utils.args.clean_kwargs(**kwargs)
    mine_func = kwargs.pop('mine_function', func)
    if mine_func not in __salt__:
        return False
    data = {}
    arg_data = salt.utils.args.arg_lookup(__salt__[mine_func])
    func_data = copy.deepcopy(kwargs)
    for ind, _ in enumerate(arg_data.get('args', [])):
        try:
            func_data[arg_data['args'][ind]] = args[ind]
        except IndexError:
            # Safe error, arg may be in kwargs
            pass
    f_call = salt.utils.args.format_call(
        __salt__[mine_func],
        func_data,
        expected_extra_kws=MINE_INTERNAL_KEYWORDS)
    for arg in args:
        if arg not in f_call['args']:
            f_call['args'].append(arg)
    try:
        if 'kwargs' in f_call:
            data[func] = __salt__[mine_func](*f_call['args'], **f_call['kwargs'])
        else:
            data[func] = __salt__[mine_func](*f_call['args'])
    except Exception as exc:
        log.error('Function %s in mine.send failed to execute: %s',
                  mine_func, exc)
        return False
    if __opts__['file_client'] == 'local':
        old = __salt__['data.get']('mine_cache')
        if isinstance(old, dict):
            old.update(data)
            data = old
        return __salt__['data.update']('mine_cache', data)
    load = {
            'cmd': '_mine',
            'data': data,
            'id': __opts__['id'],
    }
    return _mine_send(load, __opts__)
862782
def sg_producer_func(func):
    
    """
    @wraps(func)
    def wrapper(**kwargs):
        r"""Manages arguments of `tf.sg_opt`.

        Args:
          **kwargs:
            source: A source queue list to enqueue
            dtypes: Input data types of each tensor
            out_dtypes: Output data types of each tensor ( If None, same as dtypes )
            capacity: Queue capacity. Default is 32.
            num_threads: Number of threads. Default is 1.
        """

        # default option
        opt = tf.sg_opt(kwargs) + tf.sg_opt(dtypes=[tf.sg_floatx], capacity=32, num_threads=1)

        # source queue list check
        assert opt.source is not None, 'source is mandatory.'
        if type(opt.source) is not list and type(opt.source) is not tuple:
            opt.source = [opt.source]
        if type(opt.dtypes) is not list and type(opt.dtypes) is not tuple:
            opt.dtypes = [opt.dtypes]
        # default out_dtypes
        if opt.out_dtypes is None:
            opt.out_dtypes = opt.dtypes
        if type(opt.out_dtypes) is not list and type(opt.out_dtypes) is not tuple:
            opt.out_dtypes = [opt.out_dtypes]
        assert len(opt.source) == len(opt.dtypes), 'Source and dtypes should have same length.'

        # enqueue function
        def enqueue_func(sess, op):
            # read data from source queue
            data = func(sess.run(opt.source))
            # create feeder dict
            feed_dict = {}
            for ph, col in zip(placeholders, data):
                feed_dict[ph] = col
            # run session
            sess.run(op, feed_dict=feed_dict)

        # create place holder list
        placeholders = []
        for dtype in opt.dtypes:
            placeholders.append(tf.placeholder(dtype=dtype))

        # create FIFO queue
        queue = tf.FIFOQueue(opt.capacity, dtypes=opt.out_dtypes)

        # enqueue operation
        enqueue_op = queue.enqueue(placeholders)

        # create queue runner
        runner = _FuncQueueRunner(enqueue_func, queue, [enqueue_op] * opt.num_threads)

        # register to global collection
        tf.train.add_queue_runner(runner)

        # return de-queue operation
        return queue.dequeue()

    return wrapper