<commit_before>from __future__ import print_function
import os
import datetime
import subprocess
from future.utils import raise_with_traceback
from pathlib import Path
import numpy as np
import time
import progressbar
import shutil
from collections import defaultdict
import sys

from dps.parallel.base import ReadOnlyJob, zip_root
from dps.utils import cd, parse_timedelta, make_filename, make_symlink


DEFAULT_HOST_POOL = ['<EMAIL>'.format(i) for i in range(1, 32)]


class ParallelSession(object):
    """ Run a Job in parallel using gnu-parallel.

    A directory for this Job execution is created in `scratch`, and results are saved there.

    Parameters
    ----------
    name: str
        Name for the experiment.
    input_zip: str
        Path to a zip archive storing the Job.
    pattern: str
        Pattern to use to select which ops to run within the Job.
    scratch: str
        Path to location where the results of running the selected ops will be
        written. Must be writeable by the master process.
    local_scratch_prefix: str
        Path to scratch directory that is local to each remote host.
    ppn: int
        Number of processors per node.
    wall_time: str
        String specifying the maximum wall-time allotted to running the selected ops.
    cleanup_time: str
        String specifying the amount of cleanup time to allow per step.
    slack_time: float
        String specifying the amount of slack time to allow per step. Corresponds to
        time allotted to each process to respond to the signal that the step's time is up.
    add_date: bool
        Whether to add current date/time to the name of the directory where results are stored.
    dry_run: bool
        If True, control script will be generated but not executed/submitted.
    parallel_exe: str
        Path to the `gnu-parallel` executable to use.
    host_pool: list of str
        A list of names of hosts to use to execute the job.
    max_hosts: int
        Maximum number of hosts to use.
    env_vars: dict (str -> str)
        Dictionary mapping environment variable names to values. These will be accessible
        by the submit script, and will also be sent to the worker nodes.
    redirect: bool
        If True, stderr and stdout of jobs is saved in files rather than being printed to screen.
    n_retries: int
        Number of retries per job.
    gpu_set: str
        Comma-separated list of indices of gpus to use.
    step_time_limit: str
        String specifying time limit for each step. If not supplied, a time limit is inferred
        automatically based on wall_time and number of steps (giving each step an equal amount
        of time).
    ignore_gpu: bool
        If True, GPUs will be requested by as part of the job, but will not be used at run time.
    store_experiments: bool
        If True, after each step we fetch experiment data from each of the nodes and store it locally
        along with the results. Either way, experiment data is deleted from nodes after each step.
    ssh_options: string
        String of options to pass to ssh.

    """
    def __init__(
            self, name, input_zip, pattern, scratch, local_scratch_prefix='/tmp/dps/hyper/', ppn=12, cpp=1,
            pmem=None, wall_time="1hour", cleanup_time="1min", slack_time="1min", add_date=True, dry_run=0,
            parallel_exe="$HOME/.local/bin/parallel", kind="parallel", host_pool=None,
            min_hosts=1, max_hosts=1, env_vars=None, redirect=False, n_retries=0, gpu_set="",
            step_time_limit="", ignore_gpu=False, store_experiments=True, ssh_options=None, readme=""):

        if ssh_options is None:
            ssh_options = (
                "-oPasswordAuthentication=no "
                "-oStrictHostKeyChecking=no "
                "-oConnectTimeout=5 "
                "-oServerAliveInterval=2"
            )

        if kind == "pbs":
            local_scratch_prefix = "\\$RAMDISK"
        elif "slurm" in kind:
            local_scratch_prefix = "/tmp/dps/hyper"

        assert kind in "parallel pbs slurm slurm-local".split()
        hpc = kind != "parallel"
        clean_pattern = pattern.replace(' ', '_')

        # Create directory to run the job from - should be on scratch.
        scratch = os.path.abspath(os.path.expandvars(str(scratch)))
        job_directory = make_filename(
            '{}_{}'.format(name, clean_pattern),
            directory=scratch,
            add_date=add_date)
        os.makedirs(os.path.realpath(job_directory + "/experiments"))

        if readme:
            with open(os.path.join(job_directory, 'README.md'), 'w') as f:
                f.write(readme)
            del f

        input_zip_stem = Path(input_zip).stem
        input_zip = shutil.copy(str(input_zip), os.path.join(job_directory, "orig.zip"))
        input_zip = Path(input_zip)
        input_zip_abs = input_zip.resolve()
        input_zip_base = input_zip.name
        archive_root = zip_root(input_zip)
<commit_msg>Create git summary when running a Parallel session.<commit_after>from __future__ import print_function
import os
import datetime
import subprocess
from future.utils import raise_with_traceback
from pathlib import Path
import numpy as np
import time
import progressbar
import shutil
from collections import defaultdict
import sys

from dps.parallel.base import ReadOnlyJob, zip_root
from dps.utils import cd, parse_timedelta, make_filename, make_symlink, dps_git_summary


