use native-promise-pool

Author: satackey

action.yml

@@ -14,6 +14,10 @@ inputs:
     description: An ordered list of keys to use for restoring the cache if no cache hit occurred for key 
     required: false
     default: docker-layer-caching-${{ github.workflow }}-
+  concurrency:
+    description: The number of concurrency when restoring and saving layers
+    required: true
+    default: '4'
 
 runs:
   using: node12

main.ts

@@ -11,6 +11,7 @@ const main = async () => {
   core.saveState(`already-existing-images`, JSON.stringify(await new ImageDetector().getExistingImages()))
 
   const layerCache = new LayerCache([])
+  layerCache.concurrency = parseInt(core.getInput(`concurrency`, { required: true }), 10)
   const restoredKey = await layerCache.restore(primaryKey, restoreKeys)
   await layerCache.cleanUp()
 

package.json

@@ -10,6 +10,7 @@
     "@actions/core": "^1.2.4",
     "actions-exec-listener": "^0.0.2",
     "crypto": "^1.0.1",
+    "native-promise-pool": "^3.13.0",
     "string-format": "^2.0.0",
     "typescript-is": "^0.16.3"
   },

post.ts

@@ -18,6 +18,7 @@ const main = async () => {
   await imageDetector.getExistingImages()
   core.debug(JSON.stringify({ imageIdsToSave: imageDetector.getImagesShouldSave() }))
   const layerCache = new LayerCache(imageDetector.getImagesShouldSave())
+  layerCache.concurrency = parseInt(core.getInput(`concurrency`, { required: true }), 10)
 
   layerCache.unformattedOrigianlKey = primaryKey
   core.debug(JSON.stringify({ restoredKey, formattedOriginalCacheKey: layerCache.getFormattedOriginalCacheKey()}))

src/LayerCache.ts

@@ -7,6 +7,21 @@ import { ExecOptions } from '@actions/exec/lib/interfaces'
 import { promises as fs } from 'fs'
 import { assertManifests, Manifest, Manifests } from './Tar'
 import format from 'string-format'
+import PromisePool from 'native-promise-pool'
+
+type ReturnPromiseFunc = <T>(arg: any) => Promise<T[]>
+
+const createPromiseProducerFromArray = <T, T2>(array: T[], callBackFunc: ((t: T) => Promise<T2>)) => {
+  const currentIndex = 0;
+
+  return () => callBackFunc(array[currentIndex])
+}
+
+const generatePromise = function * <T1, T2>(array: T1[], callbackFunc: ((t: T1) => Promise<T2>)) {
+  for (let i = 0; i < array.length; i++) {
+    yield callbackFunc(array[i])
+  }
+}
 
 class LayerCache {
   // repotag: string
@@ -18,6 +33,7 @@ class LayerCache {
   enabledParallel = true
   // unpackedTarDir: string = ''
   // manifests: Manifests = []
+  concurrency: number = 4
 
   constructor(ids: string[]) {
     // this.repotag = repotag
@@ -108,7 +124,16 @@ class LayerCache {
   }
 
   private async storeLayers(): Promise<number[]> {
-    return await Promise.all((await this.getLayerIds()).map(layerId => this.storeSingleLayerBy(layerId)))
+    const pool = new PromisePool(this.concurrency)
+
+    const result =  Promise.all(
+      (await this.getLayerIds()).map(
+        layerId => {
+          return pool.open(() => this.storeSingleLayerBy(layerId))
+        }
+      )
+    )
+    return result
   }
 
   static async dismissCacheAlreadyExistsError(promise: Promise<number>): Promise<number> {
@@ -174,15 +199,34 @@ class LayerCache {
 
   private async restoreLayers() {
     const restoring = (await this.getLayerIds()).map(layerId => this.restoreSingleLayerBy(layerId));
-    const restoredLayerKeysThatMayContainUndefined = await Promise.all(restoring)
+
+    const promiseIter = generatePromise(await this.getLayerIds(), layerId => this.restoreSingleLayerBy(layerId))
+
+    const pool = new PromisePool(this.concurrency)
+
+    const restoredLayerKeysThatMayContainUndefined = await Promise.all(
+      (await this.getLayerIds()).map(
+        layerId => {
+          return pool.open(() => this.restoreSingleLayerBy(layerId))
+        }
+      )
+    )
+
     core.debug(JSON.stringify({ log: `restoreLayers`, restoredLayerKeysThatMayContainUndefined }))
     const FailedToRestore = (restored: string | undefined) => restored === undefined
     return restoredLayerKeysThatMayContainUndefined.filter(FailedToRestore).length === 0
   }
 
-  private async restoreSingleLayerBy(id: string): Promise<string | undefined> {
+  private async restoreSingleLayerBy(id: string): Promise<string> {
     core.debug(JSON.stringify({ log: `restoreSingleLayerBy`, id }))
-    return await cache.restoreCache([this.genSingleLayerStorePath(id)], this.genSingleLayerStoreKey(id))
+
+    const result = await cache.restoreCache([this.genSingleLayerStorePath(id)], this.genSingleLayerStoreKey(id))
+
+    if (result == null) {
+      throw new Error(`Layer cache not found: ${JSON.stringify({ id })}`)
+    }
+
+    return result
   }
 
   private async loadImageFromUnpacked() {

yarn.lock

@@ -329,6 +329,11 @@ minimatch@^3.0.4:
   dependencies:
     brace-expansion "^1.1.7"
 
+native-promise-pool@^3.13.0:
+  version "3.13.0"
+  resolved "https://registry.yarnpkg.com/native-promise-pool/-/native-promise-pool-3.13.0.tgz#faeaf9b4ee769a79328cc0435930079ad951b27b"
+  integrity sha512-CnUhv57AXMyxGazhYrqhlhwRg1rG36ygrDlnaCa70jBOePhV+OLM9CZkUAgceTXO6BZEQ3F5GRRWwB+wSzI4Aw==
+
 nested-error-stacks@^2:
   version "2.1.0"
   resolved "https://registry.yarnpkg.com/nested-error-stacks/-/nested-error-stacks-2.1.0.tgz#0fbdcf3e13fe4994781280524f8b96b0cdff9c61"