Apply the given function to each model replica in each worker.

        Returns:
            List of results from applying the function.
161281
def foreach_model(self, fn):
        

        results = ray.get([w.foreach_model.remote(fn) for w in self.workers])
        out = []
        for r in results:
            out.extend(r)
        return out
1218914
def collect(self):
        
        class_functions = []
        for key in self.__class__.__dict__.keys():
            func = self.__class__.__dict__[key]
            if (inspect.isfunction(func)):
                class_functions.append(func)

        functions = sorted(class_functions, key=lambda func: func.__name__)
        for function in functions:
            value = function(self)
            if value:
                return value
223757
def _execute_model(self, feed_dict, out_dict):
        
        network_out = self.sess.run(list(out_dict.values()), feed_dict=feed_dict)
        run_out = dict(zip(list(out_dict.keys()), network_out))
        return run_out
495259
def _flatten_models(tasklist):
        "  "
        ans = gvar.BufferDict()
        for task, mlist in tasklist:
            if task != 'fit':
                continue
            for m in mlist:
                id_m = id(m)
                if id_m not in ans:
                    ans[id_m] = m
        return ans.buf.tolist()
1237711
def warm_all
      threads = []
      all_functions.each do |function_name|
        next if function_name.include?('jets-public_controller') # handled by warm_public_controller_more
        next if function_name.include?('jets-rack_controller') # handled by warm_rack_controller_more
        threads << Thread.new do
          warm(function_name)
        end
      end
      threads.each { |t| t.join }

      # Warm the these controllers more since they can be hit more often
      warm_public_controller_more
      warm_rack_controller_more

      # return the funciton names so we can see in the Lambda console
      # the functions being prewarmed
      all_functions
    end
1093308
def match(self, models, results, relation):
        
        return self.match_one(models, results, relation)
759568
def apply(self, func, args=(), kwds=dict()):
        """"""
        return self.apply_async(func, args, kwds).get()
500619
def match(self, models, results, relation):
        
        return self.match_many(models, results, relation)
781568
def _process_function_list(self, routines):
    

    def processing(*args, **kwargs):
        """Execute routines with input args and kwargs and add result in
        chaining.___.

        :param tuple args: routines varargs.
        :param dict kwargs: routines kwargs.
        :return: chaining chaining.
        :rtype: Chaining
        """

        results = [None] * len(routines)
        for index, routine in enumerate(routines):
            if isinstance(routine, Exception):
                result = routine
            else:
                try:
                    result = routine(*args, **kwargs)
                except AttributeError as excp:
                    result = excp
            results[index] = result
        self.___.append(results)

        return self

    return processing
523804
def _thread_worker_fn(samples, batchify_fn, dataset):
    """"""
    if isinstance(samples[0], (list, tuple)):
        batch = [batchify_fn([dataset[i] for i in shard]) for shard in samples]
    else:
        batch = batchify_fn([dataset[i] for i in samples])
    return batch
714542
def submit_all(self):
        
        for args in self.task_args:
            self.submit(*args)
        return self.get_results()
776834
def do_function(self, prov, func, kwargs):
        '''
        
        '''
        matches = self.lookup_providers(prov)
        if len(matches) > 1:
            raise SaltCloudSystemExit(
                'More than one results matched \'{0}\'. Please specify '
                'one of: {1}'.format(
                    prov,
                    ', '.join([
                        '{0}:{1}'.format(alias, driver) for
                        (alias, driver) in matches
                    ])
                )
            )

        alias, driver = matches.pop()
        fun = '{0}.{1}'.format(driver, func)
        if fun not in self.clouds:
            raise SaltCloudSystemExit(
                'The \'{0}\' cloud provider alias, for the \'{1}\' driver, does '
                'not define the function \'{2}\''.format(alias, driver, func)
            )

        log.debug(
            'Trying to execute \'%s\' with the following kwargs: %s',
            fun, kwargs
        )

        with salt.utils.context.func_globals_inject(
            self.clouds[fun],
            __active_provider_name__=':'.join([alias, driver])
        ):
            if kwargs:
                return {
                    alias: {
                        driver: self.clouds[fun](
                            call='function', kwargs=kwargs
                        )
                    }
                }
            return {
                alias: {
                    driver: self.clouds[fun](call='function')
                }
            }
