<commit_before>#!/usr/bin/python
#
# Copyright (c) 2005-2006 rPath, Inc.
#
# All rights reserved
#


import os, sys
import logging
import math
import simplejson
import tempfile
import threading
import time
import traceback
import signal
import weakref

from jobmaster import master_error
from jobmaster import imagecache
from jobmaster import xencfg

from mcp import queue
from mcp import response
from mcp import client

from conary.lib import cfgtypes, util, log
from conary import conarycfg
from conary import conaryclient
from conary.conaryclient import cmdline
from conary.deps import deps
from conary import versions

def getAvailableArchs(arch):
    if arch in ('i686', 'i586', 'i486', 'i386'):
        return ('x86',)
    elif arch == 'x86_64':
        return ('x86', 'x86_64')

def getIP():
    p = os.popen("""ifconfig `route | grep "^default" | sed "s/.* //"` | grep "inet addr" | awk -F: '{print $2}' | sed 's/ .*//'""")
    data = p.read().strip()
    p.close()
    return data

def controlMethod(func):
    func._controlMethod = True
    return func

PROTOCOL_VERSIONS = set([1])

filterArgs = lambda d, *args: dict([x for x in d.iteritems() \
                                        if x[0] not in args])

getJsversion = lambda troveSpec: str(versions.VersionFromString( \
        cmdline.parseTroveSpec(troveSpec)[1]).trailingRevision())

def protocols(protocolList):
    if type(protocolList) in (int, long):
        protocolList = (protocolList,)
    def deco(func):
        def wrapper(self, *args, **kwargs):
            if kwargs.get('protocolVersion') in protocolList:
                return func(self, *args,
                            **(filterArgs(kwargs, 'protocolVersion')))
            else:
                raise master_error.ProtocolError(\
                    'Unsupported ProtocolVersion: %s' % \
                        str(kwargs.get('protocolVersion')))
        return wrapper
    return deco

def catchErrors(func):
    def wrapper(self, *args, **kwargs):
        try:
            func(self, *args, **kwargs)
        except:
            exc, e, bt = sys.exc_info()
            try:
                log.error(traceback.format_stack(bt))
                log.error(e)
            except:
                print >> sys.stderr, "couldn't log error", e
    return wrapper

class MasterConfig(client.MCPClientConfig):
    basePath = os.path.join(os.path.sep, 'srv', 'rbuilder', 'jobmaster')
    slaveLimit = (cfgtypes.CfgInt, 1)
    nodeName = (cfgtypes.CfgString, None)
    slaveMemory = (cfgtypes.CfgInt, 512) # memory in MB

