<commit_before>#! /usr/bin/env python
# coding=utf-8

import os
import os.path
import sys
import time
import json
import shutil
import socket
import logging as logger
from multiprocessing import Process, Queue
import couchdb
import subprocess
import pickle
import io
import marshal, types
import importlib
#from contextlib import redirect_stdout



class Task:
    def __init__(self, name, opts, resdir):
        self.__name = name
        self.__opts = opts
        self.__resdir = resdir
        self.__proc = None
        self.__is_finished = False
        self.__results = None

        task_mod = __import__(name)

        print opts 
        installed_version = int(task_mod.Task.version)
        actual_version = int(self.__opts['version'])

        if installed_version < actual_version: # update installed task
            raise Exception("Version {0} for task {1} is too old, task must be updated".format("", name))

        f = io.StringIO()
        #with redirect_stdout(f):
        self.__cls = task_mod.Task()
        #print('Got stdout: "{0}"'.format(f.getvalue()))


    def collect_argrefs(self, tasks):
        for k, v in self.__refs.items(): # TODO maybe better store task config and use conf['refs']
            ref_task_name, ref_task_retarg = v.split('.')
            ref_task = tasks[ref_task_name]
            self.__args[k] = ref_task.get_result()[ref_task_retarg] # TODO raise missing key


    def run(self):
        self.__q = Queue()
        self.__proc = Process(target=self.__cls, args=(self.__opts['args'], self.__resdir, self.__q))
        self.__proc.start()


    def is_alive(self):
        return self.__proc is not None and self.__proc.is_alive()


    def is_finished(self):
        if self.__is_finished:
            return True
        if self.__proc is not None and not self.__proc.is_alive():
            self.__proc.join()
            self.__is_finished = True
            self.__results = self.__q.get()
        return self.__is_finished


    def get_result(self):
        return self.__results


    def get_name(self):
        return self.__name


    def get_log(self):
        return "xxx"


class Req:
    def __init__(self, idx, tasks):
        self.__idx = idx
        self.__tasks = []
        self.__name2task = {}
        self.__proc_task = None

        for task_name, task_opts in tasks.items():
            logger.debug('Adding new task {0} with args {1}'.format(task_name, task_opts['args']))
            
            resdir = os.path.join(script_path, 'results', self.__idx, task_name)
            if not os.path.isdir(resdir):
                os.makedirs(resdir)

            T = Task(task_name, task_opts, resdir)
            self.__tasks.append(T)
            self.__name2task[task_name] = T


    def probe(self, doc, db):
        # prepare correct currently running task
        if self.__proc_task is None:
            logger.debug('probing with {0} {1}'.format(self.__tasks, self.__proc_task))
            if len(self.__tasks) > 0:
                self.__proc_task = self.__tasks[0]
                self.__proc_task.run()
                del self.__tasks[0]
        else:
            logger.debug('running task {0} is alive'.format(self.__proc_task))
            if self.__proc_task.is_finished():
                doc['tasks'][self.__proc_task.get_name()]['result'] = self.__proc_task.get_result()
            task_log_buf = self.__proc_task.get_log()
            db.save(doc)
            db.put_attachment(doc, task_log_buf, 'log')
            if self.__proc_task.is_finished():
                self.__proc_task = None


    def is_finished(self):
        return self.__proc_task is None and len(self.__tasks) == 0


    def get_idx(self):
        return self.__idx


    def kill(self):
        pass # TODO


def __idx_lock(doc, db):
    doc['status'] = 'Processed'
    doc['host'] = socket.gethostname()
    db.save(doc)


def __idx_unlock(doc, db):
    doc['status'] = 'Done'
    db.save(doc)


def __idx_locked(doc):
    return doc['status'] != 'Waiting'


def update_tasks(couch):
    if 'tasks' not in couch:
        couch.create('tasks')
    db = couch['tasks']

    if 'config' not in db:
        db['config'] = {'names': [], 'opts': {}, 'locked': False}

    doc = db['config']

    while doc['locked']:
        time.sleep(0.2)

    doc['locked'] = True
    db[doc.id] = doc

    doc = db['config']
    conf_names = doc['names']
    conf_opts = doc['opts']
<commit_msg>Reload doc when it's locked.<commit_after>#! /usr/bin/env python
# coding=utf-8

