<commit_before>'use strict';

const _ = require('lodash');
const pWhilst = require('p-whilst');
const Promise = require('bluebird');
const retry = require('bluebird-retry');
const parseError = require('@terascope/error-parser');
const Messaging = require('@terascope/teraslice-messaging');

const Scheduler = require('./scheduler');
const ExecutionAnalytics = require('./analytics');
const makeExecutionRecovery = require('./recovery');
const makeSliceAnalytics = require('./slice-analytics');
const { waitForWorkerShutdown } = require('../helpers/worker-shutdown');
const { makeStateStore, makeExStore } = require('../../cluster/storage');
const { makeLogger, generateWorkerId } = require('../helpers/terafoundation');

const ExecutionControllerServer = Messaging.ExecutionController.Server;
const ClusterMasterClient = Messaging.ClusterMaster.Client;
const { formatURL } = Messaging;

// const immediate = Promise.promisify(setImmediate);

class ExecutionController {
    constructor(context, executionContext) {
        const workerId = generateWorkerId(context);
        const logger = makeLogger(context, executionContext, 'execution_controller');
        const events = context.apis.foundation.getSystemEvents();
        const slicerPort = executionContext.slicer_port;
        const networkLatencyBuffer = _.get(context, 'sysconfig.teraslice.network_latency_buffer');
        const actionTimeout = _.get(context, 'sysconfig.teraslice.action_timeout');
        const workerDisconnectTimeout = _.get(context, 'sysconfig.teraslice.worker_disconnect_timeout');
        const nodeDisconnectTimeout = _.get(context, 'sysconfig.teraslice.node_disconnect_timeout');
        const shutdownTimeout = _.get(context, 'sysconfig.teraslice.shutdown_timeout');

        this.server = new ExecutionControllerServer({
            port: slicerPort,
            networkLatencyBuffer,
            actionTimeout,
            workerDisconnectTimeout,
        });

        const clusterMasterPort = _.get(context, 'sysconfig.teraslice.port');
        const clusterMasterHostname = _.get(context, 'sysconfig.teraslice.master_hostname');

        this.client = new ClusterMasterClient({
            clusterMasterUrl: formatURL(clusterMasterHostname, clusterMasterPort),
            executionContext,
            networkLatencyBuffer,
            actionTimeout,
            connectTimeout: nodeDisconnectTimeout,
            exId: executionContext.ex_id,
        });

        this.executionAnalytics = new ExecutionAnalytics(
            context,
            executionContext,
            this.client
        );

        this.scheduler = new Scheduler(context, executionContext);

        this.exId = _.get(executionContext, 'config.ex_id');
        this.workerId = workerId;
        this.logger = logger;
        this.events = events;
        this.context = context;
        this.executionContext = executionContext;
        this.collectAnalytics = this.executionContext.config.analytics;
        this.shutdownTimeout = shutdownTimeout;
        this.workerDisconnectTimeout = workerDisconnectTimeout;
        this.stores = {};

        this.isPaused = false;
        this.isShutdown = false;
        this.isShuttingDown = false;
        this.isInitialized = false;
        this.isStarted = false;
        this.pendingDispatches = 0;
        this.pendingSlices = 0;
        this.isDoneProcessing = false;
        this.isExecutionFinished = false;
        this.isExecutionDone = false;
        this.workersHaveConnected = false;
        this._handlers = {};

        this.setFailingStatus = this.setFailingStatus.bind(this);
        this.terminalError = this.terminalError.bind(this);
    }

