Enable parallel work pool

vgrichina · Mar 17, 2024 · 74a1d85 · 74a1d85
1 parent c2f437d
commit 74a1d85
Showing 1 changed file with 18 additions and 6 deletions.
diff --git a/src/util/near-lake.js b/src/util/near-lake.js
@@ -85,14 +85,26 @@ async function* blockStream({ bucketName, region, endpoint, startAfter, limit, p
     }
 
     try {
+        const workPool = [];
+
         for await (const blockNumber of blockNumbersStream(client, bucketName, startAfter, limit, pageSize)) {
-            const block = JSON.parse((await getFile('block.json', blockNumber)).data.toString());
-            const result = { block, shards: [] };
-            for (let shard = 0; shard < block.chunks.length; shard++) {
-                const chunk =  JSON.parse((await getFile(`shard_${shard}.json`, blockNumber)).data.toString());
-                result.shards.push( chunk );
+            if (workPool.length >= pageSize) {
+                yield await workPool.shift();
             }
-            yield result;
+
+            workPool.push((async () => {
+                const block = JSON.parse((await getFile('block.json', blockNumber)).data.toString());
+                const result = { block, shards: [] };
+                for (let shard = 0; shard < block.chunks.length; shard++) {
+                    const chunk =  JSON.parse((await getFile(`shard_${shard}.json`, blockNumber)).data.toString());
+                    result.shards.push( chunk );
+                }
+                return result;
+            })());
+        }
+
+        for (const work of workPool) {
+            yield await work;
         }
     } finally {
         client.destroy();