parallel

rzcoder · rzcoder · commit 995e18ad002d · 2016-11-03T01:08:30.000+05:00
diff --git a/lib/functions/parallel.d.ts b/lib/functions/parallel.d.ts
@@ -0,0 +1,3 @@
+import { Tasks } from "../types";
+export declare function parallel(tasks: Tasks, maxThreads?: number): Promise<any>;
+export declare function queue(tasks: Tasks): Promise<any>;
diff --git a/lib/functions/parallel.js b/lib/functions/parallel.js
@@ -1,13 +1,46 @@
-/*import {
-    Tasks
-} from "../types";
-
-export function parallel(tasks: Tasks, maxThreads: number = tasks.length): Promise<any> {
+"use strict";
+function promiseHandler(index, data) {
+    this.results[index] = data;
+    this.processes--;
+    if (this.processes === 0 && index === this.tasks.length - 1) {
+        this.resolve(this.results);
+    }
+    else {
+        execute(this);
+    }
+}
+function execute(scope) {
+    if (scope.processes < scope.maxThreads && scope.pointer < scope.tasks.length) {
+        var handler = promiseHandler.bind(scope, scope.pointer);
+        scope.tasks[scope.pointer]().then(handler, handler);
+        scope.pointer++;
+        scope.processes++;
+    }
+}
+function parallel(tasks, maxThreads) {
+    if (maxThreads === void 0) { maxThreads = tasks.length; }
     if (tasks.length === 0) {
         return Promise.resolve();
     }
-
+    var scope = {
+        processes: 0,
+        pointer: 0,
+        tasks: tasks,
+        maxThreads: maxThreads,
+        results: [],
+        resolve: null,
+        reject: null
+    };
     return new Promise(function (resolve, reject) {
-
+        for (var i = 0; i < maxThreads && i < tasks.length; i++) {
+            scope.resolve = resolve;
+            scope.reject = reject;
+            execute(scope);
+        }
     });
-}*/
+}
+exports.parallel = parallel;
+function queue(tasks) {
+    return parallel(tasks, 1);
+}
+exports.queue = queue;
diff --git a/lib/prow.d.ts b/lib/prow.d.ts
@@ -4,3 +4,4 @@ export { timeout } from "./functions/timeout";
 export { waterfall } from "./functions/waterfall";
 export { retry } from "./functions/retry";
 export { times } from "./functions/times";
+export { parallel, queue } from "./functions/parallel";
diff --git a/lib/prow.js b/lib/prow.js
@@ -11,3 +11,6 @@ var retry_1 = require("./functions/retry");
 exports.retry = retry_1.retry;
 var times_1 = require("./functions/times");
 exports.times = times_1.times;
+var parallel_1 = require("./functions/parallel");
+exports.parallel = parallel_1.parallel;
+exports.queue = parallel_1.queue;
diff --git a/src/functions/parallel.ts b/src/functions/parallel.ts
@@ -1,13 +1,60 @@
-/*import {
+import {
     Tasks
 } from "../types";
 
+interface IScope {
+    processes: number;
+    pointer: number;
+    tasks: Tasks;
+    maxThreads: number;
+    results: any[];
+    resolve: any;
+    reject: any;
+}
+
+function promiseHandler (index: number, data: any) {
+    this.results[index] = data;
+    this.processes--;
+    if (this.processes === 0 && index === this.tasks.length - 1) {
+        this.resolve(this.results);
+    } else {
+        execute(this);
+    }
+}
+
+function execute(scope: IScope) {
+    if (scope.processes < scope.maxThreads && scope.pointer < scope.tasks.length) {
+        const handler = promiseHandler.bind(scope, scope.pointer);
+        scope.tasks[scope.pointer]().then(handler, handler);
+        scope.pointer++;
+        scope.processes++;
+    }
+}
+
 export function parallel(tasks: Tasks, maxThreads: number = tasks.length): Promise<any> {
     if (tasks.length === 0) {
         return Promise.resolve();
     }
 
-    return new Promise(function (resolve, reject) {
+    const scope: IScope = {
+        processes: 0,
+        pointer: 0,
+        tasks: tasks,
+        maxThreads: maxThreads,
+        results: [],
+        resolve: null,
+        reject: null
+    };
 
+    return new Promise(function (resolve, reject) {
+        for (let i = 0; i < maxThreads && i < tasks.length; i++) {
+            scope.resolve = resolve;
+            scope.reject = reject;
+            execute(scope);
+        }
     });
-}*/
+}
+
+export function queue(tasks: Tasks): Promise<any> {
+    return parallel(tasks, 1);
+}
diff --git a/src/prow.ts b/src/prow.ts
@@ -6,4 +6,5 @@ export {delay} from "./functions/delay";
 export {timeout} from "./functions/timeout";
 export {waterfall} from "./functions/waterfall";
 export {retry} from "./functions/retry";
