<commit_before># Mrs
# Copyright 2008-2011 Brigham Young University
#
# This file is part of Mrs.
#
# Mrs is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# Mrs is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# Mrs.  If not, see <http://www.gnu.org/licenses/>.
#
# Inquiries regarding any further use of Mrs, please contact the Copyright
# Licensing Office, Brigham Young University, 3760 HBLL, Provo, UT 84602,
# (801) 422-9339 or 422-3821, e-mail <EMAIL>.

"""Mrs Slave

The Mrs Slave runs in two processes: the main process and the worker process.
The main process has a main thread, a slave thread and an rpc thread.

The main thread doesn't really do anything.  It starts the other two threads
and waits for them to finish.  If the user hits CTRL-C, the main thread will
be interrupted, and it will shut down the event thread.  The only reason that
the main thread exists at all is to deal with signals.

The worker process executes the user's map function and reduce function.
That's it.  It just does what the main process tells it to.  The worker
process is terminated when the main process exits.
"""

# Number of ping timeouts before giving up:
PING_ATTEMPTS = 50

COOKIE_LEN = 8

import datetime
import multiprocessing
import optparse
import select
import socket
import tempfile
import threading
import urlparse
import worker

from . import bucket
from . import http
from . import registry
from . import util
from .version import VERSION

from logging import getLogger
logger = getLogger('mrs')


class Slave(object):
    """State of a Mrs slave

    Attributes:
        _outdirs: map from a (dataset_id, source) pair to an output directory
    """

    def __init__(self, program_class, master_url, local_shared, pingdelay,
            timeout):
        self.program_class = program_class
        self.master_url = master_url
        self.local_shared = local_shared
        self.pingdelay = pingdelay
        self.timeout = timeout

        self.request_pipe_slave, self.request_pipe_worker = (
                multiprocessing.Pipe())
        self.exit_pipe_recv, self.exit_pipe_send = multiprocessing.Pipe(False)

        self.id = None
        self.cookie = util.random_string(COOKIE_LEN)
        self.timestamp = None
        self.watchdog_stamp = None
        self.rpc_port = None
        self.bucket_port = None
        self.master_rpc = None
        self.url_converter = None

        self.setup_complete = False
        self.current_request_id = None
        self._outdirs = {}
        self._outdirs_lock = threading.Lock()

    def run(self):
        self.start_rpc_server_thread()
        self.master_rpc = http.ServerProxy(self.master_url, self.timeout)

        result = self.signin()
        if not result:
            return
        self.id, addr, jobdir, opts, args = result
        default_dir = self.init_default_dir(jobdir)

        if self.local_shared:
            self.start_bucket_server_thread(default_dir)
            self.url_converter = bucket.URLConverter(addr, self.bucket_port,
                    default_dir)

        # Tell the Worker to run the user_setup function and wait for
        # a response.
        if not self.worker_setup(opts, args, default_dir):
            return

        self.setup_complete = True

        # TODO: start a ping and/or watchdog thread

        self.report_ready()
        try:
            self.workloop()
        finally:
            util.remove_recursive(default_dir)

    def start_rpc_server_thread(self):
        rpc_interface = SlaveInterface(self)
        rpc_server = http.RPCServer(('', 0), rpc_interface)
        _, self.rpc_port = rpc_server.socket.getsockname()

        rpc_thread = threading.Thread(target=rpc_server.serve_forever,
                name='RPC Server')
        rpc_thread.daemon = True
        rpc_thread.start()

    def start_bucket_server_thread(self, default_dir):
        bucket_server = http.ThreadingBucketServer(('', 0), default_dir)
        _, self.bucket_port = bucket_server.socket.getsockname()

        bucket_thread = threading.Thread(target=bucket_server.serve_forever,
                name='Bucket Server')
        bucket_thread.daemon = True
        bucket_thread.start()

    def signin(self):
        """Start Slave RPC Server and sign in to master.

        Returns (slave_id, jobdir, opts, args): the slave id, job
        directory, options object, and args list given by the server.
        """
        cookie = self.cookie
        program_hash = registry.object_hash(self.program_class)

        try:
            slave_id, addr, jobdir, optdict, args = self.master_rpc.signin(
                    VERSION, cookie, self.rpc_port, program_hash)
        except socket.error, e:
            msg = e.args[1]
            logger.critical('Unable to contact master: %s' % msg)
            return None

        if slave_id < 0:
            logger.critical('Master rejected signin.')
            return None

        # Parse the opts given by the master.
        opts = optparse.Values(optdict)

        return slave_id, addr, jobdir, opts, args

    def init_default_dir(self, jobdir):
        if self.local_shared:
            logger.info("local_shared dir IS: %s" % self.local_shared)
            util.try_makedirs(self.local_shared)
            directory = self.local_shared
            prefix = 'mrs_slave_'
        else:
            hostname, _, _ = socket.gethostname().partition('.')
            directory = jobdir
            logger.info("else IS: %s" % jobdir) 
            prefix = hostname
        return tempfile.mkdtemp(dir=directory, prefix=prefix)

    def report_ready(self):
        """Report to the master that we are ready to accept tasks.

        This is the callback after user_setup is called.
        """
        assert self.current_request_id is None

        # TODO: make this try a few times if there's a timeout
        try:
            self.master_rpc.ready(self.id, self.cookie)
        except socket.error, e:
            msg = e.args[0]
            logger.critical('Failed to report due to network error: %s' % msg)

        logger.info('Reported ready to master.')
        self.update_timestamp()

    def worker_setup(self, opts, args, default_dir):
        request = worker.WorkerSetupRequest(opts, args, default_dir)
        self.request_pipe_slave.send(request)
        response = self.request_pipe_slave.recv()
        if isinstance(response, worker.WorkerSetupSuccess):<commit_msg>Removed to temporary print lines<commit_after># Mrs
