<commit_before>"""Contains code for managing and processing optimisation requests."""

import datetime
import importlib
import multiprocessing as mp
import time
import os
import sys

import database
import model_building


def main():
    """Establishes the manager and listener subprocesses"""
    processes = int(os.getenv(key='OPT_PROCS', default='1'))
    queue = mp.Queue()
    listeners = [
        mp.Process(target=get_and_process_opt_jobs,
                   args=(queue, ))
        for _ in range(processes)
    ]
    for listener in listeners:
        listener.start()

    while True:
        jobs = database.opt_jobs.find(
            {'status': database.JobStatus.SUBMITTED.name})
        for job in sorted(jobs, key=lambda x: x['time_submitted']):
            queue.put(job['_id'])
            database.opt_jobs.update_one(
                {'_id': job['_id']},
                {'$set': {'status': database.JobStatus.QUEUED.name}})
        time.sleep(10)
    return


def get_and_process_opt_jobs(opt_job_queue):
    """Collect and run optimisation jobs from queue.

    Used by the the OptimizationManager to initialise Processes.

    Parameters
    ----------
    opt_job_queue : multiprocessing.Queue
        Optimisation job queue.
    """
    # The module is reloaded to establish a new connection
    # to the database for the process fork
    importlib.reload(database)
    while True:
        job_id = opt_job_queue.get()
        print("Got opt job {}!".format(job_id), file=sys.stderr)
        opt_job = database.opt_jobs.find_one(job_id)
        parameters = list(map(
            database.parameters_store.find_one,
            opt_job['initial_parameter_ids']))
        update_job_status(job_id, database.JobStatus.RUNNING)
        print("Running opt job {}!".format(job_id), file=sys.stderr)
        model_id = run_optimisation(job_id, opt_job['helix_type'], parameters)
        print("Finished opt job {}!".format(job_id), file=sys.stderr)
    return


def update_job_status(opt_job_id, status):
    """Updates status in database entry for optimisation job."""<commit_msg>Adds assigned_jobs list of running jobs.<commit_after>"""Contains code for managing and processing optimisation requests."""

import datetime
import importlib
import multiprocessing as mp
import time
import os
import sys

import database
import model_building


def main():
    """Establishes the manager and listener subprocesses"""
    processes = int(os.getenv(key='OPT_PROCS', default='1'))
    with mp.Manager() as manager:
        queue = manager.Queue()
        assigned_jobs = manager.list([None] * processes)
        listeners = [
            mp.Process(target=get_and_process_opt_jobs,
                       args=(queue, assigned_jobs, proc_i))
            for proc_i in range(processes)
        ]
        for listener in listeners:
            listener.start()

        while True:
            submitted_jobs = database.opt_jobs.find(
                {'status': database.JobStatus.SUBMITTED.name})
            for job in sorted(submitted_jobs,
                              key=lambda x: x['time_submitted']):
                queue.put(job['_id'])
                database.opt_jobs.update_one(
                    {'_id': job['_id']},
                    {'$set': {'status': database.JobStatus.QUEUED.name}})
            running_jobs = database.opt_jobs.find(
                {'status': database.JobStatus.RUNNING.name})
            for job in running_jobs:
                if job['_id'] not in assigned_jobs:
                    update_job_status(job['_id'], database.JobStatus.FAILED)
            time.sleep(10)
    return


def get_and_process_opt_jobs(opt_job_queue, assigned_jobs, proc_i):
    """Collect and run optimisation jobs from queue.

    Used by the the OptimizationManager to initialise Processes.

    Parameters
    ----------
    opt_job_queue : multiprocessing.Queue
        Optimisation job queue.
    """
    # The module is reloaded to establish a new connection
    # to the database for the process fork
    importlib.reload(database)
    while True:
        job_id = opt_job_queue.get()
        print("Got opt job {}!".format(job_id), file=sys.stderr)
        opt_job = database.opt_jobs.find_one(job_id)
        parameters = list(map(
            database.parameters_store.find_one,
            opt_job['initial_parameter_ids']))
        update_job_status(job_id, database.JobStatus.RUNNING)
        assigned_jobs[proc_i] = job_id
        print("Running opt job {}!".format(job_id), file=sys.stderr)
        run_optimisation(job_id, opt_job['helix_type'],
                         parameters)
        print("Finished opt job {}!".format(job_id), file=sys.stderr)
        assigned_jobs[proc_i] = None
    return


def update_job_status(opt_job_id, status):
    """Updates status in database entry for optimisation job."""
{'fraction_non_alphanumeric': 0.10493827160493827, 'fraction_numerical': 0.0012345679012345679, 'mean_word_length': 2.5927568366592757, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '2199701', 'n_tokens_mistral': 1421, 'n_tokens_neox': 1380, 'n_words': 344}