    async initialize() {
        const { context } = this;

        const exStore = makeExStore(context);
        this.stores.exStore = await exStore;

        await this.client.start();

        let verified;
        let verifiedErr;
        try {
            verified = await this._verifyExecution();
        } catch (err) {
            verifiedErr = err;
        }
        if (!verified) {
            this.isShutdown = true;
            await this.stores.exStore.shutdown(true);
            await this.client.shutdown();
            if (verifiedErr) {
                throw verifiedErr;
            }
            return;
        }

        const stateStore = makeStateStore(context);
        this.stores.stateStore = await stateStore;

        await this.server.start();

        this.isInitialized = true;

        this.server.onClientOnline((workerId) => {
            this.logger.trace(`worker ${workerId} is online`);
            this._adjustSlicerQueueLength();
            this.workersHaveConnected = true;
            clearTimeout(this.workerConnectTimeoutId);
            this.workerConnectTimeoutId = null;
            this.executionAnalytics.increment('workers_joined');
        });

        this.server.onClientAvailable((workerId) => {
            this.logger.trace(`worker ${workerId} is available`);
            this.executionAnalytics.set('workers_available', this.server.availableClientCount);
            this.executionAnalytics.set('workers_active', this.server.activeWorkers.length);
        });

        this.server.onClientUnavailable(() => {
            this.executionAnalytics.set('workers_active', this.server.activeWorkers.length);
            this.executionAnalytics.set('workers_available', this.server.availableClientCount);
        });

        this.server.onClientDisconnect((workerId) => {
            this.logger.trace(`worker ${workerId} is disconnected but it may reconnect`);
            this.executionAnalytics.increment('workers_disconnected');
            this.executionAnalytics.set('workers_active', this.server.activeWorkers.length);

            this._startWorkerDisconnectWatchDog();
        });

        this.server.onClientOffline((workerId) => {
            this.logger.trace(`worker ${workerId} is offline`);
            this._adjustSlicerQueueLength();
        });

        this.server.onClientReconnect((workerId) => {
            clearTimeout(this.workerDisconnectTimeoutId);
            this.workerConnectTimeoutId = null;

            this.logger.trace(`worker ${workerId} is reconnected`);
            this.executionAnalytics.increment('workers_reconnected');
        });

        this.client.onExecutionPause(() => this.pause());
        this.client.onExecutionResume(() => this.resume());
        this.client.onServerShutdown(() => {
            this.logger.warn('Cluster Master shutdown, exiting...');
            this.executionAnalytics.sendingAnalytics = false;
            this._endExecution();
        });

        this.server.onSliceSuccess((workerId, response) => {
            this.logger.info(`worker ${workerId} has completed its slice ${response.slice_id}`);
            this.events.emit('slice:success', response);
            this.pendingSlices -= 1;
        });

        this.server.onSliceFailure((workerId, response) => {
            this.logger.error(`worker: ${workerId} has failure completing its slice`, response);
            this.events.emit('slice:failure', response);

            if (this.scheduler.canComplete()) {
                this.setFailingStatus();
            } else if (this.scheduler.isRecovering()) {
                this.terminalError(new Error('Slice failed while recovering'));
            } else {
                // in persistent mode we set watchdogs to monitor
                // when failing can be set back to running
                this._checkAndUpdateExecutionState();
            }

            this.pendingSlices -= 1;
        });

        this._handlers['slicer:execution:update'] = ({ update }) => {
            this.logger.trace('slicer sending a execution update', update);

            // this is updating the opConfig for elasticsearch start and/or end dates for ex,
            // this assumes elasticsearch is first
            this.stores.exStore.update(this.exId, { operations: update });
        };

        this._handlers['slicers:finished'] = (err) => {
            if (err) {
                this.logger.error('slicers finished due to failure', err);
                this.terminalError(err);
            }
        };

        _.forEach(this._handlers, (handler, event) => {
            this.events.on(event, handler);
        });

        this.logger.debug(`execution ${this.exId} is initialized`);

        this.isInitialized = true;
    }

