<commit_before>import os
import sys
import time
import signal
import shutil
import logging
import tempfile
import subprocess
import multiprocessing
import psutil
from openquake.baselib import (
    zeromq as z, general, parallel, config, sap, InvalidFile)
try:
    from setproctitle import setproctitle
except ImportError:
    def setproctitle(title):
        "Do nothing"


class TimeoutError(RuntimeError):
    pass


def _streamer():
    # streamer for zmq workers running on the master node
    port = int(config.zworkers.ctrl_port)
    task_input_url = 'tcp://127.0.0.1:%d' % (port + 2)
    task_output_url = 'tcp://%s:%s' % (config.dbserver.listen, port + 1)
    try:
        z.zmq.proxy(z.bind(task_input_url, z.zmq.PULL),
                    z.bind(task_output_url, z.zmq.PUSH))
    except (KeyboardInterrupt, z.zmq.ContextTerminated):
        pass  # killed cleanly by SIGINT/SIGTERM


def check_status(**kw):
    """
    :returns: a non-empty error string if the streamer or worker pools are down
    """
    c = config.zworkers.copy()
    c.update(kw)
    hostport = config.dbserver.listen, int(c['ctrl_port']) + 1
    errors = []
    if not general.socket_ready(hostport):
        errors.append('The task streamer on %s:%s is down' % hostport)
    for host, status in WorkerMaster(**c).status():
        if status != 'running':
            errors.append('The workerpool on %s is down' % host)
    return '\n'.join(errors)


class WorkerMaster(object):
    """
    :param ctrl_port: port on which the worker pools listen
    :param host_cores: names of the remote hosts and number of cores to use
    :param remote_python: path of the Python executable on the remote hosts
    """
    def __init__(self, ctrl_port=config.zworkers.ctrl_port, host_cores=None,
                 remote_python=None, receiver_ports=None):
        # NB: receiver_ports is not used but needed for compliance
        self.ctrl_port = int(ctrl_port)
        self.host_cores = ([hc.split() for hc in host_cores.split(',')]
                           if host_cores else [])
        for host, cores in self.host_cores:
            if int(cores) < -1:
                raise InvalidFile('openquake.cfg: found %s %s' %
                                  (host, cores))
        self.remote_python = remote_python or sys.executable
        self.popens = []

    def wait(self, seconds=30):
        """
        Wait until all workerpools start
        """
        for _ in range(seconds):
            time.sleep(1)
            status = self.status()
            if all(st == 'running' for host, st in status):
                break
        else:
            raise TimeoutError(status)
        return status

    def status(self, host=None):
        """
        :returns: a list of pairs (hostname, 'running'|'not-running')
        """
        if host is None:
            host_cores = self.host_cores
        else:
            host_cores = [hc for hc in self.host_cores if hc[0] == host]
        lst = []
        for host, _ in host_cores:
            ready = general.socket_ready((host, self.ctrl_port))
            lst.append((host, 'running' if ready else 'not-running'))
        return lst

    def start(self):
        """
        Start multiple workerpools, possibly on remote servers via ssh,
        assuming there is an active streamer.
        """
        starting = []
        for host, cores in self.host_cores:
            if self.status(host)[0][1] == 'running':
                print('%s:%s already running' % (host, self.ctrl_port))
                continue
            ctrl_url = 'tcp://%s:%s' % (0.0.0.0, self.ctrl_port)
            if host == '127.0.0.1':  # localhost
                args = [sys.executable]
            else:
                args = ['ssh -f -T', host, self.remote_python]
            args += ['-m', 'openquake.baselib.workerpool', ctrl_url,
                     '-n', cores]
            if host != '127.0.0.1':
                logging.warning(
                    '%s: if it hangs, check the ssh keys', ' '.join(args))
            self.popens.append(subprocess.Popen(args))
            starting.append(host)
        return 'starting %s' % starting

    def stop(self):
        """
        Send a "stop" command to all worker pools
        """
        stopped = []
        for host, _ in self.host_cores:
            if self.status(host)[0][1] == 'not-running':
                print('%s not running' % host)
                continue
            ctrl_url = 'tcp://%s:%s' % (host, self.ctrl_port)
            with z.Socket(ctrl_url, z.zmq.REQ, 'connect') as sock:
                sock.send('stop')
                stopped.append(host)
        for popen in self.popens:
            popen.terminate()<commit_msg>use only one substituion %s for tcp address<commit_after>import os
