Sending and retrieving files from the cluster

[mpiraux]

ipyparallel does not offer means to exchange files from the controller to the cluster. In the code, there are currently a couple of rsync that I would like to get rid of for a more integrated solution.

ipyparallel only offers push and pull to retrieve objects from the global namespace of each node, from which we could build utilities for file transfer. There could be an applicative penalty in terms of RTTs involved compared to rsync but that should outweight the maintenance burden.

This is probably also a blocker for #6.

[nrybowski]

We already implemented the send operation some time ago.

mpf/mpf/__init__.py

Lines 274 to 290 in 9243b2d

	def send(role: str, content: dict):
	""" Send files to the specified role
	@param[in] content: A dictionary whose keys are paths on the host and values the file
	content to dump
	"""
	assert setup_done, f"mpf.setup must be called first"
	assert role in roles, f'role {role} not defined in roles'
	assert type(content) is dict, f'content must be a dict'
	machine_id = roles[role].machine_id
	_apply_send = """
	for dst, content in _files_to_dump.items():
	with open(dst, 'w') as fd:
	fd.write(content)
	del _files_to_dump
	"""
	client[machine_id].push({'_files_to_dump': content}, block=True)
	client[machine_id].execute(_apply_send)

[mpiraux]

I have an implem of a retrieve function but I'm reluctant merging it. We can't stop depending on rsync because we need to send the magics file before the cluster is running, thus this receive function would only spare the rsync used to copy files in %md directories. I would like to see a use case for an experiment that needs to retrieve files from roles during its setup phase. I mentioned #6, but this one could reuse rsync as it is provided by the mpf lib.

def retrieve(role: str, paths: Dict[str, str], remove_source_files=False):
    """ Retrieves files and directories from this role.
        @param[in]  paths: A dictionary mapping paths on the role host to local paths.
    """
    assert setup_done, f"mpf.setup must be called first"
    assert role in roles, f'role {role} not defined in roles'
    assert type(paths) is dict, f'paths must be a dict'
    machine_id = roles[role].machine_id
    _apply_retrieve = """
    import tempfile, shutil
    _mpf_paths_to_retrieve = {}
    _mpf_paths_is_dir = set()
    for remote_path in _mpf_retrieve_path_list:
        if os.path.isdir(remote_path):
            with tempfile.TemporaryDirectory() as tmpdir:
                with open(shutil.make_archive(os.path.join(tmpdir, 'tmp_mpf'), 'zip', remote_path), 'rb') as f:
                    _mpf_paths_to_retrieve[remote_path] = f.read()
                    _mpf_paths_is_dir.add(remote_path)
            if _mpf_paths_remove_source_files:
                shutil.rmtree(remote_path)
        elif os.path.isfile(remote_path):
            with open(remote_path, 'rb') as f:
                _mpf_paths_to_retrieve[remote_path] = f.read()
            if _mpf_paths_remove_source_files:
                os.unlink(remote_path)
        elif os.path.islink(remote_path):
            raise NotImplemented()
    """
    client[machine_id].push({'_mpf_retrieve_path_list': list(paths.keys()), '_mpf_paths_remove_source_files': remove_source_files}, block=True)
    client[machine_id].execute(_apply_retrieve)
    paths_content, paths_is_dir = client[machine_id].pull(['_mpf_paths_to_retrieve', '_mpf_paths_is_dir'], block=True)
    for path, content in paths_content:
        if path in paths_is_dir:
            zf = zipfile.ZipFile(io.BytesIO(content))
            zf.extractall(paths[path])
        else:
            with open(paths[path], 'wb') as f:
                f.write(content)

I propose to close this if no use-case shows up.