<commit_before>#!/usr/bin/env python

# THANKS TO http://ilab.cs.byu.edu/python/socket/echoserver.html

import select
import socket
import sys
import threading
from threading import Lock
import pickle
from collections import deque
import logging
from time import sleep
from morpheus.JobEntry import SubJobEntry


# THANKS TO http://www.shutupandship.com/2012/02/how-python-logging-module-works.html

logger = logging.getLogger('MORPHEUS')
logger.setLevel(logging.DEBUG)
frmt = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh = logging.FileHandler('master.log')
fh.setFormatter(frmt)
fh.setLevel(logging.DEBUG)
logger.addHandler(fh)

class Master(object):
    ScheduledSubJobs = dict()
    scheduledLock = Lock()

    UnScheduledSubJobs = deque()

    results = dict()
    resultLock = Lock()

    subJobCounter = dict()
    counterLock = Lock()

    class Rescheduler(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)

        def run(self):
            print 'Rescheduler!!!'
            while 1:
                sleep(15)
                sLock = Master.scheduledLock
                sJobs = Master.ScheduledSubJobs
                sLock.acquire()
                for key in sJobs.keys():
                    Master.UnScheduledSubJobs.append(sJobs.pop(key))
                    logger.debug("putting back key--->>" + key)
                print 'Moved elements back'
                print len(Master.ScheduledSubJobs)
                sLock.release()


    class ClientHandler(threading.Thread):
        def __init__(self, (client, address)):
            threading.Thread.__init__(self)
            self.client = client
            self.address = address
            self.size = 1024

        def partitionJob(self, je):
            dataParts  = je._data.split('\n')
            for i, dataPart in enumerate(dataParts) :
                print i, ": ", dataPart
                Master.UnScheduledSubJobs.append(SubJobEntry(je._func, dataPart, je._jobID, i))
            return len(dataParts)

        def run(self):
            print 'Slave started work'
            data = self.client.recv(self.size)
            if data:
                message = data[0]

                if(message == "A"): # AVAILABILITY
                    logger.debug("Request: A.")
                    try:
                        sbj = Master.UnScheduledSubJobs.pop()
                        key = "{jId}.{sjid}".format(jId=sbj._jobID,sjid=sbj._subJobID)
                        Master.scheduledLock.acquire()
                        Master.ScheduledSubJobs[key] = sbj
                        Master.scheduledLock.release()
                        logger.debug("Job Scheduled: Job ID: " + str(sbj._jobID) + "subJob ID: " + str(sbj._subJobID))
                        self.client.send(pickle.dumps(sbj))
                    except IndexError:
                        self.client.send("No job to do.")
                    self.client.close()

                elif(message == "S"): # JOB SUBMISSION
                    filename = data[1:]
                    logger.debug("Request: S, " + filename)
                    with open(filename, "r") as f:
                        je = pickle.load(f)
                    self.client.close()
                    #record number of partitions
                    count = self.partitionJob(je)
                    Master.counterLock.acquire()
                    Master.subJobCounter[je._jobID] = count
                    Master.counterLock.release()

                    logger.debug("SJBs: " + str(Master.UnScheduledSubJobs))

                elif(message == "D"): # SUB-JOB COMPLETION
                    print 'D message'
                    jobID, subJobID, result = data[1:].split(',')
                    logger.debug("Request: D, for job ID: " + jobID + ", " + subJobID + " done.")
                    key = "{jId}.{sjid}".format(jId=jobID,sjid=subJobID)
                    Master.scheduledLock.acquire()
                    subJob = Master.ScheduledSubJobs.pop(key, None)
                    Master.scheduledLock.release()
                    message = 'F'
                    if subJob:
                        print 'sub job exists'
                        message = 'S'
                        Master.resultLock.acquire()
                        Master.results[jobID] = Master.results.get(jobID, 0) + int(result)
                        Master.resultLock.release()
                        Master.counterLock.acquire()
                        print Master.subJobCounter
                        int_jobID = int(jobID)
                        Master.subJobCounter[int_jobID] -= 1
                        if Master.subJobCounter[int_jobID] == 0:
                            message = 'Z' + (jobID+','+str(Master.results.get(jobID)))
                        Master.counterLock.release()

                    # subjob done
                    self.client.send(message)
                    self.client.close()

                else:
                    logger.debug("Invalid request.")
                    self.client.close()

            else:
                self.client.close()

    def __init__(self):
        self.host = 'localhost'
        self.port = 5000
        self.backlog = 5
        self.size = 1024
        self.server = None
        self.threads = []
        r = Master.Rescheduler()
        r.start()


    def open_socket(self):
        try:
            # use TCP and IPv4
            self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server.bind((self.host, self.port))
            self.server.listen(5)
        except socket.error, (value, message):
            if self.server:
                self.server.close()
            print "Could not open socket: " + message
            logger.debug("Could not open socket: " + message)
            sys.exit(1)

    def run(self):<commit_msg>changed log<commit_after>#!/usr/bin/env python

# THANKS TO http://ilab.cs.byu.edu/python/socket/echoserver.html