class SlaveHandler(threading.Thread):
    # A slave handler is tied to a specific slave instance. do not re-use.
    def __init__(self, master, troveSpec):
        self.master = weakref.ref(master)
        self.imageCache = weakref.ref(master.imageCache)
        self.cfgPath = ''
        self.slaveName = None
        self.troveSpec = troveSpec
        self.lock = threading.RLock()
        threading.Thread.__init__(self)
        self.pid = None

    def slaveStatus(self, status):
        self.master().slaveStatus(self.slaveName, status,
                                  getJsversion(self.troveSpec))

    def start(self):
        fd, self.imagePath = tempfile.mkstemp()
        os.close(fd)
        xenCfg = xencfg.XenCfg(self.imagePath,
                               {'memory' : self.master().cfg.slaveMemory})
        self.slaveName = xenCfg.cfg['name']
        self.slaveStatus('building')
        fd, self.cfgPath = tempfile.mkstemp()
        os.close(fd)
        f = open(self.cfgPath, 'w')
        xenCfg.write(f)
        f.close()
        threading.Thread.start(self)
        log.info('starting slave: %s' % self.slaveName)
        return xenCfg.cfg['name']

    def stop(self):
        log.info('stopping slave %s' % self.slaveName)
        pid = self.pid
        if pid:
            try:
                os.kill(pid, signal.SIGTERM)
            except OSError, e:
                # ignore race condition where child died right after we recorded
                # it's pid
                if errno != 3:
                    raise
        os.system('xm destroy %s' % self.slaveName)
        if os.path.exists(self.imagePath):
            util.rmtree(self.imagePath, ignore_errors = True)
        self.slaveStatus('stopped')
        self.join()

    def getJobQueueName(self):
        name, verStr, flv = cmdline.parseTroveSpec(self.troveSpec)
        ver = versions.VersionFromString(verStr)
        jsVersion = str(ver.trailingRevision())

        arch = 'unknown'
        for refFlv, refArch in (('1#x86_64', 'x86_64'), ('1#x86', 'x86')):
            if flv.satisfies(deps.ThawFlavor(refFlv)):
                arch = refArch
                break
        return 'job%s:%s' % (jsVersion, arch)

    def run(self):
        self.pid = os.fork()
        if not self.pid:
            os.setsid()
            try:
                # don't use original. make a backup
                log.info('getting slave image')
                cachedImage = self.imageCache().getImage(self.troveSpec)
                util.copyfile(cachedImage, self.imagePath)
                # now add per-instance settings. such as path to MCP
                mntPoint = tempfile.mkdtemp()
                f = None
                try:
                    log.info('inserting runtime settings into slave')
                    os.system('mount -o loop %s %s' % (self.imagePath, mntPoint))
                    cfg = self.master().cfg

                    # write python SlaveConfig
                    cfgPath = os.path.join(mntPoint, 'srv', 'jobslave', 'config.d',
                                          'runtime')
                    util.mkdirChain(os.path.split(cfgPath)[0])
                    f = open(cfgPath, 'w')

                    # It never makes sense to direct a remote machine to
                    # 127.0.0.1
                    f.write('queueHost %s\n' % ((cfg.queueHost != '127.0.0.1') \
                                                    and cfg.queueHost \
                                                    or getIP()))<commit_msg>use shutil for image copy<commit_after>#!/usr/bin/python
#
# Copyright (c) 2005-2006 rPath, Inc.
#
# All rights reserved
#


import os, sys
import logging
import math
import simplejson
import tempfile
import threading
import time
import traceback
import shutil
import signal
import weakref

from jobmaster import master_error
from jobmaster import imagecache
from jobmaster import xencfg

from mcp import queue
from mcp import response
from mcp import client

from conary.lib import cfgtypes, util, log
from conary import conarycfg
from conary import conaryclient
from conary.conaryclient import cmdline
from conary.deps import deps
from conary import versions

def getAvailableArchs(arch):
    if arch in ('i686', 'i586', 'i486', 'i386'):
        return ('x86',)
    elif arch == 'x86_64':
        return ('x86', 'x86_64')

def getIP():
    p = os.popen("""ifconfig `route | grep "^default" | sed "s/.* //"` | grep "inet addr" | awk -F: '{print $2}' | sed 's/ .*//'""")
    data = p.read().strip()
    p.close()
    return data

def controlMethod(func):
    func._controlMethod = True
    return func

PROTOCOL_VERSIONS = set([1])

filterArgs = lambda d, *args: dict([x for x in d.iteritems() \
                                        if x[0] not in args])

getJsversion = lambda troveSpec: str(versions.VersionFromString( \
        cmdline.parseTroveSpec(troveSpec)[1]).trailingRevision())

def protocols(protocolList):
    if type(protocolList) in (int, long):
        protocolList = (protocolList,)
    def deco(func):
        def wrapper(self, *args, **kwargs):
            if kwargs.get('protocolVersion') in protocolList:
                return func(self, *args,
                            **(filterArgs(kwargs, 'protocolVersion')))
            else:
                raise master_error.ProtocolError(\
                    'Unsupported ProtocolVersion: %s' % \
                        str(kwargs.get('protocolVersion')))
        return wrapper
    return deco

def catchErrors(func):
    def wrapper(self, *args, **kwargs):
        try:
            func(self, *args, **kwargs)
        except:
            exc, e, bt = sys.exc_info()
            try:
                log.error(traceback.format_stack(bt))
                log.error(e)
            except:
                print >> sys.stderr, "couldn't log error", e
    return wrapper

