<commit_before>/*jslint node: true */
'use strict';

var jpp = require('json-path-processor'),
    cache = require('simple-lru-cache'),
    taskpool = null,

later = function (func, I) {
    setTimeout(function () {
        func(I);
    }, 1);
},

safelater = function (func, I) {
    if ('function' === (typeof func)) {
        later(func, I);
    }
},

subtask = function (tasks) {
    var executed = false,
        type = (typeof tasks),
        count = 0,
        all = 0,
        result = {},
        callbacks = [],
        runner = function (index, task) {
            try {
                task.execute(function (D) {
                    result[index] = D;
                    ender();
                });
            } catch (E) {
                later(function () {
                    result[index] = ('function' === (typeof task)) ? undefined : task;
                    ender();
                });
            }
        },
        ender = function () {
            count++;
            if (count === all) {
                executed = true;
                while (callbacks.length) {
                    safelater(callbacks.pop(), result);
                }
            }
        };

    this.execute = function (cb) {
        // do nothing when no subtask or input string
        if (!tasks || ('string' === type)) {
            cb(tasks);
            return this;
        }

        // executed, return cached result
        if (executed) {
            cb(result);
            return this;
        }

        // wrap a function
        if ('function' === type) {
            try {
                tasks(function (D) {
                    result = D;
                    executed = true;
                    cb(result);
                });
            } catch (E) {
                result = undefined;
                executed = true;
                cb();
            }
            return this;
        }

        // wait for result
        callbacks.push(cb);

        // started, not again
        if (all) {
            return this;
        }

        // execute
        for (var I in tasks) {
            all++;
            runner(I, tasks[I]);
        }

        if (all === 0) {
            callbacks.pop()(tasks);
        }

        return this;
    };
},

SUBTASK = function(tasks) {
    return new subtask(tasks);
};

SUBTASK.isSubtask = function (O) {
    return O instanceof subtask;
};

SUBTASK.initCache = function (size) {
    if (taskpool) {
        taskpool.reset();
    }
    taskpool = new cache({maxSize: size});
};

SUBTASK.cache = function (tasks, key) {
    var T,
        thisPool = (this && this.pool) ? this.pool : false,
        thisKey = ((this && this.taskKey) ? this.taskKey : '') + key;

    if (thisPool && thisPool[thisKey]) {
        T = thisPool[thisKey];
    }

    if (taskpool) {
        T = taskpool.get(thisKey);
    }

    if (!T) {
        T = SUBTASK(tasks);
    }

    if (thisPool) {
        thisPool[thisKey] = T;
    }

    if (taskpool) {
        taskpool.set(thisKey, T);
    }

    T.taskKey = thisKey;
    return T;
};

subtask.prototype = {
    pipe: function (task) {
        var T = this;<commit_msg>support timeout<commit_after>/*jslint node: true */
'use strict';

var jpp = require('json-path-processor'),
    cache = require('simple-lru-cache'),
    taskpool = null,

later = function (func, I) {
    setTimeout(function () {
        func(I);
    }, 1);
},

safelater = function (func, I) {
    if ('function' === (typeof func)) {
        later(func, I);
    }
},

subtask = function (tasks) {
    var executed = false,
        type = (typeof tasks),
        count = 0,
        all = 0,
        result = {},
        callbacks = [],
        runner = function (index, task) {
            try {
                task.execute(function (D) {
                    result[index] = D;
                    ender();
                });
            } catch (E) {
                later(function () {
                    result[index] = ('function' === (typeof task)) ? undefined : task;
                    ender();
                });
            }
        },
        ender = function () {
            count++;
            if (count === all) {
                executed = true;
                while (callbacks.length) {
                    safelater(callbacks.pop(), result);
                }
            }
        };

    this.execute = function (cb) {
        // do nothing when no subtask or input string
        if (!tasks || ('string' === type)) {
            cb(tasks);
            return this;
        }

        // executed, return cached result
        if (executed) {
            cb(result);
            return this;
        }

        // wrap a function
        if ('function' === type) {
            try {
                tasks(function (D) {
                    result = D;
                    executed = true;
                    cb(result);
                });
            } catch (E) {
                result = undefined;
                executed = true;
                cb();
            }
            return this;
        }

        // wait for result
        callbacks.push(cb);

        // started, not again
        if (all) {
            return this;
        }

        // execute
        for (var I in tasks) {
            all++;
            runner(I, tasks[I]);
        }

        if (all === 0) {
            callbacks.pop()(tasks);
        }

        return this;
    };
},

SUBTASK = function(tasks) {
    return new subtask(tasks);
};

SUBTASK.isSubtask = function (O) {
    return O instanceof subtask;
};

SUBTASK.initCache = function (size) {
    if (taskpool) {
        taskpool.reset();
    }
    taskpool = new cache({maxSize: size});
};

SUBTASK.cache = function (tasks, key, timeout) {
    var T,
        thisPool = (this && this.pool) ? this.pool : false,
        thisKey = ((this && this.taskKey) ? this.taskKey : '') + key,
        now;

    if (timeout) {
        now = (new Date()).getTime();
    }

    if (thisPool && thisPool[thisKey]) {
        T = thisPool[thisKey];
    }

    if (taskpool) {
        T = taskpool.get(thisKey);
    }

    if (timeout && ((T.taskTime + timeout) < now)) {
        T = undefined;
    }

    if (!T) {
        T = SUBTASK(tasks);
    }

    if (thisPool) {
        thisPool[thisKey] = T;
    }

    if (taskpool) {
        taskpool.set(thisKey, T);
    }

    T.taskKey = thisKey;
    T.taskTime = now;

    return T;
};

subtask.prototype = {
    pipe: function (task) {
        var T = this;
{'fraction_non_alphanumeric': 0.13644716692189893, 'fraction_numerical': 0.001225114854517611, 'mean_word_length': 1.2474191328286304, 'pattern_counts': {'":': 0, '<': 4, '<?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': '1239553', 'n_tokens_mistral': 1948, 'n_tokens_neox': 1869, 'n_words': 460}