fix: use utils.mapParallel for parallel processing of peers (#166)

Author: kumavis

src/content-fetching/index.js

@@ -109,15 +109,15 @@ module.exports = (dht) => {
       let counterAll = 0
       let counterSuccess = 0
 
-      for await (const peer of dht.getClosestPeers(key, { shallow: true })) {
+      await utils.mapParallel(dht.getClosestPeers(key, { shallow: true }), async (peer) => {
         try {
           counterAll += 1
           await dht._putValueToPeer(key, record, peer)
           counterSuccess += 1
         } catch (err) {
           dht._log.error('Failed to put to peer (%b): %s', peer.id, err)
         }
-      }
+      })
 
       // verify if we were able to put to enough peers
       const minPeers = options.minPeers || counterAll // Ensure we have a default `minPeers`

src/content-routing/index.js

@@ -44,14 +44,14 @@ module.exports = (dht) => {
       }]
 
       // Notify closest peers
-      for await (const peer of dht.getClosestPeers(key.buffer)) {
+      await utils.mapParallel(dht.getClosestPeers(key.buffer), async (peer) => {
         dht._log('putProvider %s to %s', key.toBaseEncodedString(), peer.toB58String())
         try {
           await dht.network.sendMessage(peer, msg)
         } catch (err) {
           errors.push(err)
         }
-      }
+      })
 
       if (errors.length) {
         // TODO:

src/utils.js

@@ -206,3 +206,22 @@ exports.withTimeout = (asyncFn, time) => {
     ])
   }
 }
+
+/**
+ * Iterates the given `asyncIterator` and runs each item through the given `asyncFn` in parallel.
+ * Returns a promise that resolves when all items of the `asyncIterator` have been passed
+ * through `asyncFn`.
+ *
+ * @param {AsyncIterable} [asyncIterator]
+ * @param {Function} [asyncFn]
+ * @returns {Array}
+ *
+ * @private
+ */
+exports.mapParallel = async function (asyncIterator, asyncFn) {
+  const tasks = []
+  for await (const item of asyncIterator) {
+    tasks.push(asyncFn(item))
+  }
+  return Promise.all(tasks)
+}