fix: FileShare: fast-path to avoid a timeout issue when many read/write-calls are queued at the same time

Author: nytamin

apps/appcontainer-node/packages/generic/src/appContainer.ts

@@ -105,6 +105,8 @@ export class AppContainer {
 								client.once('close', () => {
 									this.logger.warn(`Connection to Worker "${client.clientId}" closed`)
 									app.workerAgentApi = null
+
+									this.workerStorage.releaseLockForTag(client.clientId)
 								})
 								this.logger.info(`Connection to Worker "${client.clientId}" established`)
 								app.workerAgentApi = api
@@ -236,7 +238,7 @@ export class AppContainer {
 				dataId: string,
 				customTimeout?: number
 			): Promise<{ lockId: string; current: any | undefined }> => {
-				return this.workerStorage.getWriteLock(dataId, customTimeout)
+				return this.workerStorage.getWriteLock(dataId, customTimeout, clientId)
 			},
 			workerStorageReleaseLock: async (dataId: string, lockId: string): Promise<void> => {
 				return this.workerStorage.releaseLock(dataId, lockId)

shared/packages/api/src/__tests__/lib.spec.ts

@@ -1,4 +1,4 @@
-import { diff, promiseTimeout, stringifyError, waitTime } from '../lib'
+import { deferGets, diff, promiseTimeout, stringifyError, waitTime } from '../lib'
 
 describe('lib', () => {
 	test('diff', () => {
@@ -136,5 +136,46 @@ describe('lib', () => {
 
 		expect(stringifyError(errWithContext, true)).toEqual('Error: errr, Context: a context')
 	})
+	test('deferGets', async () => {
+		let i = 0
+		const deferred = deferGets(async (val: string) => {
+			await waitTime(10)
+			return `${val}_${i++}`
+		})
+
+		const values = await Promise.all([
+			deferred('a', 'a1'), // will be executed
+			deferred('b', 'b2'), // will be executed
+			deferred('a', 'a3'), // will not be executed
+			deferred('a', 'a4'), // will not be executed
+			deferred('c', 'c5'), // will be executed
+			deferred('a', 'a6'), // will not be executed
+			deferred('b', 'b7'), // will not be executed
+		])
+
+		const values2 = await Promise.all([
+			deferred('a', 'a8'), // will be executed
+			deferred('a', 'a9'), // will not be executed
+			deferred('b', 'b10'), // will be executed
+			deferred('a', 'a11'), // will not be executed
+		])
+
+		expect(values).toEqual([
+			'a1_0', // was executed
+			'b2_1', // was executed
+			'a1_0', // was not executed
+			'a1_0', // was not executed
+			'c5_2', // was executed
+			'a1_0', // was not executed
+			'b2_1', // was not executed
+		])
+		expect(values2).toEqual([
+			'a8_3', // was executed
+			'a8_3', // was not executed
+			'b10_4', // was executed
+			'a8_3', // was not executed
+		])
+		expect(i).toBe(5)
+	})
 })
 export {}

shared/packages/api/src/dataStorage.ts

@@ -16,6 +16,7 @@ export class DataStore {
 				ttl: number
 				timeout?: NodeJS.Timeout
 			} | null
+			tag: string | undefined
 			/** The data */
 			data: any
 		}
@@ -54,14 +55,19 @@ export class DataStore {
 	}
 
 	/** Request to aquire a write lock */
-	async getWriteLock(dataId: string, customTimeout?: number): Promise<{ lockId: string; current: any | undefined }> {
+	async getWriteLock(
+		dataId: string,
+		customTimeout?: number,
+		tag?: string
+	): Promise<{ lockId: string; current: any | undefined }> {
 		// Wait for getting access to the data:
 		await this._waitForAccess(dataId)
 		// Set a write lock:
 		if (!this.storage.has(dataId)) {
 			this.storage.set(dataId, {
 				accessLock: null,
 				data: undefined,
+				tag: tag,
 			})
 		}
 		const data = this.storage.get(dataId)
@@ -93,6 +99,14 @@ export class DataStore {
 			this._triggerHandleClaims(true)
 		}
 	}
+	/** Release all locks for a certain tag */
+	releaseLockForTag(tag: string): void {
+		this.storage.forEach((value, dataId) => {
+			if (value.accessLock && value.tag === tag) {
+				this.releaseLock(dataId, value.accessLock.lockId)
+			}
+		})
+	}
 	write(dataId: string, lockId: string, writeData: string): void {
 		const data = this.storage.get(dataId)
 		if (!data) throw new Error(`DataStorage: Error when trying to write data: "${dataId}" not found`)
@@ -184,7 +198,7 @@ export class DataStore {
 								this.logger.warn(
 									`AccessLock timed out after ${Date.now() - data.accessLock.created} ms "${
 										data.accessLock.lockId
-									}"`
+									}", claim count: ${claims.length}`
 								)
 								isWaitingForLock = false
 							}

shared/packages/api/src/lib.ts

@@ -49,6 +49,7 @@ export function waitTime(duration: number): Promise<void> {
 		setTimeout(resolve, duration)
 	})
 }
+/** Intercepts a promise and rejects if the promise doesn't resolve in time. */
 export function promiseTimeout<T>(
 	p: Promise<T>,
 	timeoutTime: number,
@@ -214,3 +215,47 @@ export function isNodeRunningInDebugMode(): boolean {
 		typeof v8debug === 'object' || /--debug|--inspect/.test(process.execArgv.join(' ') + process.env.NODE_OPTIONS)
 	)
 }
+
+/**
+ * Wraps a function, so that multiple calls to it will be grouped together,
+ * if the calls are close enough in time so that the resulting promise havent resolved yet.
+ * The subsequent calls will resolve with the same result as the first call.
+ */
+export function deferGets<Args extends any[], Result>(
+	fcn: (...args: Args) => Promise<Result>
+): (groupId: string, ...args: Args) => Promise<Result> {
+	const defers = new Map<
+		string,
+		{
+			resolve: (value: Result) => void
+			reject: (err: any) => void
+		}[]
+	>()
+
+	return (groupId: string, ...args: Args) => {
+		return new Promise<Result>((resolve, reject) => {
+			// Check if there already is a call waiting:
+			const waiting = defers.get(groupId)
+			if (waiting) {
+				waiting.push({ resolve, reject })
+			} else {
+				const newWaiting = [{ resolve, reject }]
+				defers.set(groupId, newWaiting)
+
+				fcn(...args)
+					.then((result) => {
+						defers.delete(groupId)
+						for (const w of newWaiting) {
+							w.resolve(result)
+						}
+					})
+					.catch((err) => {
+						defers.delete(groupId)
+						for (const w of newWaiting) {
+							w.reject(err)
+						}
+					})
+			}
+		})
+	}
+}

shared/packages/worker/src/worker/accessorHandlers/fileShare.ts

@@ -388,102 +388,134 @@ export class FileShareAccessorHandle<Metadata> extends GenericFileAccessorHandle
 			// On windows, we can assign the share to a drive letter, as that increases performance quite a lot:
 			const windowsWorker = this.worker as WindowsWorker
 
+			const STORE_DRIVELETTERS = `fileShare_driveLetters_${this.worker.agentAPI.location.localComputerId}`
 			// Note: Use the mappedDriveLetters as a WorkerStorage, to avoid a potential issue where other workers
 			// mess with the drive letter at the same time that we do, and we all end up to be unsynced with reality.
-			await this.worker.agentAPI.workerStorageWrite<MappedDriveLetters>(
-				`fileShare_driveLetters_${this.worker.agentAPI.location.localComputerId}`,
-				PREPARE_FILE_ACCESS_TIMEOUT,
-				async (mappedDriveLetters0): Promise<MappedDriveLetters> => {
-					const mappedDriveLetters: MappedDriveLetters = mappedDriveLetters0 ?? {}
-					// First we check if the drive letter has already been assigned in our cache:
-					let foundMappedDriveLetter: string | null = null
-					for (const [driveLetter, mountedPath] of Object.entries(mappedDriveLetters)) {
-						if (mountedPath === folderPath) {
-							foundMappedDriveLetter = driveLetter
-						}
-					}
 
-					if (foundMappedDriveLetter && forceRemount) {
-						// Force a re-mount of the drive letter:
-						delete mappedDriveLetters[foundMappedDriveLetter]
-						await networkDrive.unmount(foundMappedDriveLetter)
-						foundMappedDriveLetter = null
-					}
+			if (!forceRemount) {
+				// Fast-path, just read the drive letter from the store:
+				// Note: This is a fast path in the case of many jobs fired at several Workers simultaneously,
+				// they can all read in parallel from the same store. When doing a .workerStorageWrite(), that is a single-threaded process.
 
-					if (foundMappedDriveLetter) {
-						// It seems a drive letter is already mapped up.
-						this.actualFolderPath = `${foundMappedDriveLetter}:\\`
-						handlingDone = true
+				const mappedDriveLetters: MappedDriveLetters =
+					(await this.worker.agentAPI.workerStorageRead<MappedDriveLetters>(STORE_DRIVELETTERS)) ?? {}
+
+				// Check if the drive letter has already been assigned in our cache:
+				let foundMappedDriveLetter: string | null = null
+				for (const [driveLetter, mountedPath] of Object.entries(mappedDriveLetters)) {
+					if (mountedPath === folderPath) {
+						foundMappedDriveLetter = driveLetter
 					}
-					if (!handlingDone) {
-						// Update our cache of mounted drive letters:
-						for (const [driveLetter, mountedPath] of Object.entries(await this.getMountedDriveLetters())) {
-							mappedDriveLetters[driveLetter] = mountedPath
-							// If the mounted path is the one we want, we don't have to mount a new one:
+				}
+				if (foundMappedDriveLetter) {
+					// It seems a drive letter is already mapped up.
+					this.actualFolderPath = `${foundMappedDriveLetter}:\\`
+					handlingDone = true
+				}
+			}
+
+			if (!handlingDone) {
+				await this.worker.agentAPI.workerStorageWrite<MappedDriveLetters>(
+					STORE_DRIVELETTERS,
+					PREPARE_FILE_ACCESS_TIMEOUT,
+					async (mappedDriveLetters0): Promise<MappedDriveLetters> => {
+						const mappedDriveLetters: MappedDriveLetters = mappedDriveLetters0 ?? {}
+						// First we check if the drive letter has already been assigned in our cache:
+						let foundMappedDriveLetter: string | null = null
+						for (const [driveLetter, mountedPath] of Object.entries(mappedDriveLetters)) {
 							if (mountedPath === folderPath) {
 								foundMappedDriveLetter = driveLetter
 							}
 						}
+
+						if (foundMappedDriveLetter && forceRemount) {
+							// Force a re-mount of the drive letter:
+							delete mappedDriveLetters[foundMappedDriveLetter]
+							await networkDrive.unmount(foundMappedDriveLetter)
+							foundMappedDriveLetter = null
+						}
+
 						if (foundMappedDriveLetter) {
+							// It seems a drive letter is already mapped up.
 							this.actualFolderPath = `${foundMappedDriveLetter}:\\`
 							handlingDone = true
 						}
-					}
-
-					if (!handlingDone) {
-						// Find next free drive letter:
-						const freeDriveLetter = windowsWorker.agentAPI.config.windowsDriveLetters?.find(
-							(driveLetter) => !mappedDriveLetters[driveLetter]
-						)
-
-						if (freeDriveLetter) {
-							// Try to map the remote share onto a drive:
-
-							try {
-								await networkDrive.mount(
-									folderPath,
-									freeDriveLetter,
-									this.accessor.userName,
-									this.accessor.password
-								)
-							} catch (e) {
-								const errStr = `${e}`
-								if (
-									errStr.match(/invalid response/i) ||
-									errStr.match(/Ugyldig svar/i) // "Invalid response" in Norvegian
-								) {
-									// Temporary handling of the error
-
-									const mappedDrives = await this.getMountedDriveLetters()
-
-									if (mappedDrives[freeDriveLetter] === folderPath) {
-										this.worker.logger.warn(`Supressed error: ${errStr}`)
-
-										this.worker.logger.warn(`Mapped drives: ${Object.keys(mappedDrives).join(',')}`)
-										this.worker.logger.warn(
-											`${freeDriveLetter} is currently mapped to ${mappedDrives[freeDriveLetter]}`
-										)
-									} else {
-										this.worker.logger.warn(`Mapped drives: ${Object.keys(mappedDrives).join(',')}`)
-										this.worker.logger.warn(
-											`${freeDriveLetter} is currently mapped to ${mappedDrives[freeDriveLetter]}`
-										)
-										throw e
-									}
-								} else throw e
+						if (!handlingDone) {
+							// Update our cache of mounted drive letters:
+							for (const [driveLetter, mountedPath] of Object.entries(
+								await this.getMountedDriveLetters()
+							)) {
+								mappedDriveLetters[driveLetter] = mountedPath
+								// If the mounted path is the one we want, we don't have to mount a new one:
+								if (mountedPath === folderPath) {
+									foundMappedDriveLetter = driveLetter
+								}
 							}
+							if (foundMappedDriveLetter) {
+								this.actualFolderPath = `${foundMappedDriveLetter}:\\`
+								handlingDone = true
+							}
+						}
 
-							mappedDriveLetters[freeDriveLetter] = folderPath
-							this.actualFolderPath = `${freeDriveLetter}:\\`
-							handlingDone = true
-						} else {
-							// Not able to find any free drive letters.
-							// Revert to direct access then
+						if (!handlingDone) {
+							// Find next free drive letter:
+							const freeDriveLetter = windowsWorker.agentAPI.config.windowsDriveLetters?.find(
+								(driveLetter) => !mappedDriveLetters[driveLetter]
+							)
+
+							if (freeDriveLetter) {
+								// Try to map the remote share onto a drive:
+
+								try {
+									await networkDrive.mount(
+										folderPath,
+										freeDriveLetter,
+										this.accessor.userName,
+										this.accessor.password
+									)
+								} catch (e) {
+									const errStr = `${e}`
+									if (
+										errStr.match(/invalid response/i) ||
+										errStr.match(/Ugyldig svar/i) // "Invalid response" in Norvegian
+									) {
+										// Temporary handling of the error
+
+										const mappedDrives = await this.getMountedDriveLetters()
+
+										if (mappedDrives[freeDriveLetter] === folderPath) {
+											this.worker.logger.warn(`Supressed error: ${errStr}`)
+
+											this.worker.logger.warn(
+												`Mapped drives: ${Object.keys(mappedDrives).join(',')}`
+											)
+											this.worker.logger.warn(
+												`${freeDriveLetter} is currently mapped to ${mappedDrives[freeDriveLetter]}`
+											)
+										} else {
+											this.worker.logger.warn(
+												`Mapped drives: ${Object.keys(mappedDrives).join(',')}`
+											)
+											this.worker.logger.warn(
+												`${freeDriveLetter} is currently mapped to ${mappedDrives[freeDriveLetter]}`
+											)
+											throw e
+										}
+									} else throw e
+								}
+
+								mappedDriveLetters[freeDriveLetter] = folderPath
+								this.actualFolderPath = `${freeDriveLetter}:\\`
+								handlingDone = true
+							} else {
+								// Not able to find any free drive letters.
+								// Revert to direct access then
+							}
 						}
+						return mappedDriveLetters
 					}
-					return mappedDriveLetters
-				}
-			)
+				)
+			}
 		}
 
 		if (!handlingDone) {