DEFAULT_HOST_POOL = ['<EMAIL>'.<EMAIL>(i) for i in range(1, 32)]


class ParallelSession(object):
    """ Run a Job in parallel using gnu-parallel.

    A directory for this Job execution is created in `scratch`, and results are saved there.

    Parameters
    ----------
    name: str
        Name for the experiment.
    input_zip: str
        Path to a zip archive storing the Job.
    pattern: str
        Pattern to use to select which ops to run within the Job.
    scratch: str
        Path to location where the results of running the selected ops will be
        written. Must be writeable by the master process.
    local_scratch_prefix: str
        Path to scratch directory that is local to each remote host.
    ppn: int
        Number of processors per node.
    wall_time: str
        String specifying the maximum wall-time allotted to running the selected ops.
    cleanup_time: str
        String specifying the amount of cleanup time to allow per step.
    slack_time: float
        String specifying the amount of slack time to allow per step. Corresponds to
        time allotted to each process to respond to the signal that the step's time is up.
    add_date: bool
        Whether to add current date/time to the name of the directory where results are stored.
    dry_run: bool
        If True, control script will be generated but not executed/submitted.
    parallel_exe: str
        Path to the `gnu-parallel` executable to use.
    host_pool: list of str
        A list of names of hosts to use to execute the job.
    max_hosts: int
        Maximum number of hosts to use.
    env_vars: dict (str -> str)
        Dictionary mapping environment variable names to values. These will be accessible
        by the submit script, and will also be sent to the worker nodes.
    redirect: bool
        If True, stderr and stdout of jobs is saved in files rather than being printed to screen.
    n_retries: int
        Number of retries per job.
    gpu_set: str
        Comma-separated list of indices of gpus to use.
    step_time_limit: str
        String specifying time limit for each step. If not supplied, a time limit is inferred
        automatically based on wall_time and number of steps (giving each step an equal amount
        of time).
    ignore_gpu: bool
        If True, GPUs will be requested by as part of the job, but will not be used at run time.
    store_experiments: bool
        If True, after each step we fetch experiment data from each of the nodes and store it locally
        along with the results. Either way, experiment data is deleted from nodes after each step.
    ssh_options: string
        String of options to pass to ssh.

    """
    def __init__(
            self, name, input_zip, pattern, scratch, local_scratch_prefix='/tmp/dps/hyper/', ppn=12, cpp=1,
            pmem=None, wall_time="1hour", cleanup_time="1min", slack_time="1min", add_date=True, dry_run=0,
            parallel_exe="$HOME/.local/bin/parallel", kind="parallel", host_pool=None,
            min_hosts=1, max_hosts=1, env_vars=None, redirect=False, n_retries=0, gpu_set="",
            step_time_limit="", ignore_gpu=False, store_experiments=True, ssh_options=None, readme=""):

        if ssh_options is None:
            ssh_options = (
                "-oPasswordAuthentication=no "
                "-oStrictHostKeyChecking=no "
                "-oConnectTimeout=5 "
                "-oServerAliveInterval=2"
            )

        if kind == "pbs":
            local_scratch_prefix = "\\$RAMDISK"
        elif "slurm" in kind:
            local_scratch_prefix = "/tmp/dps/hyper"

        assert kind in "parallel pbs slurm slurm-local".split()
        hpc = kind != "parallel"
        clean_pattern = pattern.replace(' ', '_')

        # Create directory to run the job from - should be on scratch.
        scratch = os.path.abspath(os.path.expandvars(str(scratch)))
        job_directory = make_filename(
            '{}_{}'.format(name, clean_pattern),
            directory=scratch,
            add_date=add_date)
        os.makedirs(os.path.realpath(job_directory + "/experiments"))

        if readme:
            with open(os.path.join(job_directory, 'README.md'), 'w') as f:
                f.write(readme)
            del f

        git_summary = dps_git_summary()
        with open(os.path.join(job_directory, 'git_summary.txt'), 'w') as f:
            f.write(git_summary.summarize(diff=True))
        del f

        input_zip_stem = Path(input_zip).stem
        input_zip = shutil.copy(str(input_zip), os.path.join(job_directory, "orig.zip"))
        input_zip = Path(input_zip)
        input_zip_abs = input_zip.resolve()
        input_zip_base = input_zip.name
        archive_root = zip_root(input_zip)

{'fraction_non_alphanumeric': 0.07142857142857142, 'fraction_numerical': 0.0029354207436399216, 'mean_word_length': 2.573776223776224, 'pattern_counts': {'":': 2, '<': 6, '<?xml version=': 0, '>': 8, '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}}
{'dir': 'git-commits-cleaned', 'id': '2102498', 'n_tokens_mistral': 2896, 'n_tokens_neox': 2804, 'n_words': 1101}