import sys
import time
import signal
import shutil
import logging
import tempfile
import subprocess
import multiprocessing
import psutil
from openquake.baselib import (
    zeromq as z, general, parallel, config, sap, InvalidFile)
try:
    from setproctitle import setproctitle
except ImportError:
    def setproctitle(title):
        "Do nothing"


class TimeoutError(RuntimeError):
    pass


def _streamer():
    # streamer for zmq workers running on the master node
    port = int(config.zworkers.ctrl_port)
    task_input_url = 'tcp://127.0.0.1:%d' % (port + 2)
    task_output_url = 'tcp://%s:%s' % (config.dbserver.listen, port + 1)
    try:
        z.zmq.proxy(z.bind(task_input_url, z.zmq.PULL),
                    z.bind(task_output_url, z.zmq.PUSH))
    except (KeyboardInterrupt, z.zmq.ContextTerminated):
        pass  # killed cleanly by SIGINT/SIGTERM


def check_status(**kw):
    """
    :returns: a non-empty error string if the streamer or worker pools are down
    """
    c = config.zworkers.copy()
    c.update(kw)
    hostport = config.dbserver.listen, int(c['ctrl_port']) + 1
    errors = []
    if not general.socket_ready(hostport):
        errors.append('The task streamer on %s:%s is down' % hostport)
    for host, status in WorkerMaster(**c).status():
        if status != 'running':
            errors.append('The workerpool on %s is down' % host)
    return '\n'.join(errors)


class WorkerMaster(object):
    """
    :param ctrl_port: port on which the worker pools listen
    :param host_cores: names of the remote hosts and number of cores to use
    :param remote_python: path of the Python executable on the remote hosts
    """
    def __init__(self, ctrl_port=config.zworkers.ctrl_port, host_cores=None,
                 remote_python=None, receiver_ports=None):
        # NB: receiver_ports is not used but needed for compliance
        self.ctrl_port = int(ctrl_port)
        self.host_cores = ([hc.split() for hc in host_cores.split(',')]
                           if host_cores else [])
        for host, cores in self.host_cores:
            if int(cores) < -1:
                raise InvalidFile('openquake.cfg: found %s %s' %
                                  (host, cores))
        self.remote_python = remote_python or sys.executable
        self.popens = []

    def wait(self, seconds=30):
        """
        Wait until all workerpools start
        """
        for _ in range(seconds):
            time.sleep(1)
            status = self.status()
            if all(st == 'running' for host, st in status):
                break
        else:
            raise TimeoutError(status)
        return status

    def status(self, host=None):
        """
        :returns: a list of pairs (hostname, 'running'|'not-running')
        """
        if host is None:
            host_cores = self.host_cores
        else:
            host_cores = [hc for hc in self.host_cores if hc[0] == host]
        lst = []
        for host, _ in host_cores:
            ready = general.socket_ready((host, self.ctrl_port))
            lst.append((host, 'running' if ready else 'not-running'))
        return lst

    def start(self):
        """
        Start multiple workerpools, possibly on remote servers via ssh,
        assuming there is an active streamer.
        """
        starting = []
        for host, cores in self.host_cores:
            if self.status(host)[0][1] == 'running':
                print('%s:%s already running' % (host, self.ctrl_port))
                continue
            ctrl_url = 'tcp://0.0.0.0:%s' % (self.ctrl_port)
            if host == '127.0.0.1':  # localhost
                args = [sys.executable]
            else:
                args = ['ssh -f -T', host, self.remote_python]
            args += ['-m', 'openquake.baselib.workerpool', ctrl_url,
                     '-n', cores]
            if host != '127.0.0.1':
                logging.warning(
                    '%s: if it hangs, check the ssh keys', ' '.join(args))
            self.popens.append(subprocess.Popen(args))
            starting.append(host)
        return 'starting %s' % starting

    def stop(self):
        """
        Send a "stop" command to all worker pools
        """
        stopped = []
        for host, _ in self.host_cores:
            if self.status(host)[0][1] == 'not-running':
                print('%s not running' % host)
                continue
            ctrl_url = 'tcp://%s:%s' % (host, self.ctrl_port)
            with z.Socket(ctrl_url, z.zmq.REQ, 'connect') as sock:
                sock.send('stop')
                stopped.append(host)
        for popen in self.popens:
            popen.terminate()
{'fraction_non_alphanumeric': 0.11419818297063174, 'fraction_numerical': 0.007183604479188675, 'mean_word_length': 1.987377721678763, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 8, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '498200', 'n_tokens_mistral': 2920, 'n_tokens_neox': 2791, 'n_words': 797}