-
Notifications
You must be signed in to change notification settings - Fork 10.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(gatsby-core-utils): create proper mutex (#34761)
- Loading branch information
Showing
11 changed files
with
333 additions
and
26 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| import path from "path" | ||
| import { remove, mkdirp } from "fs-extra" | ||
| import { createMutex } from "../mutex" | ||
| import * as storage from "../utils/get-storage" | ||
|
|
||
| jest.spyOn(storage, `getDatabaseDir`) | ||
|
|
||
| function sleep(timeout = 100): Promise<void> { | ||
| return new Promise(resolve => setTimeout(resolve, timeout)) | ||
| } | ||
|
|
||
| async function doAsync( | ||
| mutex: ReturnType<typeof createMutex>, | ||
| result: Array<string> = [], | ||
| waitTime: number, | ||
| id: string | ||
| ): Promise<Array<string>> { | ||
| await mutex.acquire() | ||
| result.push(`start ${id}`) | ||
| await sleep(waitTime) | ||
| result.push(`stop ${id}`) | ||
| await mutex.release() | ||
|
|
||
| return result | ||
| } | ||
|
|
||
| describe(`mutex`, () => { | ||
| const cachePath = path.join(__dirname, `.cache`) | ||
| beforeAll(async () => { | ||
| await mkdirp(cachePath) | ||
| storage.getDatabaseDir.mockReturnValue(cachePath) | ||
| }) | ||
|
|
||
| afterAll(async () => { | ||
| await storage.closeDatabase() | ||
| await remove(cachePath) | ||
| }) | ||
|
|
||
| it(`should only allow one action go through at the same time`, async () => { | ||
| const mutex = createMutex(`test-key`, 300) | ||
|
|
||
| const result: Array<string> = [] | ||
|
|
||
| doAsync(mutex, result, 50, `1`) | ||
| await sleep(0) | ||
| await doAsync(mutex, result, 10, `2`) | ||
|
|
||
| expect(result).toMatchInlineSnapshot(` | ||
| Array [ | ||
| "start 1", | ||
| "stop 1", | ||
| "start 2", | ||
| "stop 2", | ||
| ] | ||
| `) | ||
| }) | ||
|
|
||
| it(`should generate the same mutex if key are identical`, async () => { | ||
| const mutex1 = createMutex(`test-key`, 300) | ||
| const mutex2 = createMutex(`test-key`, 300) | ||
|
|
||
| const result: Array<string> = [] | ||
|
|
||
| const mutexPromise = doAsync(mutex1, result, 50, `1`) | ||
| await sleep(0) | ||
| await doAsync(mutex2, result, 10, `2`) | ||
| await mutexPromise | ||
|
|
||
| expect(result).toMatchInlineSnapshot(` | ||
| Array [ | ||
| "start 1", | ||
| "stop 1", | ||
| "start 2", | ||
| "stop 2", | ||
| ] | ||
| `) | ||
| }) | ||
|
|
||
| it(`shouldn't wait if keys are different`, async () => { | ||
| const mutex1 = createMutex(`test-key`, 300) | ||
| const mutex2 = createMutex(`other-key`, 300) | ||
|
|
||
| const result: Array<string> = [] | ||
|
|
||
| const mutexPromise = doAsync(mutex1, result, 50, `1`) | ||
| await sleep(0) | ||
| await doAsync(mutex2, result, 10, `2`) | ||
| await mutexPromise | ||
|
|
||
| expect(result).toMatchInlineSnapshot(` | ||
| Array [ | ||
| "start 1", | ||
| "start 2", | ||
| "stop 2", | ||
| "stop 1", | ||
| ] | ||
| `) | ||
| }) | ||
| }) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| import { getStorage, LockStatus, getDatabaseDir } from "./utils/get-storage" | ||
|
|
||
| interface IMutex { | ||
| acquire(): Promise<void> | ||
| release(): Promise<void> | ||
| } | ||
|
|
||
| // Random number to re-check if mutex got released | ||
| const DEFAULT_MUTEX_INTERVAL = 3000 | ||
|
|
||
| async function waitUntilUnlocked( | ||
| storage: ReturnType<typeof getStorage>, | ||
| key: string, | ||
| timeout: number | ||
| ): Promise<void> { | ||
| const isUnlocked = await storage.mutex.ifNoExists(key, () => { | ||
| storage.mutex.put(key, LockStatus.Locked) | ||
| }) | ||
|
|
||
| if (isUnlocked) { | ||
| return | ||
| } | ||
|
|
||
| await new Promise<void>(resolve => { | ||
| setTimeout(() => { | ||
| resolve(waitUntilUnlocked(storage, key, timeout)) | ||
| }, timeout) | ||
| }) | ||
| } | ||
|
|
||
| /** | ||
| * Creates a mutex, make sure to call `release` when you're done with it. | ||
| * | ||
| * @param {string} key A unique key | ||
| */ | ||
| export function createMutex( | ||
| key: string, | ||
| timeout = DEFAULT_MUTEX_INTERVAL | ||
| ): IMutex { | ||
| const storage = getStorage(getDatabaseDir()) | ||
| const BUILD_ID = global.__GATSBY?.buildId ?? `` | ||
| const prefixedKey = `${BUILD_ID}-${key}` | ||
|
|
||
| return { | ||
| acquire: (): Promise<void> => | ||
| waitUntilUnlocked(storage, prefixedKey, timeout), | ||
| release: async (): Promise<void> => { | ||
| await storage.mutex.remove(prefixedKey) | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| export async function releaseAllMutexes(): Promise<void> { | ||
| const storage = getStorage(getDatabaseDir()) | ||
|
|
||
| await storage.mutex.clearAsync() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| import path from "path" | ||
| import importFrom from "import-from" | ||
| import resolveFrom from "resolve-from" | ||
|
|
||
| export function getLmdb(): typeof import("lmdb") { | ||
| const gatsbyPkgRoot = path.dirname( | ||
| resolveFrom(process.cwd(), `gatsby/package.json`) | ||
| ) | ||
|
|
||
| // Try to use lmdb from gatsby if not we use our own version | ||
| try { | ||
| return importFrom(gatsbyPkgRoot, `lmdb`) as typeof import("lmdb") | ||
| } catch (err) { | ||
| return require(`lmdb`) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| import path from "path" | ||
| import { getLmdb } from "./get-lmdb" | ||
| import type { RootDatabase, Database } from "lmdb" | ||
|
|
||
| export enum LockStatus { | ||
| Locked = 0, | ||
| Unlocked = 1, | ||
| } | ||
|
|
||
| interface ICoreUtilsDatabase { | ||
| mutex: Database<LockStatus, string> | ||
| } | ||
|
|
||
| let databases: ICoreUtilsDatabase | undefined | ||
| let rootDb: RootDatabase | ||
|
|
||
| export function getDatabaseDir(): string { | ||
| const rootDir = global.__GATSBY?.root ?? process.cwd() | ||
| return path.join(rootDir, `.cache`, `data`, `gatsby-core-utils`) | ||
| } | ||
|
|
||
| export function getStorage(fullDbPath: string): ICoreUtilsDatabase { | ||
| if (!databases) { | ||
| if (!fullDbPath) { | ||
| throw new Error(`LMDB path is not set!`) | ||
| } | ||
|
|
||
| // __GATSBY_OPEN_LMDBS tracks if we already opened given db in this process | ||
| // In `gatsby serve` case we might try to open it twice - once for engines | ||
| // and second to get access to `SitePage` nodes (to power trailing slashes | ||
| // redirect middleware). This ensure there is single instance within a process. | ||
| // Using more instances seems to cause weird random errors. | ||
| if (!globalThis.__GATSBY_OPEN_LMDBS) { | ||
| globalThis.__GATSBY_OPEN_LMDBS = new Map() | ||
| } | ||
|
|
||
| databases = globalThis.__GATSBY_OPEN_LMDBS.get(fullDbPath) | ||
|
|
||
| if (databases) { | ||
| return databases | ||
| } | ||
|
|
||
| const open = getLmdb().open | ||
|
|
||
| rootDb = open({ | ||
| name: `root`, | ||
| path: fullDbPath, | ||
| compression: true, | ||
| sharedStructuresKey: Symbol.for(`structures`), | ||
| }) | ||
|
|
||
| databases = { | ||
| mutex: rootDb.openDB({ | ||
| name: `mutex`, | ||
| }), | ||
| } | ||
|
|
||
| globalThis.__GATSBY_OPEN_LMDBS.set(fullDbPath, databases) | ||
| } | ||
|
|
||
| return databases as ICoreUtilsDatabase | ||
| } | ||
|
|
||
| export async function closeDatabase(): Promise<void> { | ||
| if (rootDb) { | ||
| await rootDb.close() | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.