    async run() {
        if (this.isShuttingDown) {
            this.logger.error('Cannot run execution while shutting down');
            return;
        }

        if (!this.isInitialized) {
            this.logger.error('Cannot run execution is not initialized');
            return;
        }

        try {
            await this._runExecution();
        } catch (err) {
            this.logger.error('Run execution error', err);
        }

        this.events.emit('worker:shutdown');

        // help the workers go offline
        this.server.isShuttingDown = true;

        await this._finishExecution();

        try {
            await Promise.all([
                this.client.sendExecutionFinished(),
                this._waitForWorkersToExit(),
            ]);
        } catch (err) {
            this.logger.error('Failure sending execution finished', err);
        }

        this.logger.debug(`execution ${this.exId} is done`);
    }

    async resume() {
        if (!this.isPaused) return;

        this.logger.info(`execution ${this.exId} is resuming...`);

        this.isPaused = false;
        this.scheduler.start();

        await Promise.delay(100);
    }

    async pause() {
        if (this.isPaused) return;

        this.logger.info(`execution ${this.exId} is pausing...`);

        this.isPaused = true;
        this.scheduler.pause();

        await Promise.delay(100);
    }

    async setFailingStatus() {
        const { exStore } = this.stores;

        const errMsg = `slicer: ${this.exId} has encountered a processing_error`;
        this.logger.error(errMsg);

        const executionStats = this.executionAnalytics.getAnalytics();
        const errorMeta = await exStore.executionMetaData(executionStats, errMsg);
        await exStore.setStatus(this.exId, 'failing', errorMeta);
    }

    async terminalError(err) {
        if (this.isExecutionDone) return;

        const { exStore } = this.stores;

        this.slicerFailed = true;

        const msg = `slicer for ex ${this.exId} had an error, shutting down execution`;
        this.logger.error(msg, err);

        const errMsg = `${msg}, caused by ${err.stack ? err.stack : _.toString(err)}`;

        const executionStats = this.executionAnalytics.getAnalytics();
        const errorMeta = await exStore.executionMetaData(executionStats, errMsg);

        await exStore.setStatus(this.exId, 'failed', errorMeta);

        this.logger.fatal(`execution ${this.exId} is done because of slice failure`);
        this._endExecution();
    }

    async shutdown(block = true) {
        if (this.isShutdown) return;
        if (!this.isInitialized) return;

        if (this.isShuttingDown && block) {
            this.logger.debug(`execution shutdown was called for ex ${this.exId} but it was already shutting down, will block until done`);
            await waitForWorkerShutdown(this.context, 'worker:shutdown:complete');
            return;
        }

        this.logger.debug(`execution shutdown was called for ex ${this.exId}`);

        const shutdownErrs = [];

        // allow clients to go immediately from disconnect to offline
        this.server.isShuttingDown = true;

        // tell the scheduler to stop producing slices
        this.scheduler.stop();

        // remove any listeners
        _.forEach(this._handlers, (handler, event) => {
            this.events.removeListener(event, handler);
        });

        this.isShuttingDown = true;
        this.isPaused = false;

        clearInterval(this.watcher);
        clearTimeout(this.workerConnectTimeoutId);
        clearTimeout(this.workerDisconnectTimeoutId);

        await this._waitForExecutionFinished();

        try {
            await this.executionAnalytics.shutdown();
        } catch (err) {
            this.logger.error('execution analytics error');
        }

        this.scheduler.cleanup();

        if (this.recover) {
            try {
                await this.recover.shutdown();
            } catch (err) {
                shutdownErrs.push(err);
            }
        }

        try {
            await this.server.shutdown();
        } catch (err) {
            shutdownErrs.push(err);
        }

        try {
            await this.client.shutdown();
        } catch (err) {
            shutdownErrs.push(err);
        }

        try {
            await Promise.map(Object.values(this.stores), store => store.shutdown(true));
        } catch (err) {
            shutdownErrs.push(err);
        }

        this.logger.warn(`execution controller ${this.exId} is shutdown`);
        this.isShutdown = true;

        if (shutdownErrs.length) {
            const errMsg = shutdownErrs.map(e => e.stack).join(', and');
            const shutdownErr = new Error(`Failed to shutdown correctly: ${errMsg}`);
            this.events.emit(this.context, 'worker:shutdown:complete', shutdownErr);
            await Promise.delay(0);
            throw shutdownErr;
        }

        this.events.emit(this.context, 'worker:shutdown:complete');
    }

