<commit_before>#! /usr/bin/env python3
# vim:ts=4:sw=4:ai:et:si:sts=4

import logging
from flask import Flask, request
from flask_jsonrpc import JSONRPC
from flask_jsonrpc.helpers import extract_raw_data_request
import os
import sys
import shutil
import json
from subprocess import Popen, PIPE, STDOUT
from bs4 import BeautifulSoup
from threading import Thread, Lock
from queue import Queue
import time
import selectors

logdir = "/opt/video/render/logs"
logfile = os.path.join(logdir, "rpcserver.log")

FORMAT = "%(asctime)s: %(name)s:%(lineno)d (%(threadName)s) - %(levelname)s - %(message)s"
logging.basicConfig(filename=logfile, format=FORMAT)
logging.getLogger(None).setLevel(logging.INFO)
logging.captureWarnings(True)
logger = logging.getLogger(__name__)

queues = {
    "cpu-bound": Queue(),
    "internal-network-bound": Queue(),
    "external-network-bound": Queue(),
    "local": Queue(),
}

queueMap = {
    "upload_inputs": "internal-network-bound",
    "convert_inputs": "cpu-bound",
    "download_proxies": "internal-network-bound",
    "download_editables": "internal-network-bound",
    "upload_edl": "local",
    "upload_proxy_edl": "local",
    "render_edl": "cpu-bound",
    "upload_to_youtube": "external-network-bound",
    "archive_to_s3": "external-network-bound",
    "make_slideshow": "cpu-bound",
}

handlerThreads = {}
handlers = {}
outputThread = None


class OutputThread(Thread):
    def __init__(self):
        Thread.__init__(self)

        self.sel = selectors.DefaultSelector()
        self.idMap = {}
        self.name = "OutputThread"
        self.daemon = True
        global handlers
        self.handlers = handlers

    def run(self):
        while True:
            events = self.sel.select(0.1)
            for (key, mask) in events:
                callback = key.data
                callback(key.fileobj, mask)

    def add(self, id_):
        fd = self.handlers[id_]['pipe']
        self.idMap[fd] = id_
        logger.info("Adding %s (%s)" % (fd, id_))
        self.sel.register(fd, selectors.EVENT_READ, self.read)

    def removeId(self, id_):
        fd = self.handlers[id_]['pipe']
        logger.info("Removing %s (%s)" % (fd, id_))
        self.remove(fd)

    def remove(self, fd):
        del self.idMap[fd]
        self.sel.unregister(fd)<commit_msg>Ignore already registered pipes.  This shouldn't happen!<commit_after>#! /usr/bin/env python3
# vim:ts=4:sw=4:ai:et:si:sts=4

import logging
from flask import Flask, request
from flask_jsonrpc import JSONRPC
from flask_jsonrpc.helpers import extract_raw_data_request
import os
import sys
import shutil
import json
from subprocess import Popen, PIPE, STDOUT
from bs4 import BeautifulSoup
from threading import Thread, Lock
from queue import Queue
import time
import selectors

logdir = "/opt/video/render/logs"
logfile = os.path.join(logdir, "rpcserver.log")

FORMAT = "%(asctime)s: %(name)s:%(lineno)d (%(threadName)s) - %(levelname)s - %(message)s"
logging.basicConfig(filename=logfile, format=FORMAT)
logging.getLogger(None).setLevel(logging.INFO)
logging.captureWarnings(True)
logger = logging.getLogger(__name__)

queues = {
    "cpu-bound": Queue(),
    "internal-network-bound": Queue(),
    "external-network-bound": Queue(),
    "local": Queue(),
}

queueMap = {
    "upload_inputs": "internal-network-bound",
    "convert_inputs": "cpu-bound",
    "download_proxies": "internal-network-bound",
    "download_editables": "internal-network-bound",
    "upload_edl": "local",
    "upload_proxy_edl": "local",
    "render_edl": "cpu-bound",
    "upload_to_youtube": "external-network-bound",
    "archive_to_s3": "external-network-bound",
    "make_slideshow": "cpu-bound",
}

handlerThreads = {}
handlers = {}
outputThread = None


class OutputThread(Thread):
    def __init__(self):
        Thread.__init__(self)

        self.sel = selectors.DefaultSelector()
        self.idMap = {}
        self.name = "OutputThread"
        self.daemon = True
        global handlers
        self.handlers = handlers

    def run(self):
        while True:
            events = self.sel.select(0.1)
            for (key, mask) in events:
                callback = key.data
                callback(key.fileobj, mask)

    def add(self, id_):
        fd = self.handlers[id_]['pipe']
        self.idMap[fd] = id_
        logger.info("Adding %s (%s)" % (fd, id_))
        try:
            self.sel.register(fd, selectors.EVENT_READ, self.read)
        except KeyError:
            pass

    def removeId(self, id_):
        fd = self.handlers[id_]['pipe']
        logger.info("Removing %s (%s)" % (fd, id_))
        self.remove(fd)

    def remove(self, fd):
        del self.idMap[fd]
        self.sel.unregister(fd)
{'fraction_non_alphanumeric': 0.13804071246819338, 'fraction_numerical': 0.0033927056827820186, 'mean_word_length': 3.2153708668453977, 'pattern_counts': {'":': 28, '<': 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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6776041', 'n_tokens_mistral': 1578, 'n_tokens_neox': 1550, 'n_words': 337}