Javascript - How To Control How Many Promises Access Network In Parallel

in my application, I have and array of promises that access network in parallel, but some times, when my app is running full speed, my network slows down, due to many promises acc

Solution 1:

You can use Bluebird's .map() which has a concurrency option to control how many requests are in-flight at the same time:

constPromise = require('bluebird');
const http = Promise.promisifyAll(require('http');

var ids = {1: 1, 2: 2, 3: 3, 4: 4, 5: 5, 6: 6, 6: 56, 7: 7, 8: 8, 5:6 }; // this is random Promise.map(Object.keys(ids).map(function(dp){
    return  http.post({url: addr, form: { data: dp }).then(function(body) {
        return body.xx;
    });
}), {concurrency: 2}).then(function(results) {
    // process results here
});

FYI, I didn't understand what you were trying to do with your second http.post() because you where referencing data.x when data is an array. I think the code is a bit too much pseudo-code to illustrate what you were really trying to do with that second http.post().

---

Otherwise, you can code your own concurrency control where you fire up N requests initially and then each time one finishes, you fire up another until you have no more to do. Here's an example of coding the concurrency control manually:

Fire off 1,000,000 requests 100 at a time

Or, you could write it yourself like this:

const http = require('http');

functionhttpPost(options) {
    returnnewPromise(function(resolve, reject) {
        http.post(options, function(err, res, body) {
            if (err) {
                reject(err);
            } else {
                resolve(body);
            }
        });
    });
}

// takes an array of items and a function that returns a promisefunctionmapConcurrent(items, maxConcurrent, fn) {
    let index = 0;
    let inFlightCntr = 0;
    let doneCntr = 0;
    let results = newArray(items.length);
    let stop = false;

    returnnewPromise(function(resolve, reject) {

        functionrunNext() {
            let i = index;
            ++inFlightCntr;
            fn(items[index], index++).then(function(val) {
                ++doneCntr;
                --inFlightCntr;
                results[i] = val;
                run();
            }, function(err) {
                // set flag so we don't launch any more requests
                stop = true;
                reject(err);
            });
        }

        functionrun() {
            // launch as many as we're allowed towhile (!stop && inflightCntr < maxConcurrent && index < items.length) {
                runNext();
            }
            // if all are done, then resolve parent promise with resultsif (doneCntr === items.length) {
                resolve(results);
            }
        }

        run();
    });
}

var ids = {1: 1, 2: 2, 3: 3, 4: 4, 5: 5, 6: 6, 6: 56, 7: 7, 8: 8, 5:6 }; // this is random mapConcurrent(Object.keys(ids), 2, function(item, index) {
    returnhttpPost({url: addr, form: {data: item}}).then(function(body) {
        return body.xxx;
    });
}).then(function(results) {
    // array of results here
}, function(err) {
    // error here    
});

Solution 2:

This is one way of achieving your aim without using a library. Within the promise returned from makeMaxConcurrencyRequests(), the startNew() function is recursively called, sending new requests until we have been through every id, and without exceeding a current request count of maxConcurrency.

When each request completes, its return data is pushed into the returnedData array. When all requests are completed, the promise is resolved with returnedData.

I haven't tested this, but looking at it my only concern is that startNew() is going to be called multiple times in quick succession while requests are pending. If this causes issues then rather than immediately calling startNew(), we could use setTimeout to delay the next invocation - this is commented out in my code.

functionmakeMaxConcurrencyRequests(ids, maxConcurrency) {
    returnnewPromise(function(resolve, reject) {
        let i = 0, currentlyRunning = 0, returnedData = [];
        functionstartNew() {        
            while (i < ids.length && currentlyRunning <= maxConcurrency) {
                makeRequest(ids[i++]).then(function(data) {
                    returnedData.push(data);
                    currentlyRunning--;
                    startNew();
                }).catch(function(err) {
                    reject(err);
                });
                currentlyRunning++;
            }
            if (i >= ids.length && currentlyRunning === 0) {
                resolve(returnedData);
            }
            startNew();
            // setTimeout(startNew, 200);           
        }
    }
}

functionmakeRequest(id) {
    returnnewPromise(function(resolve, reject){
        http.post({url: addr, form: { data: dp }}, function(err, res, body){
            if (err){
                reject(err)
            } 

            http.post({url: hostAddress, form: { data: body.xx }}, function(err2, res2, body2){
                if(err2) {
                    reject(err2);
                }
                resolve(body2.xx);
           }); 
       });

   });
}

Usage:

var ids = {1: 1, 2: 2, 3: 3, 4: 4, 5: 5, 6: 6, 6: 56, 7: 7, 8: 8, 5:6 };
var maxConcurrency = 3;
makeMaxConcurrencyRequests(Object.keys(ids), maxConcurrency)
.then(function(data) {
    // do something with data
}).catch(function(error) {
    // do something with error
});