# Copyright 2008-2011 Brigham Young University
#
# This file is part of Mrs.
#
# Mrs is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# Mrs is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# Mrs.  If not, see <http://www.gnu.org/licenses/>.
#
# Inquiries regarding any further use of Mrs, please contact the Copyright
# Licensing Office, Brigham Young University, 3760 HBLL, Provo, UT 84602,
# (801) 422-9339 or 422-3821, e-mail <EMAIL>.

"""Mrs Slave

The Mrs Slave runs in two processes: the main process and the worker process.
The main process has a main thread, a slave thread and an rpc thread.

The main thread doesn't really do anything.  It starts the other two threads
and waits for them to finish.  If the user hits CTRL-C, the main thread will
be interrupted, and it will shut down the event thread.  The only reason that
the main thread exists at all is to deal with signals.

The worker process executes the user's map function and reduce function.
That's it.  It just does what the main process tells it to.  The worker
process is terminated when the main process exits.
"""

# Number of ping timeouts before giving up:
PING_ATTEMPTS = 50

COOKIE_LEN = 8

import datetime
import multiprocessing
import optparse
import select
import socket
import tempfile
import threading
import urlparse
import worker

from . import bucket
from . import http
from . import registry
from . import util
from .version import VERSION

from logging import getLogger
logger = getLogger('mrs')


