Split data in `a` equally among `n_cpus` cores
321383
def partition_by_cores(a:Collection, n_cpus:int)->List[Collection]:
    ""
    return partition(a, len(a)//n_cpus + 1)
995652
def parmap(f, X, nprocs=multiprocessing.cpu_count()):
    
    q_in = multiprocessing.Queue(1)
    q_out = multiprocessing.Queue()

    proc = [multiprocessing.Process(target=fun, args=(f, q_in, q_out))
            for _ in range(nprocs)]
    for p in proc:
        p.daemon = True
        p.start()

    sent = [q_in.put((i, x)) for i, x in enumerate(X)]
    [q_in.put((None, None)) for _ in range(nprocs)]
    res = [q_out.get() for _ in range(len(sent))]

    [p.join() for p in proc]

    return [x for i, x in sorted(res)]
453164
def run(items, run_parallel):
    
    to_process = []
    extras = []
    for batch, cur_items in _group_by_batches(items).items():
        if _ready_for_het_analysis(cur_items):
            to_process.append((batch, cur_items))
        else:
            for data in cur_items:
                extras.append([data])
    processed = run_parallel("heterogeneity_estimate", ([xs, b, xs[0]["config"]] for b, xs in to_process))
    return _group_by_sample_and_batch(extras + processed)
390023
def _multi_rpush_pipeline(self, pipe, queue, values, bulk_size=0):
        ''' 
        '''
        cont = 0
        for value in values:
            pipe.rpush(queue, value)
            if bulk_size != 0 and cont % bulk_size == 0:
                pipe.execute()
905981
def enable_parallel(processnum=None):
    
    global pool, dt, cut, cut_for_search
    from multiprocessing import cpu_count
    if os.name == 'nt':
        raise NotImplementedError(
            "jieba: parallel mode only supports posix system")
    else:
        from multiprocessing import Pool
    dt.check_initialized()
    if processnum is None:
        processnum = cpu_count()
    pool = Pool(processnum)
    cut = _pcut
    cut_for_search = _pcut_for_search
868992
def merge(dicts):
    
    chunks = [args for args in np.array_split(dicts, 20)]
    results = parallel(_count, chunks, n_jobs=-1)
    return _count(results)
341763
def _get_num_parallel_threads(max_threads, num_cores, mem_available_mb):
    '''
    
    '''
    return min(max_threads, num_cores, max(int(mem_available_mb/1200), 1))
304748
def divide_by(array_list, num_workers):
    
    for i, x in enumerate(array_list):
        array_list[i] /= num_workers
    return array_list
340938
def _get_cores_and_type(numcores, paralleltype, scheduler):
    
    if scheduler is not None:
        paralleltype = "ipython"
    if paralleltype is None:
        paralleltype = "local"
    if not numcores or int(numcores) < 1:
        numcores = 1
    return paralleltype, int(numcores)
1324306
def get_parallel_raw_data(self, other):
        
        start, end = other.byte_bounds_offset()
        r = self.rawdata[start:end]
        if other.rawdata.is_indexed:
            r = r.get_indexed[other.order]
        return r
964053
def cli(env, sortby, cpu, domain, datacenter, hostname, memory, network, tag, columns, limit):
    """"""

    manager = SoftLayer.HardwareManager(env.client)
    servers = manager.list_hardware(hostname=hostname,
                                    domain=domain,
                                    cpus=cpu,
                                    memory=memory,
                                    datacenter=datacenter,
                                    nic_speed=network,
                                    tags=tag,
                                    mask="mask(SoftLayer_Hardware_Server)[%s]" % columns.mask(),
                                    limit=limit)

    table = formatting.Table(columns.columns)
    table.sortby = sortby

    for server in servers:
        table.add_row([value or formatting.blank()
                       for value in columns.row(server)])

    env.fout(table)
1009362
def _parameter_sweep(self, parameter_space, kernel_options, device_options, tuning_options):
        """"""
        results = []

        #randomize parameter space to do pseudo load balancing
        parameter_space = list(parameter_space)
        random.shuffle(parameter_space)

        #split parameter space into chunks
        work_per_thread = int(numpy.ceil(len(parameter_space) / float(self.max_threads)))
        chunks = _chunk_list(parameter_space, work_per_thread)

        for chunk in chunks:

            chunked_result = self._run_chunk(chunk, kernel_options, device_options, tuning_options)

            results.append(lift(chunked_result))

        return gather(*results)
565343
def cpu_count(logical=True):
    
    if logical:
        from multiprocessing import cpu_count
        ncpu=cpu_count()
    else:
        import psutil
        ncpu=psutil.cpu_count(logical=False)
    return ncpu
282411
def _build_pcollection(self, pipeline, folder, split):
    """"""
    beam = tfds.core.lazy_imports.apache_beam

    split_type = self.builder_config.split_type
    filename = os.path.join(folder, "{}.tar.gz".format(split_type))


    def _extract_data(inputs):
      """Extracts files from the tar archives."""
      filename, split = inputs
      with tf.io.gfile.GFile(filename, "rb") as f:
        with tarfile.open(fileobj=f, mode="r") as tar:
          for tarinfo in tar:
            split_name = tarinfo.name.split("_")
            if len(split_name) > 2 and split_name[2] == split:
              buf = six.BytesIO()
              shutil.copyfileobj(tar.extractfile(tarinfo), buf)
              yield [tarinfo.name, buf.getvalue()]

    def _process_example(inputs):
      filename, data_string = inputs
      buf = six.BytesIO(data_string)
      buf.seek(0)
      data = np.load(buf)
      # Extract the images and convert to uint8. The reshape is required, see
      # https://github.com/deepmind/abstract-reasoning-matrices.
      all_images = np.uint8(data["image"].reshape(16, 160, 160, 1))
      return {
          "relation_structure_encoded": data["relation_structure_encoded"],
          "target": data["target"],
          "meta_target": data["meta_target"],
          "context": all_images[:8],
          "answers": all_images[8:],
          "filename": filename,
      }

    # Beam might fuse together the _extract_data and _process_example which
    # defeats the purpose of parallel processing. As a result, we reshard by
    # doing a GroupByKey on random keys, and then flattening again.
    def _add_random_keys(inputs):
      key = str(random.randrange(10**10))
      return key, inputs

    def _remove_keys(inputs):
      _, rows = inputs
      for row in rows:
        yield row

    return (pipeline
            | beam.Create([(filename, split)])
            | beam.FlatMap(_extract_data)
            | beam.Map(_add_random_keys)
            | beam.GroupByKey()
            | beam.FlatMap(_remove_keys)
            | beam.Map(_process_example))
559130
def split_results(results):
    """"""
    from dask.delayed import Delayed

    def flatten(results):
        out = []
        if isinstance(results, (list, tuple)):
            for itm in results:
                out.extend(flatten(itm))
            return out
        return [results]

    sources = []
    targets = []
    delayeds = []

    for res in flatten(results):
        if isinstance(res, da.Array):
            sources.append(res)
        elif isinstance(res, Delayed):
            delayeds.append(res)
        else:
            targets.append(res)
    return sources, targets, delayeds
92186
def parallel_fn(f):
    

    def simple_parallel(func, sequence, **args):
        """ f takes an element of sequence as input and the keyword args in **args"""
        if 'processes' in args:
            processes = args.get('processes')
            del args['processes']
        else:
            processes = multiprocessing.cpu_count()

        pool = multiprocessing.Pool(processes)  # depends on available cores

        result = pool.map_async(universal_worker, pool_args(func, sequence, args))
        pool.close()
        pool.join()
        cleaned = [x for x in result.get() if x is not None]  # getting results
        cleaned = asarray(cleaned)
        return cleaned
    return partial(simple_parallel, f)
569259
def data_parallelism(daisy_chain_variables=True,
                     all_workers=False,
                     ps_replicas=0,
                     ps_job="/job:ps",
                     ps_gpu=0,
                     schedule="continuous_train_and_eval",
                     sync=False,
                     worker_gpu=1,
                     worker_replicas=1,
                     worker_id=0,
                     gpu_order="",
                     worker_job="/job:localhost",
                     no_data_parallelism=False):
  """"""
  tf.logging.info("schedule=%s" % schedule)
  tf.logging.info("worker_gpu=%s" % worker_gpu)
  tf.logging.info("sync=%s" % sync)
  def _ps_replicas(all_workers=False):
    if all_workers:
      return list(range(ps_replicas))
    # Worker K will be using replicas {0,...n-1} + K*n if we have n replicas.
    num_replicas = ps_replicas // worker_replicas
    return [d + worker_id * num_replicas for d in range(num_replicas)]

  def _gpu_order(num_gpus):
    if gpu_order:
      ret = [int(s) for s in gpu_order.split(" ")]
      if len(ret) == num_gpus:
        return ret
    return list(range(num_gpus))

  def _ps_gpus(all_workers=False):
    ps_gpus = []
    for d in _ps_replicas(all_workers=all_workers):
      ps_gpus.extend([(d, gpu) for gpu in _gpu_order(ps_gpu)])
    return ps_gpus

  def ps_devices(all_workers=False):
    """List of ps devices (where to put the experts).

    Args:
      all_workers: whether the list is for all async workers or just this one.

    Returns:
      a list of device names
    """
    if ps_replicas > 0:
      if ps_gpu > 0:
        return [
            ps_job + "/task:%d/GPU:%d" % (d, gpu)
            for (d, gpu) in _ps_gpus(all_workers=all_workers)
        ]
      else:
        return [
            ps_job + "/task:%d" % d
            for d in _ps_replicas(all_workers=all_workers)
        ]
    else:
      if worker_gpu > 0:
        return ["gpu:%d" % d for d in _gpu_order(worker_gpu)]
      else:
        return [""]

  def _replica_device_setter(worker_device):
    if ps_replicas == 0:
      return worker_device
    return tf.train.replica_device_setter(
        worker_device=worker_device,
        ps_tasks=ps_replicas,
        ps_device=ps_job + "/GPU:0" if ps_gpu > 0 else ps_job)

  is_single_machine = ps_replicas == 0 and worker_replicas == 1

  if no_data_parallelism:
    datashard_devices = [""]
    caching_devices = None
  elif is_single_machine:
    tf.logging.warn(
        "Schedule=%s. Assuming that training is running on a single machine.",
        schedule)
    datashard_devices = ["gpu:%d" % d for d in _gpu_order(worker_gpu)]
    if worker_gpu < 1:
      datashard_devices += ["cpu:0"]
    caching_devices = None
  elif sync and ps_replicas > 0:
    # compute on ps
    datashard_devices = [
        _replica_device_setter(d) for d in ps_devices(all_workers=all_workers)
    ]
    if ps_gpu > 0 and ps_replicas > 1:
      caching_devices = [
          ps_job + "/task:%d/cpu:0" % d
          for (d, _) in _ps_gpus(all_workers=all_workers)
      ]
    else:
      caching_devices = None
  else:
    # compute on worker - this is either a single-worker setup or asynchronous
    # with parameter servers.
    if worker_gpu > 1:
      datashard_devices = [
          _replica_device_setter(worker_job + "/GPU:%d" % d)
          for d in _gpu_order(worker_gpu)
      ]
      caching_devices = None
    else:
      datashard_devices = [_replica_device_setter(worker_job)]
      caching_devices = None
  tf.logging.info("datashard_devices: %s", datashard_devices)
  tf.logging.info("caching_devices: %s", caching_devices)
  tf.logging.info("ps_devices: %s", ps_devices(all_workers=all_workers))
  return eu.Parallelism(
      datashard_devices,
      caching_devices=caching_devices,
      daisy_chain_variables=daisy_chain_variables,
      ps_devices=ps_devices(all_workers=all_workers))