<jupyter_start><jupyter_text>More efficient broadcast of arrays with memmap Data movement is where IPython's naïve model suffers the most.But knowing about your cluster lets you make smarter decisions about data movement than a simple `rc[:].push`.<jupyter_code>import socket
import os, sys, re

import numpy as np

import ipyparallel as parallel
#rc = parallel.Client(profile='dirac')
rc = parallel.Client()
eall = rc[:]
engine_hosts = eall.apply_async(socket.gethostname).get_dict()
engine_hosts
host_engines = {}

for eid, host in engine_hosts.items():
    if host not in host_engines:
        host_engines[host] = []
    host_engines[host].append(eid)

host_engines
sz = 256
data = np.random.random((sz,sz))
data = data.dot(data.T)
%time _ = rc[:].apply_sync(lambda : None)
ar = rc[:].push({'data': data}, block=False)
ar.wait_interactive()
%px import numpy as np
def array_to_file(A):
    """write an array to a temporary file, return its filename"""
    import tempfile
    with tempfile.NamedTemporaryFile(suffix='.np', delete=False) as tf:
        np.save(tf, data)
        data_path = tf.name
    return data_path
@parallel.interactive
def load_memmap(name, path, mode='r+'):
    """load a file on disk into the interactive namespace as a memmapped array"""
    globals()[name] = np.memmap(path, mode=mode)
def bcast_memmap(data, name, client, host_engines):
    """broadcast a numpy array efficiently
    
    - sends data to each remote host only once
    - loads with memmap everywhere
    """

    # actually push the data, just once to each machine

    local_filename = None
    filenames_ars = {}
    for host, engines in host_engines.items():
        h0 = engines[0]
        if host == socket.gethostname():
            # Don't push at all to local engines
            local_filename = array_to_file(data)
        else:
            filenames_ars[host] = rc[h0].apply_async(array_to_file, data)

    # load the data on all engines into a memmapped array
    msg_ids = []
    for host, engines in host_engines.items():
        if host == socket.gethostname():
            filename = local_filename
        else:
            filename = filenames_ars[host].get()
        ar = rc[engines].apply_async(load_memmap, name, filename)
        msg_ids.extend(ar.msg_ids)
    
    return parallel.AsyncResult(client, msg_ids=msg_ids)
%%time
ar = bcast_memmap(data, 'data', rc, host_engines)
ar.wait_interactive()
%px np.linalg.norm(data, 2)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10747474747474747, 'fraction_numerical': 0.0028282828282828283, 'mean_word_length': 3.543119266055046, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '512842', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 794, 'n_tokens_neox': 781, 'n_words': 233, 'path': 'parallel/examples/memmap.ipynb', 'repo_name': 'minrk/ipython-cse17', 'size': '2475'}