class MasterConfig(client.MCPClientConfig):
    basePath = os.path.join(os.path.sep, 'srv', 'rbuilder', 'jobmaster')
    slaveLimit = (cfgtypes.CfgInt, 1)
    nodeName = (cfgtypes.CfgString, None)
    slaveMemory = (cfgtypes.CfgInt, 512) # memory in MB

class SlaveHandler(threading.Thread):
    # A slave handler is tied to a specific slave instance. do not re-use.
    def __init__(self, master, troveSpec):
        self.master = weakref.ref(master)
        self.imageCache = weakref.ref(master.imageCache)
        self.cfgPath = ''
        self.slaveName = None
        self.troveSpec = troveSpec
        self.lock = threading.RLock()
        threading.Thread.__init__(self)
        self.pid = None

    def slaveStatus(self, status):
        self.master().slaveStatus(self.slaveName, status,
                                  getJsversion(self.troveSpec))

    def start(self):
        fd, self.imagePath = tempfile.mkstemp()
        os.close(fd)
        xenCfg = xencfg.XenCfg(self.imagePath,
                               {'memory' : self.master().cfg.slaveMemory})
        self.slaveName = xenCfg.cfg['name']
        self.slaveStatus('building')
        fd, self.cfgPath = tempfile.mkstemp()
        os.close(fd)
        f = open(self.cfgPath, 'w')
        xenCfg.write(f)
        f.close()
        threading.Thread.start(self)
        log.info('starting slave: %s' % self.slaveName)
        return xenCfg.cfg['name']

    def stop(self):
        log.info('stopping slave %s' % self.slaveName)
        pid = self.pid
        if pid:
            try:
                os.kill(pid, signal.SIGTERM)
            except OSError, e:
                # ignore race condition where child died right after we recorded
                # it's pid
                if errno != 3:
                    raise
        os.system('xm destroy %s' % self.slaveName)
        if os.path.exists(self.imagePath):
            util.rmtree(self.imagePath, ignore_errors = True)
        self.slaveStatus('stopped')
        self.join()

    def getJobQueueName(self):
        name, verStr, flv = cmdline.parseTroveSpec(self.troveSpec)
        ver = versions.VersionFromString(verStr)
        jsVersion = str(ver.trailingRevision())

        arch = 'unknown'
        for refFlv, refArch in (('1#x86_64', 'x86_64'), ('1#x86', 'x86')):
            if flv.satisfies(deps.ThawFlavor(refFlv)):
                arch = refArch
                break
        return 'job%s:%s' % (jsVersion, arch)

    def run(self):
        self.pid = os.fork()
        if not self.pid:
            os.setsid()
            try:
                # don't use original. make a backup
                log.info('getting slave image')
                cachedImage = self.imageCache().getImage(self.troveSpec)
                shutil.copyfile(cachedImage, self.imagePath)
                # now add per-instance settings. such as path to MCP
                mntPoint = tempfile.mkdtemp()
                f = None
                try:
                    log.info('inserting runtime settings into slave')
                    os.system('mount -o loop %s %s' % (self.imagePath, mntPoint))
                    cfg = self.master().cfg

                    # write python SlaveConfig
                    cfgPath = os.path.join(mntPoint, 'srv', 'jobslave', 'config.d',
                                          'runtime')
                    util.mkdirChain(os.path.split(cfgPath)[0])
                    f = open(cfgPath, 'w')

                    # It never makes sense to direct a remote machine to
                    # 127.0.0.1
                    f.write('queueHost %s\n' % ((cfg.queueHost != '127.0.0.1') \
                                                    and cfg.queueHost \
                                                    or getIP()))
{'fraction_non_alphanumeric': 0.11194762684124386, 'fraction_numerical': 0.011129296235679214, 'mean_word_length': 1.846727230375029, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1635306', 'n_tokens_mistral': 3689, 'n_tokens_neox': 3474, 'n_words': 905}