946549
def _jit_update_fun(predict_fun, loss_fun, optimizer, lr_fun, num_devices):
  """"""
  if num_devices == 1:  # TODO(lukaszkaiser): remove branch when not needed.
    def single_update(i, opt_state, batch, rng):
      rng, subrng = jax_random.split(rng[0])
      _, opt_update = optimizer(lr_fun)
      params = trax_opt.get_params(opt_state)
      return opt_update(i, backend.grad(loss_fun)(
          params, batch, predict_fun, rng), opt_state), [subrng]
    return backend.jit(single_update)

  @functools.partial(backend.pmap, axis_name="batch")
  def mapped_update(i, opt_state, batch, rng):
    """This is a multi-device version of the update function above."""
    # We assume all tensors have the first dimension = num_devices.
    rng, subrng = jax_random.split(rng)
    _, opt_update = optimizer(lr_fun)
    params = trax_opt.get_params(opt_state)
    grads = backend.grad(loss_fun)(params, batch, predict_fun, rng)
    grads = jax.tree_util.tree_map(
        lambda g: lax.psum(g, "batch"), grads)
    return opt_update(i, grads, opt_state), subrng

  def update(i, opt_state, batch, rng):
    return mapped_update(jax.replicate(i), opt_state, batch, rng)

  return update
477503
def parallelize_func(iterable, func, chunksz=1, n_jobs=16, *args, **kwargs):
    
    chunker = func
    chunks = more_itertools.chunked(iterable, chunksz)
    chunks_results = Parallel(n_jobs=n_jobs, verbose=50)(
        delayed(chunker)(chunk, *args, **kwargs) for chunk in chunks)
    results = more_itertools.flatten(chunks_results)
    return list(results)
1369137
def update(self):
        
        ring = self._fetch()
        n_replicas = len(ring)
        replica_set = set([r[1] for r in self.replicas])
        self.ranges = []
        for n, (start, replica) in enumerate(ring):
            if replica in replica_set:
                end = ring[(n+1) % n_replicas][0] % RING_SIZE
                if start < end:
                    self.ranges.append((start, end))
                elif end < start:
                    self.ranges.append((start, RING_SIZE))
                    self.ranges.append((0, end))
                else:
                    self.ranges.append((0, RING_SIZE))
538104
def iofunctions(self):
        """"""
        lines = Lines()
        for func in ('open_files', 'close_files', 'load_data', 'save_data'):
            if ((func == 'load_data') and
                    (getattr(self.model.sequences, 'inputs', None) is None)):
                continue
            if ((func == 'save_data') and
                ((getattr(self.model.sequences, 'fluxes', None) is None) and
                 (getattr(self.model.sequences, 'states', None) is None))):
                continue
            print('            . %s' % func)
            nogil = func in ('load_data', 'save_data')
            idx_as_arg = func == 'save_data'
            lines.add(1, method_header(
                func, nogil=nogil, idx_as_arg=idx_as_arg))
            for subseqs in self.model.sequences:
                if func == 'load_data':
                    applyfuncs = ('inputs',)
                elif func == 'save_data':
                    applyfuncs = ('fluxes', 'states')
                else:
                    applyfuncs = ('inputs', 'fluxes', 'states')
                if subseqs.name in applyfuncs:
                    if func == 'close_files':
                        lines.add(2, 'self.sequences.%s.%s()'
                                     % (subseqs.name, func))
                    else:
                        lines.add(2, 'self.sequences.%s.%s(self.idx_sim)'
                                     % (subseqs.name, func))
        return lines
958165
def _apply_2d(self, func, axis):
        
        ndim = self.ndim
        axis = [self._get_axis_number(a) for a in axis]

        # construct slabs, in 2-d this is a DataFrame result
        indexer_axis = list(range(ndim))
        for a in axis:
            indexer_axis.remove(a)
        indexer_axis = indexer_axis[0]

        slicer = [slice(None, None)] * ndim
        ax = self._get_axis(indexer_axis)

        results = []
        for i, e in enumerate(ax):
            slicer[indexer_axis] = i
            sliced = self.iloc[tuple(slicer)]

            obj = func(sliced)
            results.append((e, obj))

        return self._construct_return_type(dict(results))