parallel Namespace Reference

Functions
def	prepare_cmds (pref_parallel)
None	run (shell_cmd, pref, k)
def	main ()

Function Documentation

main()

def parallel.main

(

)

Implementation of `parallel` command, relying on homemade `prefix` lib.

def main():
    """Implementation of `parallel` command, relying on homemade `prefix` lib."""
 
    cmds, args = prefix.preparse(
        sys.argv,
        tutorial=(
            "Runs the command in parallel on local machine. Only commands "
            "running over n-tuples may be prefixed, and must natively output "
            "a ROOT file. The prefix command will replace the single ROOT "
            "file with a directory containing a series of ROOT files, as well "
            "as the standard output stream in hidden text files. Note: be nice "
            "with your colleagues, and don't use all cores for a too long time "
            "unless you are sure that the machine is free."
        ),
        multi_opt=True,
    )
 
    pref_parallel = prefix.PrefixCommand(sys.argv[0], cmds)
 
    if cmds.help:
        prefix.tweak_helper_multi(pref_parallel)
 
    if cmds.git:
        prefix.git_hash(cmds.exec)
 
    if cmds.help or cmds.git:
        sys.exit()
 
    pref_parallel.parse(args)
    pref_parallel.prepare_io()
    pref_parallel.prepare_fire_and_forget()
 
    with open(pref_parallel.absoutput / "parallel", "w") as backup:
        shell_cmd = (
            ["parallel"]
            + [pref_parallel.cmds.exec]
            + pref_parallel.inputs
            + [pref_parallel.output]
            + pref_parallel.args
        )
        if pref_parallel.splittable:
            shell_cmd += ["-j", str(pref_parallel.nSplit)]
        print(" ".join(shell_cmd), file=backup)
 
    if cmds.dry_run:
        print(f"Run in parallel with `{pref_parallel.absoutput}/parallel`")
        sys.exit()
 
    with Pool(pref_parallel.nSplit) as pool:
        shell_cmds = prepare_cmds(pref_parallel)
        pool.starmap(
            run,
            [(shell_cmd, pref_parallel, k) for k, shell_cmd in enumerate(shell_cmds)],
        )
 
    if cmds.background:
        print("Check the status of your tasks with `top` (or `htop` if available).")
 
 

prepare_cmds()

def parallel.prepare_cmds

(

pref_parallel

)

Prepare the single commands to be run, including redirection of standard output.

def prepare_cmds(pref_parallel):
    """Prepare the single commands to be run, including redirection of standard output."""
 
    shell_cmds = []
    for k in range(pref_parallel.nSplit):
        output_file = f"{k}{pref_parallel.extension}"
        shell_cmd = (
            [pref_parallel.cmds.exec]
            + pref_parallel.inputs
            + [pref_parallel.output + "/" + output_file]
            + pref_parallel.args
        )
        if pref_parallel.splittable:
            shell_cmd += ["-j", str(pref_parallel.nSplit), "-k", str(k)]
        shell_cmd_str = " ".join(shell_cmd)
        print(shell_cmd_str)
 
        output_log = pref_parallel.output + f"/.{k}.out"
        if pref_parallel.cmds.background:
            shell_cmd_str += f" -p {pref_parallel.name} > {output_log} 2>&1 &"
        else:
            shell_cmd_str += f" -p {pref_parallel.name} 2>&1 | tee {output_log}"
        shell_cmds.append(shell_cmd_str)
 
    return shell_cmds
 
 

run()

None parallel.run	(		shell_cmd,
			pref,
			k
	)

Runs one single command.

def run(shell_cmd, pref, k) -> None:
    """Runs one single command."""
 
    output = pref.output
    nSplit = pref.nSplit
 
    proc = subprocess.Popen(shell_cmd, shell=True, text=True, stdout=subprocess.PIPE)
 
    formatting = ""
    for line in iter(proc.stdout.readline, ""):
        formatting = prefix.print_slice(formatting, line, k, nSplit, "")
 
    proc.wait()