    async _adjustSlicerQueueLength() {
        const { dynamicQueueLength, queueLength } = this.executionContext;
        if (!dynamicQueueLength) return;

        if (this.server.onlineClientCount > queueLength) {
            this.executionContext.queueLength = this.server.onlineClientCount;
            this.logger.info(`adjusted queue length ${this.executionContext.queueLength}`);
        }
    }

    async _runExecution() {
        this._startWorkConnectWatchDog();

        this.slicerAnalytics = makeSliceAnalytics(this.context, this.executionContext);

        this.executionAnalytics.start();

        this.logger.info(`starting execution ${this.exId}...`);
        this.startTime = Date.now();

        // attach to scheduler
        this.scheduler.ensureSliceState = this.ensureSliceState.bind(this);

        this.isStarted = true;

        if (this.scheduler.recoverExecution) {
            await this._recoverSlicesInit();
        } else {
            await this._slicerInit();
        }

        // wait for paused
        await pWhilst(() => this.isPaused && !this.isShuttdown, () => Promise.delay(100));

        await Promise.all([
            this.stores.exStore.setStatus(this.exId, 'running'),
            this.client.sendAvailable(),
            this._waitForRecovery(),
            this._runDispatch(),
            this.scheduler.run(),
        ]);

        const schedulerSuccessful = this.scheduler.isFinished && this.scheduler.slicersDone;

        await this._waitForPendingSlices();

        if (schedulerSuccessful && this.isDoneDispatching) {
            this.logger.debug(`execution ${this.exId} is done processing slices`);
            this.isDoneProcessing = true;
        } else if (!this.isShuttdown) {
            this.logger.debug(`execution ${this.exId} did not finish`);
        } else {
            this.logger.debug(`execution ${this.exId} is exiting...`);
        }
    }

