<commit_before>/**
 * Independent/Isolated file transform
 */
const analyticsCollector = require('../helpers/analytics/analytics-collector');
const analytics = require('../helpers/analytics/analytics')('ipc');
const debug = require('debug')('mendel:transformer:master');
const {fork} = require('child_process');

const parallelMode = (function() {
    const numCPUs = require('os').cpus().length;
    const workerProcesses = Array.from(Array(numCPUs)).map(() => fork(`${__dirname}/worker.js`));
    workerProcesses.forEach(cp => analyticsCollector.connectProcess(cp));
    const _idleWorkerQueue = workerProcesses.map(({pid}) => pid);
    const _queue = [];

    function next() {
        if (!_queue.length || !_idleWorkerQueue.length) return;

        const {filename, source, transforms, resolves, rejects} = _queue.shift();
        const idlePid = _idleWorkerQueue.shift();
        const workerProcess = workerProcesses.find(({pid}) => idlePid === pid);

        workerProcess.on('message', function onMessage({error, type, source, map}) {
            // debug(`[Master] <- [Slave ${workerProcess.pid}]: ${type}`);
            if (type !== 'error' && type !== 'done') return;
            if (type === 'error') {
                rejects.forEach(reject => reject(new Error(error)));
            } else if (type === 'done') {
                _idleWorkerQueue.push(workerProcess.pid);
                resolves.forEach(resolve => resolve({source, map}));
            }

            workerProcess.removeListener('message', onMessage);
            next();
        });
        analytics.tic('transform');
        workerProcess.send({type: 'start', transforms, filename, source});
        analytics.toc('transform');
        next();
    }

    process.on('exit', () => {
        workerProcesses.forEach(workerProcess => workerProcess.kill());
    });

    return function queue(transforms, {source, filename}) {
        return new Promise((resolve, reject) => {
            const existingJob = _queue.find(queued => queued.filename === filename && queued.transforms === transforms);

            if (existingJob) {
                existingJob.resolves.push(resolve);
                existingJob.rejects.push(reject);
            } else {
                _queue.push({
                    resolves: [resolve],
                    rejects: [reject],
                    transforms,
                    source,
                    filename,
                });
            }

            next();
        });
    };
})();

function singleMode(transforms, {filename, source}) {
    let promise = Promise.resolve();

    transforms.forEach(transform => {
        promise = promise.then(() => {
            const xform = typeof transform.plugin === 'string' ? require(transform.plugin) : transform.plugin;
            return xform({filename, source}, transform.options);
        });
    });

    return promise;
}

/**
 * Knows how to do all kinds of trasnforms in parallel way
 */
class TrasnformManager {
    constructor({transforms}) {
        this._transforms = new Map();
        transforms.forEach(transform => {
            this._transforms.set(transform.id, transform);
        });
    }

    transform(filename, transformIds, source) {
        debug(`Transforming "${filename}" with [${transformIds}]`);
        const transforms = transformIds.map(id => this._transforms.get(id));

        let mode;
        if (transformIds.every(transformId => typeof this._transforms.get(transformId).plugin === 'string')) {
            mode = parallelMode;
        } else {
            console.log('[MENDEL][deopt] The transform is not a known one to Mendel and we cannot parallelize it. Please contribute to IFT plugin for faster build.');
            mode = singleMode;
        }

        return mode(transforms, {filename, source});
    }
}

module.exports = TrasnformManager;<commit_msg>trasnformer to transforer<commit_after>/**
 * Independent/Isolated file transform
 */
const analyticsCollector = require('../helpers/analytics/analytics-collector');
const analytics = require('../helpers/analytics/analytics')('ipc');
const debug = require('debug')('mendel:transformer:master');
const {fork} = require('child_process');

const parallelMode = (function() {
    const numCPUs = require('os').cpus().length;
    const workerProcesses = Array.from(Array(numCPUs)).map(() => fork(`${__dirname}/worker.js`));
    workerProcesses.forEach(cp => analyticsCollector.connectProcess(cp));
    const _idleWorkerQueue = workerProcesses.map(({pid}) => pid);
    const _queue = [];

    function next() {
        if (!_queue.length || !_idleWorkerQueue.length) return;

        const {filename, source, transforms, resolves, rejects} = _queue.shift();
        const idlePid = _idleWorkerQueue.shift();
        const workerProcess = workerProcesses.find(({pid}) => idlePid === pid);

        workerProcess.on('message', function onMessage({error, type, source, map}) {
            // debug(`[Master] <- [Slave ${workerProcess.pid}]: ${type}`);
            if (type !== 'error' && type !== 'done') return;
            if (type === 'error') {
                rejects.forEach(reject => reject(new Error(error)));
            } else if (type === 'done') {
                _idleWorkerQueue.push(workerProcess.pid);
                resolves.forEach(resolve => resolve({source, map}));
            }

            workerProcess.removeListener('message', onMessage);
            next();
        });
        analytics.tic('transform');
        workerProcess.send({type: 'start', transforms, filename, source});
        analytics.toc('transform');
        next();
    }

    process.on('exit', () => {
        workerProcesses.forEach(workerProcess => workerProcess.kill());
    });

    return function queue(transforms, {source, filename}) {
        return new Promise((resolve, reject) => {
            const existingJob = _queue.find(queued => queued.filename === filename && queued.transforms === transforms);

            if (existingJob) {
                existingJob.resolves.push(resolve);
                existingJob.rejects.push(reject);
            } else {
                _queue.push({
                    resolves: [resolve],
                    rejects: [reject],
                    transforms,
                    source,
                    filename,
                });
            }

            next();
        });
    };
})();

function singleMode(transforms, {filename, source}) {
    let promise = Promise.resolve();

    transforms.forEach(transform => {
        promise = promise.then(() => {
            const xform = typeof transform.plugin === 'string' ? require(transform.plugin) : transform.plugin;
            return xform({filename, source}, transform.options);
        });
    });

    return promise;
}

/**
 * Knows how to do all kinds of trasnforms in parallel way
 */
class TransformManager {
    constructor({transforms}) {
        this._transforms = new Map();
        transforms.forEach(transform => {
            this._transforms.set(transform.id, transform);
        });
    }

    transform(filename, transformIds, source) {
        debug(`Transforming "${filename}" with [${transformIds}]`);
        const transforms = transformIds.map(id => this._transforms.get(id));

        let mode;
        if (transformIds.every(transformId => typeof this._transforms.get(transformId).plugin === 'string')) {
            mode = parallelMode;
        } else {
            console.log('[MENDEL][deopt] The transform is not a known one to Mendel and we cannot parallelize it. Please contribute to IFT plugin for faster build.');
            mode = singleMode;
        }

        return mode(transforms, {filename, source});
    }
}

module.exports = TransformManager;
{'fraction_non_alphanumeric': 0.14506570471527958, 'fraction_numerical': 0.0, 'mean_word_length': 2.4517563361494, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 33, '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': '5318826', 'n_tokens_mistral': 2067, 'n_tokens_neox': 2013, 'n_words': 451}