import os
import os.path
import sys
import time
import json
import shutil
import socket
import logging as logger
from multiprocessing import Process, Queue
import couchdb
import subprocess
import pickle
import io
import marshal, types
import importlib
#from contextlib import redirect_stdout



class Task:
    def __init__(self, name, opts, resdir):
        self.__name = name
        self.__opts = opts
        self.__resdir = resdir
        self.__proc = None
        self.__is_finished = False
        self.__results = None

        task_mod = __import__(name)

        print opts 
        installed_version = int(task_mod.Task.version)
        actual_version = int(self.__opts['version'])

        if installed_version < actual_version: # update installed task
            raise Exception("Version {0} for task {1} is too old, task must be updated".format("", name))

        f = io.StringIO()
        #with redirect_stdout(f):
        self.__cls = task_mod.Task()
        #print('Got stdout: "{0}"'.format(f.getvalue()))


    def collect_argrefs(self, tasks):
        for k, v in self.__refs.items(): # TODO maybe better store task config and use conf['refs']
            ref_task_name, ref_task_retarg = v.split('.')
            ref_task = tasks[ref_task_name]
            self.__args[k] = ref_task.get_result()[ref_task_retarg] # TODO raise missing key


    def run(self):
        self.__q = Queue()
        self.__proc = Process(target=self.__cls, args=(self.__opts['args'], self.__resdir, self.__q))
        self.__proc.start()


    def is_alive(self):
        return self.__proc is not None and self.__proc.is_alive()


    def is_finished(self):
        if self.__is_finished:
            return True
        if self.__proc is not None and not self.__proc.is_alive():
            self.__proc.join()
            self.__is_finished = True
            self.__results = self.__q.get()
        return self.__is_finished


    def get_result(self):
        return self.__results


    def get_name(self):
        return self.__name


    def get_log(self):
        return "xxx"


class Req:
    def __init__(self, idx, tasks):
        self.__idx = idx
        self.__tasks = []
        self.__name2task = {}
        self.__proc_task = None

        for task_name, task_opts in tasks.items():
            logger.debug('Adding new task {0} with args {1}'.format(task_name, task_opts['args']))
            
            resdir = os.path.join(script_path, 'results', self.__idx, task_name)
            if not os.path.isdir(resdir):
                os.makedirs(resdir)

            T = Task(task_name, task_opts, resdir)
            self.__tasks.append(T)
            self.__name2task[task_name] = T


    def probe(self, doc, db):
        # prepare correct currently running task
        if self.__proc_task is None:
            logger.debug('probing with {0} {1}'.format(self.__tasks, self.__proc_task))
            if len(self.__tasks) > 0:
                self.__proc_task = self.__tasks[0]
                self.__proc_task.run()
                del self.__tasks[0]
        else:
            logger.debug('running task {0} is alive'.format(self.__proc_task))
            if self.__proc_task.is_finished():
                doc['tasks'][self.__proc_task.get_name()]['result'] = self.__proc_task.get_result()
            task_log_buf = self.__proc_task.get_log()
            db.save(doc)
            db.put_attachment(doc, task_log_buf, 'log')
            if self.__proc_task.is_finished():
                self.__proc_task = None


    def is_finished(self):
        return self.__proc_task is None and len(self.__tasks) == 0


    def get_idx(self):
        return self.__idx


    def kill(self):
        pass # TODO


def __idx_lock(doc, db):
    doc['status'] = 'Processed'
    doc['host'] = socket.gethostname()
    db.save(doc)


def __idx_unlock(doc, db):
    doc['status'] = 'Done'
    db.save(doc)


def __idx_locked(doc):
    return doc['status'] != 'Waiting'


def update_tasks(couch):
    if 'tasks' not in couch:
        couch.create('tasks')
    db = couch['tasks']

    if 'config' not in db:
        db['config'] = {'names': [], 'opts': {}, 'locked': False}

    doc = db['config']

    while doc['locked']:
        time.sleep(0.2)
        doc = db['config']

    doc['locked'] = True
    db[doc.id] = doc

    doc = db['config']
    conf_names = doc['names']
    conf_opts = doc['opts']

{'fraction_non_alphanumeric': 0.11736406354889237, 'fraction_numerical': 0.00380398299395838, 'mean_word_length': 2.285189268651231, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '3029278', 'n_tokens_mistral': 2784, 'n_tokens_neox': 2684, 'n_words': 682}