class Slave(object):
    """State of a Mrs slave

    Attributes:
        _outdirs: map from a (dataset_id, source) pair to an output directory
    """

    def __init__(self, program_class, master_url, local_shared, pingdelay,
            timeout):
        self.program_class = program_class
        self.master_url = master_url
        self.local_shared = local_shared
        self.pingdelay = pingdelay
        self.timeout = timeout

        self.request_pipe_slave, self.request_pipe_worker = (
                multiprocessing.Pipe())
        self.exit_pipe_recv, self.exit_pipe_send = multiprocessing.Pipe(False)

        self.id = None
        self.cookie = util.random_string(COOKIE_LEN)
        self.timestamp = None
        self.watchdog_stamp = None
        self.rpc_port = None
        self.bucket_port = None
        self.master_rpc = None
        self.url_converter = None

        self.setup_complete = False
        self.current_request_id = None
        self._outdirs = {}
        self._outdirs_lock = threading.Lock()

    def run(self):
        self.start_rpc_server_thread()
        self.master_rpc = http.ServerProxy(self.master_url, self.timeout)

        result = self.signin()
        if not result:
            return
        self.id, addr, jobdir, opts, args = result
        default_dir = self.init_default_dir(jobdir)

        if self.local_shared:
            self.start_bucket_server_thread(default_dir)
            self.url_converter = bucket.URLConverter(addr, self.bucket_port,
                    default_dir)

        # Tell the Worker to run the user_setup function and wait for
        # a response.
        if not self.worker_setup(opts, args, default_dir):
            return

        self.setup_complete = True

        # TODO: start a ping and/or watchdog thread

        self.report_ready()
        try:
            self.workloop()
        finally:
            util.remove_recursive(default_dir)

    def start_rpc_server_thread(self):
        rpc_interface = SlaveInterface(self)
        rpc_server = http.RPCServer(('', 0), rpc_interface)
        _, self.rpc_port = rpc_server.socket.getsockname()

        rpc_thread = threading.Thread(target=rpc_server.serve_forever,
                name='RPC Server')
        rpc_thread.daemon = True
        rpc_thread.start()

    def start_bucket_server_thread(self, default_dir):
        bucket_server = http.ThreadingBucketServer(('', 0), default_dir)
        _, self.bucket_port = bucket_server.socket.getsockname()

        bucket_thread = threading.Thread(target=bucket_server.serve_forever,
                name='Bucket Server')
        bucket_thread.daemon = True
        bucket_thread.start()

    def signin(self):
        """Start Slave RPC Server and sign in to master.

        Returns (slave_id, jobdir, opts, args): the slave id, job
        directory, options object, and args list given by the server.
        """
        cookie = self.cookie
        program_hash = registry.object_hash(self.program_class)

        try:
            slave_id, addr, jobdir, optdict, args = self.master_rpc.signin(
                    VERSION, cookie, self.rpc_port, program_hash)
        except socket.error, e:
            msg = e.args[1]
            logger.critical('Unable to contact master: %s' % msg)
            return None

        if slave_id < 0:
            logger.critical('Master rejected signin.')
            return None

        # Parse the opts given by the master.
        opts = optparse.Values(optdict)

        return slave_id, addr, jobdir, opts, args

    def init_default_dir(self, jobdir):
        if self.local_shared:
            util.try_makedirs(self.local_shared)
            directory = self.local_shared
            prefix = 'mrs_slave_'
        else:
            hostname, _, _ = socket.gethostname().partition('.')
            directory = jobdir
            prefix = hostname
        return tempfile.mkdtemp(dir=directory, prefix=prefix)

    def report_ready(self):
        """Report to the master that we are ready to accept tasks.

        This is the callback after user_setup is called.
        """
        assert self.current_request_id is None

        # TODO: make this try a few times if there's a timeout
        try:
            self.master_rpc.ready(self.id, self.cookie)
        except socket.error, e:
            msg = e.args[0]
            logger.critical('Failed to report due to network error: %s' % msg)

        logger.info('Reported ready to master.')
        self.update_timestamp()

    def worker_setup(self, opts, args, default_dir):
        request = worker.WorkerSetupRequest(opts, args, default_dir)
        self.request_pipe_slave.send(request)
        response = self.request_pipe_slave.recv()
        if isinstance(response, worker.WorkerSetupSuccess):
{'fraction_non_alphanumeric': 0.07527983027288024, 'fraction_numerical': 0.006291608749725656, 'mean_word_length': 2.5598958333333335, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4756372', 'n_tokens_mistral': 3933, 'n_tokens_neox': 3747, 'n_words': 1282}