import select
import socket
import sys
import threading
from threading import Lock
import pickle
from collections import deque
import logging
from time import sleep
from morpheus.JobEntry import SubJobEntry


# THANKS TO http://www.shutupandship.com/2012/02/how-python-logging-module-works.html

logger = logging.getLogger('MORPHEUS')
logger.setLevel(logging.DEBUG)
frmt = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh = logging.FileHandler('master.log')
fh.setFormatter(frmt)
fh.setLevel(logging.DEBUG)
logger.addHandler(fh)

class Master(object):
    ScheduledSubJobs = dict()
    scheduledLock = Lock()

    UnScheduledSubJobs = deque()

    results = dict()
    resultLock = Lock()

    subJobCounter = dict()
    counterLock = Lock()

    class Rescheduler(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)

        def run(self):
            print 'Rescheduler!!!'
            while 1:
                sleep(15)
                sLock = Master.scheduledLock
                sJobs = Master.ScheduledSubJobs
                sLock.acquire()
                for key in sJobs.keys():
                    Master.UnScheduledSubJobs.append(sJobs.pop(key))
                    logger.debug("putting back key--->>" + key)
                print 'Moved elements back'
                print len(Master.ScheduledSubJobs)
                sLock.release()


    class ClientHandler(threading.Thread):
        def __init__(self, (client, address)):
            threading.Thread.__init__(self)
            self.client = client
            self.address = address
            self.size = 1024

        def partitionJob(self, je):
            dataParts  = je._data.split('\n')
            for i, dataPart in enumerate(dataParts) :
                print i, ": ", dataPart
                Master.UnScheduledSubJobs.append(SubJobEntry(je._func, dataPart, je._jobID, i))
            return len(dataParts)

        def run(self):
            print 'Slave started work'
            data = self.client.recv(self.size)
            if data:
                message = data[0]

                if(message == "A"): # AVAILABILITY
                    logger.debug("Request: A.")
                    try:
                        sbj = Master.UnScheduledSubJobs.pop()
                        key = "{jId}.{sjid}".format(jId=sbj._jobID,sjid=sbj._subJobID)
                        Master.scheduledLock.acquire()
                        Master.ScheduledSubJobs[key] = sbj
                        Master.scheduledLock.release()
                        logger.debug("Job Scheduled: Job ID: " + str(sbj._jobID) + "subJob ID: " + str(sbj._subJobID))
                        self.client.send(pickle.dumps(sbj))
                    except IndexError:
                        self.client.send("No job to do.")
                    self.client.close()

                elif(message == "S"): # JOB SUBMISSION
                    filename = data[1:]
                    logger.debug("Request: S, " + filename)
                    with open(filename, "r") as f:
                        je = pickle.load(f)
                    self.client.close()
                    #record number of partitions
                    count = self.partitionJob(je)
                    Master.counterLock.acquire()
                    Master.subJobCounter[je._jobID] = count
                    Master.counterLock.release()

                    logger.debug("SJBs: " + str(Master.UnScheduledSubJobs))

                elif(message == "D"): # SUB-JOB COMPLETION

                    jobID, subJobID, result = data[1:].split(',')
                    logger.debug("Request: D, for job ID: " + jobID + "-" + subJobID + " done.")
                    key = "{jId}.{sjid}".format(jId=jobID,sjid=subJobID)
                    Master.scheduledLock.acquire()
                    subJob = Master.ScheduledSubJobs.pop(key, None)
                    Master.scheduledLock.release()
                    message = 'F'
                    if subJob:
                        logger.debug('Job Exists: ' + jobID +"-"subJobID)
                        message = 'S'
                        Master.resultLock.acquire()
                        Master.results[jobID] = Master.results.get(jobID, 0) + int(result)
                        Master.resultLock.release()
                        Master.counterLock.acquire()
                        int_jobID = int(jobID)
                        Master.subJobCounter[int_jobID] -= 1

                        logger.debug(str(Master.subJobCounter))
                        if Master.subJobCounter[int_jobID] == 0:
                            message = 'Z' + (jobID+','+str(Master.results.get(jobID)))
                        Master.counterLock.release()

                    # subjob done
                    self.client.send(message)
                    self.client.close()

                else:
                    logger.debug("Invalid request.")
                    self.client.close()

            else:
                self.client.close()

    def __init__(self):
        self.host = 'localhost'
        self.port = 5000
        self.backlog = 20
        self.size = 1024
        self.server = None
        self.threads = []
        r = Master.Rescheduler()
        r.start()


    def open_socket(self):
        try:
            # use TCP and IPv4
            self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server.bind((self.host, self.port))
            self.server.listen(5)
        except socket.error, (value, message):
            if self.server:
                self.server.close()
            print "Could not open socket: " + message
            logger.debug("Could not open socket: " + message)
            sys.exit(1)

    def run(self):
{'fraction_non_alphanumeric': 0.10704035117339186, 'fraction_numerical': 0.005318250886375148, 'mean_word_length': 1.466583385384135, 'pattern_counts': {'":': 2, '<': 3, '<?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': '1809559', 'n_tokens_mistral': 3381, 'n_tokens_neox': 3149, 'n_words': 630}