nl_tokens: designed to allocate additional workers not any future slicers
pl_tokens: 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 ( <mask> , 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 ) ; }