packages/teraslice/lib/cluster/services/cluster/backends/native/index.js
function allocateWorkers(execution, numOfWorkersRequested) {
        const exId = execution.ex_id;
        const jobId = execution.job_id;
        const jobStr = JSON.stringify(execution);
        const sortedNodes = _.orderBy(clusterState, 'available', 'desc');
        let workersRequested = numOfWorkersRequested;
        let availWorkers = _availableWorkers(false, true);

        const dispatch = _makeDispatch();

        while (workersRequested > 0 && availWorkers > 0) {
            for (let i = 0; i < sortedNodes.length; i += 1) {
                // each iteration check if it can allocate
                if (workersRequested > 0 && availWorkers > 0) {
                    if (sortedNodes[i].available >= 1) {
                        dispatch.set(sortedNodes[i].node_id, 1);
                        availWorkers -= 1;
                        workersRequested -= 1;
                    }
                } else {
                    break;
                }
            }
        }
        // if left over worker requests, enqueue them, queue works based off of id
        // so it redundantly references ex_id

        const workerData = {
            job: jobStr,
            id: exId,
            ex_id: exId,
            job_id: jobId,
            workers: 1,
            assignment: 'worker'
        };

        while (workersRequested > 0) {
            logger.trace(`adding worker to pending queue for ex: ${exId}`);
            pendingWorkerRequests.enqueue(workerData);
            workersRequested -= 1;
        }
        const results = [];

        _.forOwn(dispatch.getDispatch(), (workerCount, nodeId) => {
            const requestedWorkersData = {
                job: jobStr,
                id: exId,
                ex_id: exId,
                job_id: jobId,
                workers: workerCount,
                assignment: 'worker'
            };

            const createRequest = messaging.send({
                to: 'node_master',
                address: nodeId,
                message: 'cluster:workers:create',
                payload: requestedWorkersData,
                response: true
            }).then((msg) => {
                const createdWorkers = _.get(msg, 'payload.createdWorkers');
                if (!_.isInteger(createdWorkers)) {
                    logger.error(`malformed response from create workers request to node ${nodeId}`, msg);
                    return;
                }
                if (createdWorkers < workerCount) {
                    logger.warn(`node ${nodeId} was only able to allocate ${createdWorkers} the request worker count of ${workerCount}, enqueing the remainder`);
                    const newWorkersRequest = _.cloneDeep(requestedWorkersData);
                    newWorkersRequest.workers = workerCount - createdWorkers;
                    pendingWorkerRequests.enqueue(newWorkersRequest);
                } else {
                    logger.debug(`node ${nodeId} allocated ${createdWorkers}`);
                }
            }).catch((err) => {
                logger.error(err, `An error has occurred in allocating : ${workerCount} workers to node : ${nodeId}, the worker request has been enqueued`);
                pendingWorkerRequests.enqueue(requestedWorkersData);
            });

            results.push(createRequest);
        });

        // this will resolve successfully if one worker was actually allocated
        return Promise.all(results);
    }
