Faster artifact and cache uploads: blocks joined in parallel, more blocks in flight (#224)
* Artifact and cache uploads: blocks joined in parallel, more blocks in flight
- CI / Typecheck and bundleFailed
- CI / Unit tests (1/2)Failed
- CI / Unit tests (2/2)Failed
5 files changed+49−15
Files (5)
modifiedapps/diffinery/docs/actions/caching-and-artifacts.md+3−1View file
| @@ -71,7 +71,7 @@ | ||
| 71 | 71 | | `lookup-only` | No | `false` | Check whether an entry exists without downloading it. Logs `Cache found and can be restored from key: <key>`. | |
| 72 | 72 | | `fail-on-cache-miss` | No | `false` | Fail the step when neither `key` nor any restore key matches. | |
| 73 | 73 | | `save-always` | No | `false` | Save in the post step even when the job failed or was cancelled. | |
| 74 | -| `upload-chunk-size`, `enableCrossOsArchive` | No | | Accepted. The built-in uploads in 32 MiB blocks whatever `upload-chunk-size` says. | | |
| 74 | +| `upload-chunk-size`, `enableCrossOsArchive` | No | | Accepted. The built-in uploads in 16 MiB blocks, 8 at a time (8 MiB blocks, 4 at a time, on the 1 GiB `tiny` runner), whatever `upload-chunk-size` says. | | |
| 75 | 75 | |
| 76 | 76 | `actions/cache/restore` and `actions/cache/save` take the same inputs. If `key` or `path` is missing, the step fails with `Input required and not supplied: key` or `Input required and not supplied: path`. |
| 77 | 77 | |
| @@ -159,6 +159,8 @@ | ||
| 159 | 159 | |
| 160 | 160 | Restoring is built for speed. The runner downloads the archive in 8 MiB ranges, 16 at a time (4 on the 1 GiB `tiny` runner), and pipes them straight into `tar`, so the archive extracts while it is still arriving. On hosted runners a 216 MB pnpm store restores in about 3 to 5 seconds. Every range must come from the same stored entry; if one doesn't, or a range keeps failing after its retries, the log says `Streaming the cache entry failed (<reason>); downloading it in full and extracting it again.` and the runner downloads the whole archive to a file and extracts it again, which replaces anything the stream left half done. If that fails too, the restore removes what it had extracted and behaves like a miss, so a job never runs with half a cache. Artifact downloads use the same ranged download. |
| 161 | 161 | |
| 162 | +Saving is built for speed too. Files of up to 32 MiB go up in one request; larger ones go up in 16 MiB blocks, 8 at a time (8 MiB blocks, 4 at a time, on the `tiny` runner). Diffinery then joins the blocks into one stored object. Objects over 16 MiB are joined as 8 MiB parts, copied by several workers at once, so a 200 MB artifact or cache entry is stored within seconds of its last block. The same applies to uploads from the official actions. | |
| 163 | + | |
| 162 | 164 | - The compression is part of the version, so an entry saved on a runner with zstd doesn't restore on one without it. |
| 163 | 165 | - A path outside the workspace, such as `~/.npm`, is stored relative to the workspace too. A runner whose workspace sits at a different depth restores it into a different directory. Include `runner.environment` or a label in the key if you mix runner types. |
| 164 | 166 | - The step log shows the archive size as `Cache Size: ~<n> MB (<bytes> B)` on both save and restore. |
modifiedapps/diffinery/src/lib/blob-store.ts+24−10View file
| @@ -26,12 +26,22 @@ | ||
| 26 | 26 | * each step runs in an invocation of its own (a Durable Object call). |
| 27 | 27 | */ |
| 28 | 28 | export const STEP_OPS = 400; |
| 29 | -/** Largest object a merge step writes (an intermediate, or a final object written in one put). */ | |
| 29 | +/** Largest object a merge step writes (an intermediate). */ | |
| 30 | 30 | const MAX_MERGE_BYTES = 1024 * MiB; |
| 31 | +/** | |
| 32 | + * Largest final object written in one put. One step copies about 6-7 MB/s (a 192 MB artifact's | |
| 33 | + * block list took 25-38 s on hosted runners), so anything larger goes up as multipart parts, | |
| 34 | + * copied by several steps at once. | |
| 35 | + */ | |
| 36 | +const SINGLE_PUT_BYTES = 16 * MiB; | |
| 37 | +/** Bytes of parts one step copies, so a large object's parts spread over many steps. */ | |
| 38 | +const STEP_BYTES = 16 * MiB; | |
| 39 | +/** Steps that run at once. */ | |
| 40 | +const COMPOSE_CONCURRENCY = 8; | |
| 31 | 41 | |
| 32 | 42 | /** The limits a plan keeps to (smaller ones let tests exercise every path with little data). */ |
| 33 | -export type ComposeLimits = { stepOps: number; maxMergeBytes: number }; | |
| 34 | -export const COMPOSE_LIMITS: ComposeLimits = { stepOps: STEP_OPS, maxMergeBytes: MAX_MERGE_BYTES }; | |
| 43 | +export type ComposeLimits = { stepOps: number; maxMergeBytes: number; singlePutBytes?: number; stepBytes?: number }; | |
| 44 | +export const COMPOSE_LIMITS: ComposeLimits = { stepOps: STEP_OPS, maxMergeBytes: MAX_MERGE_BYTES, singlePutBytes: SINGLE_PUT_BYTES, stepBytes: STEP_BYTES }; | |
| 35 | 45 | |
| 36 | 46 | export function finalKey(b: BlobRef): string { |
| 37 | 47 | return runtimeObjectKey(b); |
| @@ -238,7 +248,7 @@ | ||
| 238 | 248 | const intermediates: string[] = []; |
| 239 | 249 | let segs = segments.filter((s) => s.size > 0); |
| 240 | 250 | const { stepOps, maxMergeBytes } = limits; |
| 241 | - const fitsOneStep = () => segs.length <= stepOps - 1 && total <= maxMergeBytes; | |
| 251 | + const fitsOneStep = () => segs.length <= stepOps - 1 && total <= Math.min(maxMergeBytes, limits.singlePutBytes ?? maxMergeBytes); | |
| 242 | 252 | const partsFit = () => cut(segs, partSize).every((p) => p.length <= stepOps - 1); |
| 243 | 253 | // Merged segments may reach a part's size, so that a part never needs more slices than a step allows. |
| 244 | 254 | const mergeCap = Math.max(maxMergeBytes, partSize); |
| @@ -273,20 +283,24 @@ | ||
| 273 | 283 | return { rounds, total, partSize, segs, parts: cut(segs, partSize), single: false, intermediates }; |
| 274 | 284 | } |
| 275 | 285 | |
| 276 | -/** Groups consecutive parts into steps of at most {@link STEP_OPS} calls. */ | |
| 277 | -export function batchParts(parts: Slice[][], stepOps = STEP_OPS): Array<Array<{ partNumber: number; slices: Slice[]; length: number }>> { | |
| 286 | +/** Groups consecutive parts into steps of at most {@link STEP_OPS} calls and about `stepBytes` bytes. */ | |
| 287 | +export function batchParts(parts: Slice[][], stepOps = STEP_OPS, stepBytes = Infinity): Array<Array<{ partNumber: number; slices: Slice[]; length: number }>> { | |
| 278 | 288 | const out: Array<Array<{ partNumber: number; slices: Slice[]; length: number }>> = []; |
| 279 | 289 | let cur: Array<{ partNumber: number; slices: Slice[]; length: number }> = []; |
| 280 | 290 | let ops = 0; |
| 291 | + let bytes = 0; | |
| 281 | 292 | parts.forEach((slices, i) => { |
| 282 | 293 | const cost = slices.length + 1; |
| 283 | - if (cur.length && ops + cost > stepOps) { | |
| 294 | + const length = sliceBytes(slices); | |
| 295 | + if (cur.length && (ops + cost > stepOps || bytes + length > stepBytes)) { | |
| 284 | 296 | out.push(cur); |
| 285 | 297 | cur = []; |
| 286 | 298 | ops = 0; |
| 299 | + bytes = 0; | |
| 287 | 300 | } |
| 288 | - cur.push({ partNumber: i + 1, slices, length: sliceBytes(slices) }); | |
| 301 | + cur.push({ partNumber: i + 1, slices, length }); | |
| 289 | 302 | ops += cost; |
| 303 | + bytes += length; | |
| 290 | 304 | }); |
| 291 | 305 | if (cur.length) out.push(cur); |
| 292 | 306 | return out; |
| @@ -357,7 +371,7 @@ | ||
| 357 | 371 | const plan = planCompose(segments, o.scratch, limits); |
| 358 | 372 | const pool = async <T>(items: T[], f: (t: T) => Promise<void>) => { |
| 359 | 373 | let next = 0; |
| 360 | - await Promise.all(Array.from({ length: Math.min(o.concurrency ?? 4, items.length) }, async () => { | |
| 374 | + await Promise.all(Array.from({ length: Math.min(o.concurrency ?? COMPOSE_CONCURRENCY, items.length) }, async () => { | |
| 361 | 375 | while (next < items.length) await f(items[next++]!); |
| 362 | 376 | })); |
| 363 | 377 | }; |
| @@ -374,7 +388,7 @@ | ||
| 374 | 388 | const upload = await bucket.createMultipartUpload(key, { httpMetadata }); |
| 375 | 389 | try { |
| 376 | 390 | const etags: Array<{ partNumber: number; etag: string }> = []; |
| 377 | - await pool(batchParts(plan.parts, limits.stepOps), async (parts) => void etags.push(...(await run({ kind: "parts", key, uploadId: upload.uploadId, parts })))); | |
| 391 | + await pool(batchParts(plan.parts, limits.stepOps, limits.stepBytes), async (parts) => void etags.push(...(await run({ kind: "parts", key, uploadId: upload.uploadId, parts })))); | |
| 378 | 392 | etags.sort((a, b) => a.partNumber - b.partNumber); |
| 379 | 393 | await upload.complete(etags); |
| 380 | 394 | } catch (e) { |
modifiedapps/diffinery/test/unit/actions-runtime.test.ts+12−0View file
| @@ -264,6 +264,18 @@ | ||
| 264 | 264 | expect((await run([3 * MiB, 7 * MiB, 5 * MiB, 1], small)).parts).toEqual([8 * MiB, 7 * MiB + 1]); |
| 265 | 265 | }); |
| 266 | 266 | |
| 267 | + it("copies a large upload as multipart parts spread over many steps, not one long put", async () => { | |
| 268 | + // The runner's 32 MiB blocks: one put would copy 101 MiB in a single step (about 15 s); parts of | |
| 269 | + // 8 MiB, two per step, let eight steps copy at once. | |
| 270 | + const r = await run([32 * MiB, 32 * MiB, 32 * MiB, 5 * MiB]); | |
| 271 | + expect(r.multipart).toBe(1); | |
| 272 | + expect(r.parts).toEqual([...Array.from({ length: 12 }, () => 8 * MiB), 5 * MiB]); | |
| 273 | + expect(r.steps).toBe(7); | |
| 274 | + // Up to 16 MiB still goes in one put. | |
| 275 | + expect((await run([16 * MiB])).multipart).toBe(0); | |
| 276 | + expect((await run([16 * MiB, 1])).multipart).toBe(1); | |
| 277 | + }); | |
| 278 | + | |
| 267 | 279 | it("merges many small blocks in rounds so that no step exceeds its calls", async () => { |
| 268 | 280 | // 10 calls per step: 300 blocks need two rounds of merging before the parts |
| 269 | 281 | const r = await run(Array.from({ length: 300 }, (_, i) => 40 * 1024 + i), { stepOps: 10, maxMergeBytes: 1 * MiB }); |
modifiedcontainers/runner/src/runtime.ts+9−3View file
| @@ -28,12 +28,18 @@ | ||
| 28 | 28 | |
| 29 | 29 | /** Uploads at most this much in one request; larger files go up as blocks. */ |
| 30 | 30 | const SINGLE_SHOT = 32 * 1024 * 1024; |
| 31 | -const BLOCK_SIZE = 32 * 1024 * 1024; | |
| 32 | -const UPLOAD_CONCURRENCY = 4; | |
| 31 | +/** | |
| 32 | + * Larger files go up in blocks of this size, this many at a time (measured on hosted runners: 4 of | |
| 33 | + * 32 MiB ran at 23-33 MiB/s, 16 of 16 MiB at 58-62 MiB/s). Blocks in flight are held in memory, so | |
| 34 | + * the 1 GiB runner sends fewer and smaller ones (32 MiB at most). | |
| 35 | + */ | |
| 36 | +const SMALL_MACHINE = totalmem() < 2 * 1024 ** 3; | |
| 37 | +const BLOCK_SIZE = (SMALL_MACHINE ? 8 : 16) * 1024 * 1024; | |
| 38 | +const UPLOAD_CONCURRENCY = SMALL_MACHINE ? 4 : 8; | |
| 33 | 39 | /** Downloads come in ranges of this size, this many at a time (measured on hosted runners: 16 of 8 MiB ran 2-5x faster than one stream). */ |
| 34 | 40 | const DOWNLOAD_CHUNK = 8 * 1024 * 1024; |
| 35 | 41 | // Fewer on the smallest runner (1 GiB), so ranges waiting their turn hold at most 32 MiB there. |
| 36 | -const DOWNLOAD_CONCURRENCY = totalmem() < 2 * 1024 ** 3 ? 4 : 16; | |
| 42 | +const DOWNLOAD_CONCURRENCY = SMALL_MACHINE ? 4 : 16; | |
| 37 | 43 | |
| 38 | 44 | export class RuntimeClient { |
| 39 | 45 | constructor( |
modifiedcontainers/runner/test/runtime.test.ts+1−1View file
| @@ -177,7 +177,7 @@ | ||
| 177 | 177 | writeFileSync(path.join(d, "big"), big); |
| 178 | 178 | writeFileSync(path.join(d, "small"), "hello"); |
| 179 | 179 | expect(await rt.upload(`${base}/big?sig=1`, path.join(d, "big"))).toBe(big.length); |
| 180 | - expect(calls).toEqual(["PUT block", "PUT block", "PUT blocklist"]); | |
| 180 | + expect(calls).toEqual(["PUT block", "PUT block", "PUT block", "PUT blocklist"]); // 16 MiB blocks | |
| 181 | 181 | expect(await rt.upload(`${base}/small?sig=1`, path.join(d, "small"))).toBe(5); |
| 182 | 182 | expect(calls.at(-1)).toBe("PUT -"); |
| 183 | 183 | const meta = await rt.download(`${base}/big?sig=1`, path.join(d, "back")); |