import { ArrayPromise , PromiseBase , PromiseCollection , TSDNPromise } from "./Promise" ; import Type from "typescript-dotnet-core/Types" ; import Worker from "typescript-dotnet-threading/Worker" ; import { WorkerLike } from "typescript-dotnet-threading/WorkerType" ; import deferImmediate from "typescript-dotnet-threading/deferImmediate" ; import { isNodeJS } from "typescript-dotnet-core/Environment" ; import ObjectPool from "typescript-dotnet-core/Disposable/ObjectPool" ; import IMap from "typescript-dotnet-core/IMap" ; declare const navigator ; declare const __dirname ; const MAX_WORKERS = 16 , VOID0 = void 0 , URL = typeof self !== Type . UNDEFINED ? ( self . URL ? self . URL : ( < any > self ) . webkitURL ) : null , _supports = isNodeJS || ! ! ( < any > self ) . Worker ; export interface ParallelOptions { /**
     * This is the path to the file eval.js.  This is required when running in node, and required for some browsers (IE 10) in order to work around cross-domain restrictions for web workers.  Defaults to the same location as parallel.js in node environments, and null in the browser.
     **/ evalPath ? : string ; /**
     * The maximum number of permitted worker threads.  This will default to 4, or the number of CPUs on your computer if you're running node.
     **/ maxConcurrency ? : number ; /**
     * If WebWorkers are not available, whether or not to fall back to synchronous processing using setTimeout.  Defaults to true.
     **/ allowSynchronous ? : boolean ; env ? : any ; envNamespace ? : string ; } const defaults = { evalPath : isNodeJS ? __dirname + '/eval.js' : VOID0 , maxConcurrency : isNodeJS ? ( < any > require ) ( 'os' ) . cpus ( ) . length : ( navigator . hardwareConcurrency || 4 ) , allowSynchronous : true , env : { } , envNamespace : 'env' } ; function extend < TFrom extends any , TTo extends any > ( from , to ) { if ( ! to ) to = < any > { } ; for ( let i of Object . keys ( from ) ) { if ( to [ i ] === VOID0 ) to [ i ] = from [ i ] ; } return < any > to ; } function interact ( w , onMessage , onError , message ? ) { if ( onMessage ) w . onmessage = onMessage ; if ( onError ) w . onerror = onError ; if ( message !== VOID0 ) w . postMessage ( message ) ; } class WorkerPromise < T > extends TSDNPromise < T > { constructor ( worker , data ) { super ( ( resolve , reject ) => { interact ( worker , ( response ) => { resolve ( response . data ) ; } , ( e ) => { reject ( e ) ; } , data ) ; } , true ) ; } } export type RequireType = string | Function | { name ? : string ; fn : Function ; } ; module workers { function getPool ( key ) { let pool = workerPools [ key ] ; if ( ! pool ) { workerPools [ key ] = pool = new ObjectPool < WorkerLike > ( undefined , undefined , 8 ) ; } return pool ; } const workerPools = { } ; export function recycle ( w ) { if ( w ) { w . onerror = < any > null ; w . onmessage = < any > null ; const k = ( < any > w ) . __key ; if ( k ) { getPool ( k ) . give ( w ) ; } else { deferImmediate ( ( ) => w . terminate ( ) ) ; } } return null ; } export function tryGet ( key ) { return getPool ( key ) . tryTake ( ) ; } export function getNew ( key , url ) { const worker = new Worker ( url ) ; worker . __key = key ; if ( ! worker . dispose ) { worker . dispose = ( ) => { worker . onmessage = null ; worker . onerror = null ; worker . dispose = null ; worker . terminate ( ) ; } ; } return worker ; } } export class Parallel { options : ParallelOptions ; _requiredScripts : string [ ] ; _requiredFunctions : { name ? : string ; fn : Function ; } [ ] ; constructor ( options ? ) { this . options = extend ( defaults , options ) ; this . _requiredScripts = [ ] ; this . _requiredFunctions = [ ] ; this . ensureClampedMaxConcurrency ( ) ; } static maxConcurrency ( max ) { return new Parallel ( { maxConcurrency : max } ) ; } protected _getWorkerSource ( task , env ? ) { const scripts = this . _requiredScripts , functions = this . _requiredFunctions ; let preStr = '' ; if ( ! isNodeJS && scripts . length ) { preStr += 'importScripts("' + scripts . join ( '","' ) + '");\r\n' ; } for ( let { name , fn } of functions ) { const source = fn . toString ( ) ; preStr += name ? `var ${ name } = ${ source };` : source ; } env = JSON . stringify ( env || { } ) ; const ns = this . options . envNamespace ; return preStr + ( isNodeJS ? `process.on("message", function(e) {global.${ ns } = ${ env };process.send(JSON.stringify((${ task . toString ( ) })(JSON.parse(e).data)))})` : `self.onmessage = function(e) {var global = {}; global.${ ns } = ${ env };self.postMessage((${ task . toString ( ) })(e.data))}` ) ; } require ( ... required ) { return this . requireThese ( required ) ; } requireThese ( required ) { for ( let a of required ) { switch ( typeof a ) { case Type . STRING : this . _requiredScripts . push ( < string > a ) ; break ; case Type . FUNCTION : this . _requiredFunctions . push ( { fn : < Function > a } ) ; break ; case Type . OBJECT : this . _requiredFunctions . push ( < { name ? : string ; fn : Function ; } > a ) ; break ; default : throw new TypeError ( "Invalid type." ) ; } } return this ; } protected _spawnWorker ( task , env ? ) { const src = this . _getWorkerSource ( task , env ) ; if ( Worker === VOID0 ) return VOID0 ; let worker = workers . tryGet ( src ) ; if ( worker ) return worker ; const scripts = this . _requiredScripts ; let evalPath = this . options . evalPath ; if ( ! evalPath ) { if ( isNodeJS ) throw new Error ( "Can't use NodeJS without eval.js!" ) ; if ( scripts . length ) throw new Error ( "Can't use required scripts without eval.js!" ) ; if ( ! URL ) throw new Error ( "Can't create a blob URL in this browser!" ) ; } if ( isNodeJS || scripts . length || ! URL ) { worker = workers . getNew ( src , < string > evalPath ) ; worker . postMessage ( src ) ; } else if ( URL ) { const blob = new Blob ( [ src ] , { type : 'text/javascript' } ) ; const url = URL . createObjectURL ( blob ) ; worker = workers . getNew ( src , url ) ; } return worker ; } param data param task param env returns { TSDNPromise < U > | TSDNPromise } startNew < T , U > ( data , task , env ? ) { const _ = this ; const maxConcurrency = this . ensureClampedMaxConcurrency ( ) ; const worker = maxConcurrency ? _ . _spawnWorker ( task , extend ( _ . options . env , env || { } ) ) : null ; if ( worker ) { return new WorkerPromise < U > ( worker , data ) . finallyThis ( ( ) => workers . recycle ( worker ) ) ; } if ( _ . options . allowSynchronous ) return this . startLocal ( data , task ) ; throw new Error ( maxConcurrency ? "Workers do not exist and synchronous operation not allowed!" : "'maxConcurrency' set to 0 but 'allowSynchronous' is false." ) ; } param data param task returns { TSDNPromise < U > | TSDNPromise } startLocal < T , U > ( data , task ) { return new TSDNPromise < U > ( ( resolve , reject ) => { try { resolve ( task ( data ) ) ; } catch ( e ) { reject ( e ) ; } } ) ; } param data param task param env returns { PromiseCollection } pipe < T , U > ( data , task , env ? ) { let result ; if ( data && data . length ) { const len = data . length ; const taskString = task . toString ( ) ; const maxConcurrency = this . ensureClampedMaxConcurrency ( ) ; let error ; let i = 0 ; for ( let w = 0 ; ! error && i < Math . min ( len , maxConcurrency ) ; w ++ ) { let worker = maxConcurrency ? this . _spawnWorker ( taskString , env ) : null ; if ( ! worker ) { if ( ! this . options . allowSynchronous ) throw new Error ( maxConcurrency ? "Workers do not exist and synchronous operation not allowed!" : "'maxConcurrency' set to 0 but 'allowSynchronous' is false." ) ; return TSDNPromise . map ( data , task ) ; } if ( ! result ) { result = data . map ( d => new TSDNPromise < U > ( ) ) ; } let next = ( ) => { if ( error ) { worker = workers . recycle ( worker ) ; } if ( worker ) { if ( i < len ) { let ii = i ++ , p = result ! [ ii ] ; let wp = new WorkerPromise < U > ( worker , data [ ii ] ) ; wp . thenSynchronous ( r => { p . resolve ( r ) ; next ( ) ; } , e => { if ( ! error ) { error = e ; p . reject ( e ) ; worker = workers . recycle ( worker ) ; } } ) . finallyThis ( ( ) => wp . dispose ( ) ) ; } else { worker = workers . recycle ( worker ) ; } } } ; next ( ) ; } } return new PromiseCollection < U > ( result ) ; } private ensureClampedMaxConcurrency ( ) { let { maxConcurrency } = this . options ; if ( maxConcurrency && maxConcurrency > MAX_WORKERS ) { this . options . maxConcurrency = maxConcurrency = MAX_WORKERS ; console . warn ( `More than ${ MAX_WORKERS } workers can reach worker limits and cause unexpected results.  maxConcurrency reduced to ${ MAX_WORKERS }.` ) ; } return ( maxConcurrency || maxConcurrency === 0 ) ? maxConcurrency : MAX_WORKERS ; } param data param task param env returns { ArrayPromise } map < T , U > ( data , task , env ? ) { if ( ! data || ! data . length ) return ArrayPromise . fulfilled < U > ( [ ] ) ; data = data . slice ( ) ; return new ArrayPromise < U > ( ( resolve , reject ) => { const result = [ ] , len = data . length ; result . length = len ; const taskString = task . toString ( ) ; let maxConcurrency = this . ensureClampedMaxConcurrency ( ) , error ; let i = 0 , resolved = 0 ; for ( let w = 0 ; ! error && i < Math . min ( len , maxConcurrency ) ; w ++ ) { let worker = this . _spawnWorker ( taskString , env ) ; if ( ! worker ) { if ( ! this . options . allowSynchronous ) throw new Error ( 'Workers do not exist and synchronous operation not allowed!' ) ; resolve ( TSDNPromise . map ( data , task ) . all ( ) ) ; return ; } let next = ( ) => { if ( error ) { worker = workers . recycle ( worker ) ; } if ( worker ) { if ( i < len ) { let ii = i ++ ; let wp = new WorkerPromise < U > ( worker , data [ ii ] ) ; wp . thenSynchronous ( r => { result [ ii ] = < any > r ; next ( ) ; } , e => { if ( ! error ) { error = e ; reject ( e ) ; worker = workers . recycle ( worker ) ; } } ) . thenThis ( ( ) => { resolved ++ ; if ( resolved > len ) throw Error ( "Resolved count exceeds data length." ) ; if ( resolved === len ) resolve ( result ) ; } ) . finallyThis ( ( ) => wp . dispose ( ) ) ; } else { worker = workers . recycle ( worker ) ; } } } ; next ( ) ; } } ) ; } static get isSupported ( ) { return _supports ; } static options ( options ? ) { return new Parallel ( options ) ; } static require ( ... required ) { return ( new Parallel ( ) ) . requireThese ( required ) ; } static requireThese ( required ) { return ( new Parallel ( ) ) . requireThese ( required ) ; } static startNew < T , U > ( data , task , env ? ) { return ( new Parallel ( ) ) . startNew ( data , task , env ) ; } static map < T , U > ( data , task , env ? ) { return ( new Parallel ( ) ) . map ( data , task , env ) ; } } export default Parallel ;