function allocateWorkers(execution, numOfWorkersRequested) {
        const exId = execution.ex_id;
        const jobId = execution.job_id;
        const jobStr = JSON.stringify(execution);
        const sortedNodes = _.orderBy(clusterState, 'available', 'desc');
        let workersRequested = numOfWorkersRequested;
        let availWorkers = _availableWorkers(false, true);

        const dispatch = _makeDispatch();

        while (workersRequested > 0 && availWorkers > 0) {
            for (let i = 0; i < sortedNodes.length; i += 1) {
                // each iteration check if it can allocate
                if (workersRequested > 0 && availWorkers > 0) {
                    if (sortedNodes[i].available >= 1) {
                        dispatch.set(sortedNodes[i].node_id, 1);
                        availWorkers -= 1;
                        workersRequested -= 1;
                    }
                } else {
                    break;
                }
            }
        }
        // if left over worker requests, enqueue them, queue works based off of id
        // so it redundantly references ex_id

        const workerData = {
            job: jobStr,
            id: exId,
            ex_id: exId,
            job_id: jobId,
            workers: 1,
            assignment: 'worker'
        };

        while (workersRequested > 0) {
            logger.trace(`adding worker to pending queue for ex: ${exId}`);
            pendingWorkerRequests.enqueue(workerData);
            workersRequested -= 1;
        }
        const results = [];

        _.forOwn(dispatch.getDispatch(), (workerCount, nodeId) => {
            const requestedWorkersData = {
                job: jobStr,
                id: exId,
                ex_id: exId,
                job_id: jobId,
                workers: workerCount,
                assignment: 'worker'
            };

            const createRequest = messaging.send({
                to: 'node_master',
                address: nodeId,
                message: 'cluster:workers:create',
                payload: requestedWorkersData,
                response: true
            }).then((msg) => {
                const createdWorkers = _.get(msg, 'payload.createdWorkers');
                if (!_.isInteger(createdWorkers)) {
                    logger.error(`malformed response from create workers request to node ${nodeId}`, msg);
                    return;
                }
                if (createdWorkers < workerCount) {
                    logger.warn(`node ${nodeId} was only able to allocate ${createdWorkers} the request worker count of ${workerCount}, enqueing the remainder`);
                    const newWorkersRequest = _.cloneDeep(requestedWorkersData);
                    newWorkersRequest.workers = workerCount - createdWorkers;
                    pendingWorkerRequests.enqueue(newWorkersRequest);
                } else {
                    logger.debug(`node ${nodeId} allocated ${createdWorkers}`);
                }
            }).catch((err) => {
                logger.error(err, `An error has occurred in allocating : ${workerCount} workers to node : ${nodeId}, the worker request has been enqueued`);
                pendingWorkerRequests.enqueue(requestedWorkersData);
            });

            results.push(createRequest);
        });

        // this will resolve successfully if one worker was actually allocated
        return Promise.all(results);
    }
function allocateWorkers ( execution , numOfWorkersRequested ) { const exId = execution . ex_id ; const jobId = execution . job_id ; const jobStr = JSON . stringify ( execution ) ; const sortedNodes = _ . orderBy ( clusterState , 'available' , 'desc' ) ; let workersRequested = numOfWorkersRequested ; let availWorkers = _availableWorkers ( false , true ) ; const dispatch = _makeDispatch ( ) ; while ( workersRequested > 0 && availWorkers > 0 ) { for ( let i = 0 ; i < sortedNodes . length ; i += 1 ) { // each iteration check if it can allocate if ( workersRequested > 0 && availWorkers > 0 ) { if ( sortedNodes [ i ] . available >= 1 ) { dispatch . set ( sortedNodes [ i ] . node_id , 1 ) ; availWorkers -= 1 ; workersRequested -= 1 ; } } else { break ; } } } // if left over worker requests, enqueue them, queue works based off of id // so it redundantly references ex_id const workerData = { job : jobStr , id : exId , ex_id : exId , job_id : jobId , workers : 1 , assignment : 'worker' } ; while ( workersRequested > 0 ) { logger . trace ( ` ${ exId } ` ) ; pendingWorkerRequests . enqueue ( workerData ) ; workersRequested -= 1 ; } const results = [ ] ; _ . forOwn ( dispatch . getDispatch ( ) , ( workerCount , nodeId ) => { const requestedWorkersData = { job : jobStr , id : exId , ex_id : exId , job_id : jobId , workers : workerCount , assignment : 'worker' } ; const createRequest = messaging . send ( { to : 'node_master' , address : nodeId , message : 'cluster:workers:create' , payload : requestedWorkersData , response : true } ) . then ( ( msg ) => { const createdWorkers = _ . get ( msg , 'payload.createdWorkers' ) ; if ( ! _ . isInteger ( createdWorkers ) ) { logger . error ( ` ${ nodeId } ` , msg ) ; return ; } if ( createdWorkers < workerCount ) { logger . warn ( ` ${ nodeId } ${ createdWorkers } ${ workerCount } ` ) ; const newWorkersRequest = _ . cloneDeep ( requestedWorkersData ) ; newWorkersRequest . workers = workerCount - createdWorkers ; pendingWorkerRequests . enqueue ( newWorkersRequest ) ; } else { logger . debug ( ` ${ nodeId } ${ createdWorkers } ` ) ; } } ) . catch ( ( err ) => { logger . error ( err , ` ${ workerCount } ${ nodeId } ` ) ; pendingWorkerRequests . enqueue ( requestedWorkersData ) ; } ) ; results . push ( createRequest ) ; } ) ; // this will resolve successfully if one worker was actually allocated return Promise . all ( results ) ; }
designed to allocate additional workers, not any future slicers
designed to allocate additional workers not any future slicers