33
44import { Async , AsyncQueue } from '../Async' ;
55
6+ interface INumberWithWeight {
7+ n : number ;
8+ weight : number ;
9+ }
10+
611describe ( Async . name , ( ) => {
712 describe ( Async . mapAsync . name , ( ) => {
813 it ( 'handles an empty array correctly' , async ( ) => {
@@ -27,13 +32,6 @@ describe(Async.name, () => {
2732 expect ( fn ) . toHaveBeenNthCalledWith ( 3 , 3 , 2 ) ;
2833 } ) ;
2934
30- it ( 'returns the same result as built-in Promise.all' , async ( ) => {
31- const array : number [ ] = [ 1 , 2 , 3 , 4 , 5 , 6 , 7 , 8 ] ;
32- const fn : ( item : number ) => Promise < string > = async ( item ) => `result ${ item } ;
33-
34- expect ( await Async . mapAsync ( array , fn ) ) . toEqual ( await Promise . all ( array . map ( fn ) ) ) ;
35- } ) ;
36-
3735 it ( 'if concurrency is set, ensures no more than N operations occur in parallel' , async ( ) => {
3836 let running : number = 0 ;
3937 let maxRunning : number = 0 ;
@@ -61,6 +59,31 @@ describe(Async.name, () => {
6159 expect ( maxRunning ) . toEqual ( 3 ) ;
6260 } ) ;
6361
62+ it ( 'respects concurrency limit with allowOversubscription=false in mapAsync' , async ( ) => {
63+ const array : INumberWithWeight [ ] = [
64+ { n : 1 , weight : 2 } ,
65+ { n : 2 , weight : 2 }
66+ ] ;
67+
68+ let running = 0 ;
69+ let maxRunning = 0 ;
70+
71+ const result = await Async . mapAsync (
72+ array ,
73+ async ( item ) => {
74+ running ++ ;
75+ maxRunning = Math . max ( maxRunning , running ) ;
76+ await Async . sleepAsync ( 0 ) ;
77+ running -- ;
78+ return `result-${ item . n } ;
79+ } ,
80+ { concurrency : 3 , weighted : true , allowOversubscription : false }
81+ ) ;
82+
83+ expect ( result ) . toEqual ( [ 'result-1' , 'result-2' ] ) ;
84+ expect ( maxRunning ) . toEqual ( 1 ) ;
85+ } ) ;
86+
6487 it ( 'rejects if a sync iterator throws an error' , async ( ) => {
6588 const expectedError : Error = new Error ( 'iterator error' ) ;
6689 let iteratorIndex : number = 0 ;
@@ -314,11 +337,6 @@ describe(Async.name, () => {
314337 ) . rejects . toThrow ( expectedError ) ;
315338 } ) ;
316339
317- interface INumberWithWeight {
318- n : number ;
319- weight : number ;
320- }
321-
322340 it ( 'handles an empty array correctly' , async ( ) => {
323341 let running : number = 0 ;
324342 let maxRunning : number = 0 ;
@@ -469,7 +487,7 @@ describe(Async.name, () => {
469487 running -- ;
470488 } ) ;
471489
472- await Async . forEachAsync ( array , fn , { concurrency : 3 , weighted : true } ) ;
490+ await Async . forEachAsync ( array , fn , { concurrency : 3 , weighted : true , allowOversubscription : true } ) ;
473491 expect ( fn ) . toHaveBeenCalledTimes ( 8 ) ;
474492 expect ( maxRunning ) . toEqual ( 2 ) ;
475493 } ) ;
@@ -542,6 +560,10 @@ describe(Async.name, () => {
542560 } ) ;
543561
544562 describe ( Async . runWithRetriesAsync . name , ( ) => {
563+ afterEach ( ( ) => {
564+ jest . restoreAllMocks ( ) ;
565+ } ) ;
566+
545567 it ( 'Correctly handles a sync function that succeeds the first time' , async ( ) => {
546568 const expectedResult : string = 'RESULT' ;
547569 const result : string = await Async . runWithRetriesAsync ( { action : ( ) => expectedResult , maxRetries : 0 } ) ;
@@ -688,6 +710,142 @@ describe(Async.name, () => {
688710 expect ( sleepSpy ) . toHaveBeenCalledTimes ( 1 ) ;
689711 expect ( sleepSpy ) . toHaveBeenLastCalledWith ( 5 ) ;
690712 } ) ;
713+
714+ describe ( 'allowOversubscription=false operations' , ( ) => {
715+ it . each ( [
716+ {
717+ concurrency : 4 ,
718+ weight : 4 ,
719+ expectedConcurrency : 1 ,
720+ numberOfTasks : 4
721+ } ,
722+ {
723+ concurrency : 4 ,
724+ weight : 1 ,
725+ expectedConcurrency : 4 ,
726+ numberOfTasks : 4
727+ } ,
728+ {
729+ concurrency : 4 ,
730+ weight : 5 ,
731+ expectedConcurrency : 1 ,
732+ numberOfTasks : 2
733+ }
734+ ] ) (
735+ 'enforces strict concurrency limits when allowOversubscription=false: concurrency=$concurrency, weight=$weight, expects max $expectedConcurrency concurrent operations' ,
736+ async ( { concurrency, weight, expectedConcurrency, numberOfTasks } ) => {
737+ let running : number = 0 ;
738+ let maxRunning : number = 0 ;
739+
740+ const array : INumberWithWeight [ ] = Array . from ( { length : numberOfTasks } , ( v , i ) => i ) . map ( ( n ) => ( {
741+ n,
742+ weight
743+ } ) ) ;
744+
745+ const fn : ( item : INumberWithWeight ) => Promise < void > = jest . fn ( async ( ) => {
746+ running ++ ;
747+ await Async . sleepAsync ( 0 ) ;
748+ maxRunning = Math . max ( maxRunning , running ) ;
749+ running -- ;
750+ } ) ;
751+
752+ await Async . forEachAsync ( array , fn , { concurrency, weighted : true , allowOversubscription : false } ) ;
753+ expect ( fn ) . toHaveBeenCalledTimes ( numberOfTasks ) ;
754+ expect ( maxRunning ) . toEqual ( expectedConcurrency ) ;
755+ }
756+ ) ;
757+
758+ it ( 'waits for a small and large operation to finish before scheduling more' , async ( ) => {
759+ let running : number = 0 ;
760+ let maxRunning : number = 0 ;
761+
762+ const array : INumberWithWeight [ ] = [
763+ { n : 1 , weight : 1 } ,
764+ { n : 2 , weight : 10 } ,
765+ { n : 3 , weight : 1 } ,
766+ { n : 4 , weight : 10 } ,
767+ { n : 5 , weight : 1 } ,
768+ { n : 6 , weight : 10 } ,
769+ { n : 7 , weight : 1 } ,
770+ { n : 8 , weight : 10 }
771+ ] ;
772+
773+ const fn : ( item : INumberWithWeight ) => Promise < void > = jest . fn ( async ( item ) => {
774+ running ++ ;
775+ await Async . sleepAsync ( 0 ) ;
776+ maxRunning = Math . max ( maxRunning , running ) ;
777+ running -- ;
778+ } ) ;
779+
780+ await Async . forEachAsync ( array , fn , { concurrency : 3 , weighted : true , allowOversubscription : false } ) ;
781+ expect ( fn ) . toHaveBeenCalledTimes ( 8 ) ;
782+ expect ( maxRunning ) . toEqual ( 1 ) ;
783+ } ) ;
784+
785+ it ( 'handles operation with mixed weights' , async ( ) => {
786+ const concurrency : number = 3 ;
787+ let running : number = 0 ;
788+ let maxRunning : number = 0 ;
789+ const taskToMaxConcurrency : Record < number , number > = { } ;
790+
791+ const array : INumberWithWeight [ ] = [
792+ { n : 1 , weight : 1 } ,
793+ { n : 2 , weight : 2 } ,
794+ { n : 3 , weight : concurrency } ,
795+ { n : 4 , weight : 1 } ,
796+ { n : 5 , weight : 1 }
797+ ] ;
798+
799+ const fn : ( item : INumberWithWeight ) => Promise < void > = jest . fn ( async ( item ) => {
800+ running ++ ;
801+ taskToMaxConcurrency [ item . n ] = running ;
802+ await Async . sleepAsync ( 0 ) ;
803+ maxRunning = Math . max ( maxRunning , running ) ;
804+ running -- ;
805+ } ) ;
806+
807+ await Async . forEachAsync ( array , fn , { concurrency, weighted : true , allowOversubscription : false } ) ;
808+ expect ( fn ) . toHaveBeenCalledTimes ( 5 ) ;
809+ expect ( maxRunning ) . toEqual ( 2 ) ;
810+
811+ expect ( taskToMaxConcurrency [ 1 ] ) . toEqual ( 1 ) ; // task 1
812+ expect ( taskToMaxConcurrency [ 2 ] ) . toEqual ( 2 ) ; // task 1 + 2
813+ expect ( taskToMaxConcurrency [ 3 ] ) . toEqual ( 1 ) ; // task 3
814+ expect ( taskToMaxConcurrency [ 4 ] ) . toEqual ( 1 ) ; // task 4
815+ expect ( taskToMaxConcurrency [ 5 ] ) . toEqual ( 2 ) ; // task 4 + 5
816+ } ) ;
817+
818+ it ( 'allows operations with weight 0 to be picked up when system is at max concurrency' , async ( ) => {
819+ let running : number = 0 ;
820+ let maxRunning : number = 0 ;
821+ const taskToMaxConcurrency : Record < number , number > = { } ;
822+
823+ const array : INumberWithWeight [ ] = [
824+ { n : 1 , weight : 1 } ,
825+ { n : 2 , weight : 0 } ,
826+ { n : 3 , weight : 3 } ,
827+ { n : 4 , weight : 1 }
828+ ] ;
829+
830+ const fn : ( item : INumberWithWeight ) => Promise < void > = jest . fn ( async ( item ) => {
831+ running ++ ;
832+ taskToMaxConcurrency [ item . n ] = running ;
833+ maxRunning = Math . max ( maxRunning , running ) ;
834+ await Async . sleepAsync ( 0 ) ;
835+ running -- ;
836+ } ) ;
837+
838+ await Async . forEachAsync ( array , fn , { concurrency : 3 , weighted : true , allowOversubscription : false } ) ;
839+
840+ expect ( fn ) . toHaveBeenCalledTimes ( 4 ) ;
841+ expect ( maxRunning ) . toEqual ( 2 ) ;
842+
843+ expect ( taskToMaxConcurrency [ 1 ] ) . toEqual ( 1 ) ; // task 1
844+ expect ( taskToMaxConcurrency [ 2 ] ) . toEqual ( 2 ) ; // task 1 + 2
845+ expect ( taskToMaxConcurrency [ 3 ] ) . toEqual ( 2 ) ; // task 2 + 3
846+ expect ( taskToMaxConcurrency [ 4 ] ) . toEqual ( 1 ) ; // task 4
847+ } ) ;
848+ } ) ;
691849 } ) ;
692850} ) ;
693851
0 commit comments