    // dispatching should be pushed out into its own module
    // this function uses many tiny unbound functions to
    // increase performance
    async _runDispatch() {
        const {
            logger,
            scheduler,
            server,
        } = this;

        this.isDoneDispatching = false;

        logger.debug('dispatching slices...');

        // returns a boolean to indicate whether
        // dispatching should continue
        const isRunning = () => {
            if (this.isShuttingDown) return false;
            if (this.isExecutionDone) return false;
            if (scheduler.isFinished
                && !this.pendingDispatches) return false;
            return true;
        };

        const isPaused = () => this.isPaused;
        const dispatchSlice = (slice, workerId) => this._dispatchSlice(slice, workerId);

        // this isn't really ideal since we adding
        // to the beginning of the queue and
        // it may end up in a recursive loop trying
        // to process that slice
        function reenqueueSlice(slice) {
            return scheduler.enqueueSlice(slice, true);
        }

        function dequeueAndDispatch() {
            const slices = scheduler.getSlices(server.workerQueueSize);
            if (slices.length === 0) return;

            // eslint-disable-next-line no-restricted-syntax
            for (const slice of slices) {
                const workerId = server.dequeueWorker(slice);
                if (!workerId) {
                    process.nextTick(() => {
                        reenqueueSlice(slice);
                    });
                    continue;
                }

                dispatchSlice(slice, workerId);
            }
        }

        await new Promise((resolve) => {
            const intervalId = setInterval(dispatch, 1);

            function dispatch() {
                if (!isRunning()) {
                    clearInterval(intervalId);
                    resolve();
                    return;
                }

                if (isPaused()) return;

                const workers = server.workerQueueSize;
                const slices = scheduler.queueLength;

                if (workers > 0 && slices > 0) {
                    dequeueAndDispatch();
                }
            }<commit_msg>Use setInterval of 5ms in dispatch<commit_after>'use strict';

const _ = require('lodash');
const pWhilst = require('p-whilst');
const Promise = require('bluebird');
const retry = require('bluebird-retry');
const parseError = require('@terascope/error-parser');
const Messaging = require('@terascope/teraslice-messaging');

const Scheduler = require('./scheduler');
const ExecutionAnalytics = require('./analytics');
const makeExecutionRecovery = require('./recovery');
const makeSliceAnalytics = require('./slice-analytics');
const { waitForWorkerShutdown } = require('../helpers/worker-shutdown');
const { makeStateStore, makeExStore } = require('../../cluster/storage');
const { makeLogger, generateWorkerId } = require('../helpers/terafoundation');

const ExecutionControllerServer = Messaging.ExecutionController.Server;
const ClusterMasterClient = Messaging.ClusterMaster.Client;
const { formatURL } = Messaging;

// const immediate = Promise.promisify(setImmediate);

class ExecutionController {
    constructor(context, executionContext) {
        const workerId = generateWorkerId(context);
        const logger = makeLogger(context, executionContext, 'execution_controller');
        const events = context.apis.foundation.getSystemEvents();
        const slicerPort = executionContext.slicer_port;
        const networkLatencyBuffer = _.get(context, 'sysconfig.teraslice.network_latency_buffer');
        const actionTimeout = _.get(context, 'sysconfig.teraslice.action_timeout');
        const workerDisconnectTimeout = _.get(context, 'sysconfig.teraslice.worker_disconnect_timeout');
        const nodeDisconnectTimeout = _.get(context, 'sysconfig.teraslice.node_disconnect_timeout');
        const shutdownTimeout = _.get(context, 'sysconfig.teraslice.shutdown_timeout');

        this.server = new ExecutionControllerServer({
            port: slicerPort,
            networkLatencyBuffer,
            actionTimeout,
            workerDisconnectTimeout,
        });

        const clusterMasterPort = _.get(context, 'sysconfig.teraslice.port');
        const clusterMasterHostname = _.get(context, 'sysconfig.teraslice.master_hostname');

        this.client = new ClusterMasterClient({
            clusterMasterUrl: formatURL(clusterMasterHostname, clusterMasterPort),
            executionContext,
            networkLatencyBuffer,
            actionTimeout,
            connectTimeout: nodeDisconnectTimeout,
            exId: executionContext.ex_id,
        });

        this.executionAnalytics = new ExecutionAnalytics(
            context,
            executionContext,
            this.client
        );

        this.scheduler = new Scheduler(context, executionContext);

        this.exId = _.get(executionContext, 'config.ex_id');
        this.workerId = workerId;
        this.logger = logger;
        this.events = events;
        this.context = context;
        this.executionContext = executionContext;
        this.collectAnalytics = this.executionContext.config.analytics;
        this.shutdownTimeout = shutdownTimeout;
        this.workerDisconnectTimeout = workerDisconnectTimeout;
        this.stores = {};

        this.isPaused = false;
        this.isShutdown = false;
        this.isShuttingDown = false;
        this.isInitialized = false;
        this.isStarted = false;
        this.pendingDispatches = 0;
        this.pendingSlices = 0;
        this.isDoneProcessing = false;
        this.isExecutionFinished = false;
        this.isExecutionDone = false;
        this.workersHaveConnected = false;
        this._handlers = {};

        this.setFailingStatus = this.setFailingStatus.bind(this);
        this.terminalError = this.terminalError.bind(this);
    }

    async initialize() {
        const { context } = this;

        const exStore = makeExStore(context);
        this.stores.exStore = await exStore;

        await this.client.start();

        let verified;
        let verifiedErr;
        try {
            verified = await this._verifyExecution();
        } catch (err) {
            verifiedErr = err;
        }
        if (!verified) {
            this.isShutdown = true;
            await this.stores.exStore.shutdown(true);
            await this.client.shutdown();
            if (verifiedErr) {
                throw verifiedErr;
            }
            return;
        }

        const stateStore = makeStateStore(context);
        this.stores.stateStore = await stateStore;

        await this.server.start();

        this.isInitialized = true;

        this.server.onClientOnline((workerId) => {
            this.logger.trace(`worker ${workerId} is online`);
            this._adjustSlicerQueueLength();
            this.workersHaveConnected = true;
            clearTimeout(this.workerConnectTimeoutId);
            this.workerConnectTimeoutId = null;
            this.executionAnalytics.increment('workers_joined');
        });

        this.server.onClientAvailable((workerId) => {
            this.logger.trace(`worker ${workerId} is available`);
            this.executionAnalytics.set('workers_available', this.server.availableClientCount);
            this.executionAnalytics.set('workers_active', this.server.activeWorkers.length);
        });

        this.server.onClientUnavailable(() => {
            this.executionAnalytics.set('workers_active', this.server.activeWorkers.length);
            this.executionAnalytics.set('workers_available', this.server.availableClientCount);
        });

        this.server.onClientDisconnect((workerId) => {
            this.logger.trace(`worker ${workerId} is disconnected but it may reconnect`);
            this.executionAnalytics.increment('workers_disconnected');
            this.executionAnalytics.set('workers_active', this.server.activeWorkers.length);

            this._startWorkerDisconnectWatchDog();
        });

        this.server.onClientOffline((workerId) => {
            this.logger.trace(`worker ${workerId} is offline`);
            this._adjustSlicerQueueLength();
        });

        this.server.onClientReconnect((workerId) => {
            clearTimeout(this.workerDisconnectTimeoutId);
            this.workerConnectTimeoutId = null;

            this.logger.trace(`worker ${workerId} is reconnected`);
            this.executionAnalytics.increment('workers_reconnected');
        });

        this.client.onExecutionPause(() => this.pause());
        this.client.onExecutionResume(() => this.resume());
        this.client.onServerShutdown(() => {
            this.logger.warn('Cluster Master shutdown, exiting...');
            this.executionAnalytics.sendingAnalytics = false;
            this._endExecution();
        });

        this.server.onSliceSuccess((workerId, response) => {
            this.logger.info(`worker ${workerId} has completed its slice ${response.slice_id}`);
            this.events.emit('slice:success', response);
            this.pendingSlices -= 1;
        });

        this.server.onSliceFailure((workerId, response) => {
            this.logger.error(`worker: ${workerId} has failure completing its slice`, response);
            this.events.emit('slice:failure', response);

            if (this.scheduler.canComplete()) {
                this.setFailingStatus();
            } else if (this.scheduler.isRecovering()) {
                this.terminalError(new Error('Slice failed while recovering'));
            } else {
                // in persistent mode we set watchdogs to monitor
                // when failing can be set back to running
                this._checkAndUpdateExecutionState();
            }

            this.pendingSlices -= 1;
        });

        this._handlers['slicer:execution:update'] = ({ update }) => {
            this.logger.trace('slicer sending a execution update', update);

            // this is updating the opConfig for elasticsearch start and/or end dates for ex,
            // this assumes elasticsearch is first
            this.stores.exStore.update(this.exId, { operations: update });
        };

        this._handlers['slicers:finished'] = (err) => {
            if (err) {
                this.logger.error('slicers finished due to failure', err);
                this.terminalError(err);
            }
        };

        _.forEach(this._handlers, (handler, event) => {
            this.events.on(event, handler);
        });

        this.logger.debug(`execution ${this.exId} is initialized`);

        this.isInitialized = true;
    }

    async run() {
        if (this.isShuttingDown) {
            this.logger.error('Cannot run execution while shutting down');
            return;
        }

        if (!this.isInitialized) {
            this.logger.error('Cannot run execution is not initialized');
            return;
        }

        try {
            await this._runExecution();
        } catch (err) {
            this.logger.error('Run execution error', err);
        }

        this.events.emit('worker:shutdown');

        // help the workers go offline
        this.server.isShuttingDown = true;

        await this._finishExecution();

        try {
            await Promise.all([
                this.client.sendExecutionFinished(),
                this._waitForWorkersToExit(),
            ]);
        } catch (err) {
            this.logger.error('Failure sending execution finished', err);
        }

        this.logger.debug(`execution ${this.exId} is done`);
    }

    async resume() {
        if (!this.isPaused) return;

        this.logger.info(`execution ${this.exId} is resuming...`);

        this.isPaused = false;
        this.scheduler.start();

        await Promise.delay(100);
    }

    async pause() {
        if (this.isPaused) return;

        this.logger.info(`execution ${this.exId} is pausing...`);

        this.isPaused = true;
        this.scheduler.pause();

        await Promise.delay(100);
    }

    async setFailingStatus() {
        const { exStore } = this.stores;

        const errMsg = `slicer: ${this.exId} has encountered a processing_error`;
        this.logger.error(errMsg);

        const executionStats = this.executionAnalytics.getAnalytics();
        const errorMeta = await exStore.executionMetaData(executionStats, errMsg);
        await exStore.setStatus(this.exId, 'failing', errorMeta);
    }

    async terminalError(err) {
        if (this.isExecutionDone) return;

        const { exStore } = this.stores;

        this.slicerFailed = true;

        const msg = `slicer for ex ${this.exId} had an error, shutting down execution`;
        this.logger.error(msg, err);

        const errMsg = `${msg}, caused by ${err.stack ? err.stack : _.toString(err)}`;

        const executionStats = this.executionAnalytics.getAnalytics();
        const errorMeta = await exStore.executionMetaData(executionStats, errMsg);

        await exStore.setStatus(this.exId, 'failed', errorMeta);

        this.logger.fatal(`execution ${this.exId} is done because of slice failure`);
        this._endExecution();
    }

    async shutdown(block = true) {
        if (this.isShutdown) return;
        if (!this.isInitialized) return;

        if (this.isShuttingDown && block) {
            this.logger.debug(`execution shutdown was called for ex ${this.exId} but it was already shutting down, will block until done`);
            await waitForWorkerShutdown(this.context, 'worker:shutdown:complete');
            return;
        }

        this.logger.debug(`execution shutdown was called for ex ${this.exId}`);

        const shutdownErrs = [];

        // allow clients to go immediately from disconnect to offline
        this.server.isShuttingDown = true;

        // tell the scheduler to stop producing slices
        this.scheduler.stop();

        // remove any listeners
        _.forEach(this._handlers, (handler, event) => {
            this.events.removeListener(event, handler);
        });

        this.isShuttingDown = true;
        this.isPaused = false;

        clearInterval(this.watcher);
        clearTimeout(this.workerConnectTimeoutId);
        clearTimeout(this.workerDisconnectTimeoutId);

        await this._waitForExecutionFinished();

        try {
            await this.executionAnalytics.shutdown();
        } catch (err) {
            this.logger.error('execution analytics error');
        }

        this.scheduler.cleanup();

        if (this.recover) {
            try {
                await this.recover.shutdown();
            } catch (err) {
                shutdownErrs.push(err);
            }
        }

        try {
            await this.server.shutdown();
        } catch (err) {
            shutdownErrs.push(err);
        }

        try {
            await this.client.shutdown();
        } catch (err) {
            shutdownErrs.push(err);
        }

        try {
            await Promise.map(Object.values(this.stores), store => store.shutdown(true));
        } catch (err) {
            shutdownErrs.push(err);
        }

        this.logger.warn(`execution controller ${this.exId} is shutdown`);
        this.isShutdown = true;

        if (shutdownErrs.length) {
            const errMsg = shutdownErrs.map(e => e.stack).join(', and');
            const shutdownErr = new Error(`Failed to shutdown correctly: ${errMsg}`);
            this.events.emit(this.context, 'worker:shutdown:complete', shutdownErr);
            await Promise.delay(0);
            throw shutdownErr;
        }

        this.events.emit(this.context, 'worker:shutdown:complete');
    }

    async _adjustSlicerQueueLength() {
        const { dynamicQueueLength, queueLength } = this.executionContext;
        if (!dynamicQueueLength) return;

        if (this.server.onlineClientCount > queueLength) {
            this.executionContext.queueLength = this.server.onlineClientCount;
            this.logger.info(`adjusted queue length ${this.executionContext.queueLength}`);
        }
    }

    async _runExecution() {
        this._startWorkConnectWatchDog();

        this.slicerAnalytics = makeSliceAnalytics(this.context, this.executionContext);

        this.executionAnalytics.start();

        this.logger.info(`starting execution ${this.exId}...`);
        this.startTime = Date.now();

        // attach to scheduler
        this.scheduler.ensureSliceState = this.ensureSliceState.bind(this);

        this.isStarted = true;

        if (this.scheduler.recoverExecution) {
            await this._recoverSlicesInit();
        } else {
            await this._slicerInit();
        }

        // wait for paused
        await pWhilst(() => this.isPaused && !this.isShuttdown, () => Promise.delay(100));

        await Promise.all([
            this.stores.exStore.setStatus(this.exId, 'running'),
            this.client.sendAvailable(),
            this._waitForRecovery(),
            this._runDispatch(),
            this.scheduler.run(),
        ]);

        const schedulerSuccessful = this.scheduler.isFinished && this.scheduler.slicersDone;

        await this._waitForPendingSlices();

        if (schedulerSuccessful && this.isDoneDispatching) {
            this.logger.debug(`execution ${this.exId} is done processing slices`);
            this.isDoneProcessing = true;
        } else if (!this.isShuttdown) {
            this.logger.debug(`execution ${this.exId} did not finish`);
        } else {
            this.logger.debug(`execution ${this.exId} is exiting...`);
        }
    }

    // dispatching should be pushed out into its own module
    // this function uses many tiny unbound functions to
    // increase performance
    async _runDispatch() {
        const {
            logger,
            scheduler,
            server,
        } = this;

        this.isDoneDispatching = false;

        logger.debug('dispatching slices...');

        // returns a boolean to indicate whether
        // dispatching should continue
        const isRunning = () => {
            if (this.isShuttingDown) return false;
            if (this.isExecutionDone) return false;
            if (scheduler.isFinished
                && !this.pendingDispatches) return false;
            return true;
        };

        const isPaused = () => this.isPaused;
        const dispatchSlice = (slice, workerId) => this._dispatchSlice(slice, workerId);

        // this isn't really ideal since we adding
        // to the beginning of the queue and
        // it may end up in a recursive loop trying
        // to process that slice
        function reenqueueSlice(slice) {
            return scheduler.enqueueSlice(slice, true);
        }

        function dequeueAndDispatch() {
            const slices = scheduler.getSlices(server.workerQueueSize);
            if (slices.length === 0) return;

            // eslint-disable-next-line no-restricted-syntax
            for (const slice of slices) {
                const workerId = server.dequeueWorker(slice);
                if (!workerId) {
                    process.nextTick(() => {
                        reenqueueSlice(slice);
                    });
                    continue;
                }

                dispatchSlice(slice, workerId);
            }
        }

        await new Promise((resolve) => {
            const intervalId = setInterval(dispatch, 5);

            function dispatch() {
                if (!isRunning()) {
                    clearInterval(intervalId);
                    resolve();
                    return;
                }

                if (isPaused()) return;

                const workers = server.workerQueueSize;
                const slices = scheduler.queueLength;

                if (workers > 0 && slices > 0) {
                    dequeueAndDispatch();
                }
            }
{'fraction_non_alphanumeric': 0.11113306053337096, 'fraction_numerical': 0.0010441653732185692, 'mean_word_length': 2.127073773385104, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 57, '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': '5215538', 'n_tokens_mistral': 9571, 'n_tokens_neox': 9168, 'n_words': 1903}