-export {times} from "./functions/times";
+export {times} from "./functions/times";
+export {parallel, queue} from "./functions/parallel";
diff --git a/test/parallel.js b/test/parallel.js
@@ -0,0 +1,93 @@
+const _ = require("lodash");
+const chai = require("chai");
+const chaiAsPromised = require("chai-as-promised");
+chai.use(chaiAsPromised);
+const {assert, expect} = chai;
+const prow = require("../lib/prow");
+
+describe("Parallel", function () {
+    it("queue", function () {
+        const tasks = [];
+        let counter = 0;
+        for (let i = 0; i < 10; i++) {
+            tasks.push(() => prow.delay(1).then(() => counter++));
+        }
+
+        return assert.becomes(prow.queue(tasks), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+    });
+
+    it("parallel 1 thread", function () {
+        const tasks = [];
+        let counter = 0;
+        for (let i = 0; i < 10; i++) {
+            tasks.push(() => prow.delay(1).then(() => counter++));
+        }
+
+        return assert.becomes(prow.parallel(tasks, 1), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+    });
+
+    it("parallel 2 threads", function () {
+        const tasks = [];
+        let counter = 0;
+        for (let i = 0; i < 10; i++) {
+            tasks.push(() => {
+                counter++;
+                return Promise.resolve().then(() => counter--)
+            });
+        }
+
+        return assert.becomes(prow.parallel(tasks, 2), [2, 1, 2, 1, 2, 1, 2, 1, 2, 1]);
+    });
+
+    it("parallel 5 threads", function () {
+        const tasks = [];
+        let counter = 0;
+        for (let i = 0; i < 10; i++) {
+            tasks.push(() => {
+                counter++;
+                return Promise.resolve().then(() => counter--)
+            });
+        }
+
+        return assert.becomes(prow.parallel(tasks, 5), [5, 4, 3, 2, 1, 5, 4, 3, 2, 1]);
+    });
+
+    it("parallel 7 threads", function () {
+        const tasks = [];
+        let counter = 0;
+        for (let i = 0; i < 10; i++) {
+            tasks.push(() => {
+                counter++;
+                return Promise.resolve().then(() => counter--)
+            });
+        }
+
+        return assert.becomes(prow.parallel(tasks, 7), [7, 6, 5, 4, 3, 2, 1, 3, 2, 1]);
+    });
+
+    it("parallel 10 threads", function () {
+        const tasks = [];
+        let counter = 0;
+        for (let i = 0; i < 10; i++) {
+            tasks.push(() => {
+                counter++;
+                return Promise.resolve().then(() => counter--)
+            });
+        }
+
+        return assert.becomes(prow.parallel(tasks), [10, 9, 8, 7, 6, 5, 4, 3, 2, 1]);
+    });
+
+    it("parallel 2 threads wit rejects", function () {
+        const tasks = [];
+        let counter = 0;
+        for (let i = 0; i < 10; i++) {
+            tasks.push(() => {
+                counter++;
+                return Promise.reject().catch(() => counter--)
+            });
+        }
+
+        return assert.becomes(prow.parallel(tasks, 2), [2, 1, 2, 1, 2, 1, 2, 1, 2, 1]);
+    });
+});

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+import { Tasks } from "../types";`
	`2`	`+export declare function parallel(tasks: Tasks, maxThreads?: number): Promise<any>;`
	`3`	`+export declare function queue(tasks: Tasks): Promise<any>;`