feat: Create new interface for lockers with only 1 kind of lock

This commit is contained in:
Joachim Van Herwegen 2021-01-28 11:13:46 +01:00
parent 59600b07f8
commit b61d46900f
14 changed files with 264 additions and 119 deletions

View File

@ -49,11 +49,14 @@
{
"@id": "urn:solid-server:default:ResourceLocker",
"@type": "WrappedExpiringResourceLocker",
"WrappedExpiringResourceLocker:_locker": {
"@type": "WrappedExpiringReadWriteLocker",
"WrappedExpiringReadWriteLocker:_locker": {
"@type": "EqualReadWriteLocker",
"EqualReadWriteLocker:_locker": {
"@type": "SingleThreadedResourceLocker"
}
},
"WrappedExpiringResourceLocker:_expiration": 3000
"WrappedExpiringReadWriteLocker:_expiration": 3000
},

View File

@ -199,10 +199,12 @@ export * from './util/identifiers/IdentifierStrategy';
export * from './util/identifiers/SingleRootIdentifierStrategy';
// Util/Locking
export * from './util/locking/ExpiringResourceLocker';
export * from './util/locking/ExpiringReadWriteLocker';
export * from './util/locking/EqualReadWriteLocker';
export * from './util/locking/ReadWriteLocker';
export * from './util/locking/ResourceLocker';
export * from './util/locking/SingleThreadedResourceLocker';
export * from './util/locking/WrappedExpiringResourceLocker';
export * from './util/locking/WrappedExpiringReadWriteLocker';
// Util
export * from './util/ContentTypes';

View File

@ -5,7 +5,7 @@ import type { Representation } from '../ldp/representation/Representation';
import type { RepresentationPreferences } from '../ldp/representation/RepresentationPreferences';
import type { ResourceIdentifier } from '../ldp/representation/ResourceIdentifier';
import { getLoggerFor } from '../logging/LogUtil';
import type { ExpiringResourceLocker } from '../util/locking/ExpiringResourceLocker';
import type { ExpiringReadWriteLocker } from '../util/locking/ExpiringReadWriteLocker';
import type { AtomicResourceStore } from './AtomicResourceStore';
import type { Conditions } from './Conditions';
import type { ResourceStore } from './ResourceStore';
@ -19,9 +19,9 @@ export class LockingResourceStore implements AtomicResourceStore {
protected readonly logger = getLoggerFor(this);
private readonly source: ResourceStore;
private readonly locks: ExpiringResourceLocker;
private readonly locks: ExpiringReadWriteLocker;
public constructor(source: ResourceStore, locks: ExpiringResourceLocker) {
public constructor(source: ResourceStore, locks: ExpiringReadWriteLocker) {
this.source = source;
this.locks = locks;
}

View File

@ -0,0 +1,37 @@
import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
import type { ReadWriteLocker } from './ReadWriteLocker';
import type { ResourceLocker } from './ResourceLocker';
/**
* A {@link ReadWriteLocker} that gives no priority to read or write operations: both use the same lock.
*/
export class EqualReadWriteLocker implements ReadWriteLocker {
private readonly locker: ResourceLocker;
public constructor(locker: ResourceLocker) {
this.locker = locker;
}
public async withReadLock<T>(identifier: ResourceIdentifier, whileLocked: () => (Promise<T> | T)): Promise<T> {
return this.withLock(identifier, whileLocked);
}
public async withWriteLock<T>(identifier: ResourceIdentifier, whileLocked: () => (Promise<T> | T)): Promise<T> {
return this.withLock(identifier, whileLocked);
}
/**
* Acquires a new lock for the requested identifier.
* Will resolve when the input function resolves.
* @param identifier - Identifier of resource that needs to be locked.
* @param whileLocked - Function to resolve while the resource is locked.
*/
private async withLock<T>(identifier: ResourceIdentifier, whileLocked: () => T | Promise<T>): Promise<T> {
await this.locker.acquire(identifier);
try {
return await whileLocked();
} finally {
await this.locker.release(identifier);
}
}
}

View File

@ -1,12 +1,12 @@
import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
import type { ResourceLocker } from './ResourceLocker';
import type { ReadWriteLocker } from './ReadWriteLocker';
/**
* A {@link ResourceLocker} where the locks expire after a given time.
* A {@link ReadWriteLocker} where the locks expire after a given time.
*/
export interface ExpiringResourceLocker extends ResourceLocker {
export interface ExpiringReadWriteLocker extends ReadWriteLocker {
/**
* As {@link ResourceLocker.withReadLock} but the locked function gets called with a `maintainLock` callback function
* As {@link ReadWriteLocker.withReadLock} but the locked function gets called with a `maintainLock` callback function
* to reset the lock expiration every time it is called.
* The resulting promise will reject once the lock expires.
*
@ -18,8 +18,8 @@ export interface ExpiringResourceLocker extends ResourceLocker {
=> Promise<T>;
/**
* As {@link ResourceLocker.withWriteLock} but the locked function gets called with a `maintainLock` callback function
* to reset the lock expiration every time it is called.
* As {@link ReadWriteLocker.withWriteLock} but the locked function gets called with a `maintainLock`
* callback function to reset the lock expiration every time it is called.
* The resulting promise will reject once the lock expires.
*
* @param identifier - Identifier of the resource that needs to be locked.

View File

@ -0,0 +1,30 @@
import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
/**
* Allows the locking of resources which is needed for non-atomic {@link ResourceStore}s.
*/
export interface ReadWriteLocker {
/**
* Run the given function while the resource is locked.
* The lock will be released when the (async) input function resolves.
* This function should be used for operations that only require reading the resource.
*
* @param identifier - Identifier of the resource that needs to be locked.
* @param whileLocked - A function to execute while the resource is locked.
*
* @returns A promise resolving when the lock is released.
*/
withReadLock: <T>(identifier: ResourceIdentifier, whileLocked: () => T | Promise<T>) => Promise<T>;
/**
* Run the given function while the resource is locked.
* The lock will be released when the (async) input function resolves.
* This function should be used for operations that could modify the resource.
*
* @param identifier - Identifier of the resource that needs to be locked.
* @param whileLocked - A function to execute while the resource is locked.
*
* @returns A promise resolving when the lock is released.
*/
withWriteLock: <T>(identifier: ResourceIdentifier, whileLocked: () => T | Promise<T>) => Promise<T>;
}

View File

@ -1,30 +1,23 @@
import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
/**
* Allows the locking of resources which is needed for non-atomic {@link ResourceStore}s.
* An interface for classes that only have 1 way to lock interfaces.
* In general this should only be used by components implementing the {@link ReadWriteLocker} interface.
* Other components that require locking of resources should use that interface.
*/
export interface ResourceLocker {
/**
* Run the given function while the resource is locked.
* The lock will be released when the (async) input function resolves.
* This function should be used for operations that only require reading the resource.
*
* @param identifier - Identifier of the resource that needs to be locked.
* @param whileLocked - A function to execute while the resource is locked.
*
* @returns A promise resolving when the lock is released.
* Acquires a lock on the requested identifier.
* The promise will resolve when the lock has been acquired.
* @param identifier - Resource to acquire a lock on.
*/
withReadLock: <T>(identifier: ResourceIdentifier, whileLocked: () => T | Promise<T>) => Promise<T>;
acquire: (identifier: ResourceIdentifier) => Promise<void>;
/**
* Run the given function while the resource is locked.
* The lock will be released when the (async) input function resolves.
* This function should be used for operations that could modify the resource.
*
* @param identifier - Identifier of the resource that needs to be locked.
* @param whileLocked - A function to execute while the resource is locked.
*
* @returns A promise resolving when the lock is released.
* Releases a lock on the requested identifier.
* The promise will resolve when the lock has been released.
* In case there is no lock on the resource an error should be thrown.
* @param identifier - Resource to release the lock on.
*/
withWriteLock: <T>(identifier: ResourceIdentifier, whileLocked: () => T | Promise<T>) => Promise<T>;
release: (identifier: ResourceIdentifier) => Promise<void>;
}

View File

@ -1,46 +1,54 @@
import AsyncLock from 'async-lock';
import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
import { getLoggerFor } from '../../logging/LogUtil';
import { InternalServerError } from '../errors/InternalServerError';
import type { ResourceLocker } from './ResourceLocker';
/**
* A resource locker making use of the `async-lock` library.
* Read and write locks use the same locks so no preference is given to any operations.
* This should be changed at some point though, see #542.
* Note that all locks are kept in memory until they are unlocked which could potentially result
* in a memory leak if locks are never unlocked, so make sure this is covered with expiring locks for example,
* and/or proper `finally` handles.
*/
export class SingleThreadedResourceLocker implements ResourceLocker {
protected readonly logger = getLoggerFor(this);
private readonly locks: AsyncLock;
private readonly locker: AsyncLock;
private readonly unlockCallbacks: Record<string, () => void>;
public constructor() {
this.locks = new AsyncLock();
this.locker = new AsyncLock();
this.unlockCallbacks = {};
}
public async withReadLock<T>(identifier: ResourceIdentifier, whileLocked: () => T | Promise<T>): Promise<T> {
return this.withLock(identifier, whileLocked);
public async acquire(identifier: ResourceIdentifier): Promise<void> {
const { path } = identifier;
this.logger.debug(`Acquiring lock for ${path}`);
return new Promise((resolve): void => {
this.locker.acquire(path, (done): void => {
this.unlockCallbacks[path] = done;
this.logger.debug(`Acquired lock for ${path}. ${this.getLockCount()} locks active.`);
resolve();
}, (): void => {
// eslint-disable-next-line @typescript-eslint/no-dynamic-delete
delete this.unlockCallbacks[path];
this.logger.debug(`Released lock for ${path}. ${this.getLockCount()} active locks remaining.`);
});
});
}
public async withWriteLock<T>(identifier: ResourceIdentifier, whileLocked: () => T | Promise<T>): Promise<T> {
return this.withLock(identifier, whileLocked);
public async release(identifier: ResourceIdentifier): Promise<void> {
const { path } = identifier;
if (!this.unlockCallbacks[path]) {
throw new InternalServerError(`Trying to unlock resource that is not locked: ${path}`);
}
this.unlockCallbacks[path]();
}
/**
* Acquires a new lock for the requested identifier.
* Will resolve when the input function resolves.
* @param identifier - Identifier of resource that needs to be locked.
* @param whileLocked - Function to resolve while the resource is locked.
* Counts the number of active locks.
*/
private async withLock<T>(identifier: ResourceIdentifier, whileLocked: () => T | Promise<T>): Promise<T> {
this.logger.debug(`Acquiring lock for ${identifier.path}`);
try {
return await this.locks.acquire(identifier.path, async(): Promise<T> => {
this.logger.debug(`Acquired lock for ${identifier.path}`);
return whileLocked();
});
} finally {
this.logger.debug(`Released lock for ${identifier.path}`);
}
private getLockCount(): number {
return Object.keys(this.unlockCallbacks).length;
}
}

View File

@ -1,24 +1,24 @@
import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
import { getLoggerFor } from '../../logging/LogUtil';
import { InternalServerError } from '../errors/InternalServerError';
import type { ExpiringResourceLocker } from './ExpiringResourceLocker';
import type { ResourceLocker } from './ResourceLocker';
import type { ExpiringReadWriteLocker } from './ExpiringReadWriteLocker';
import type { ReadWriteLocker } from './ReadWriteLocker';
import Timeout = NodeJS.Timeout;
/**
* Wraps around an existing {@link ResourceLocker} and adds expiration logic to prevent locks from getting stuck.
* Wraps around an existing {@link ReadWriteLocker} and adds expiration logic to prevent locks from getting stuck.
*/
export class WrappedExpiringResourceLocker implements ExpiringResourceLocker {
export class WrappedExpiringReadWriteLocker implements ExpiringReadWriteLocker {
protected readonly logger = getLoggerFor(this);
protected readonly locker: ResourceLocker;
protected readonly locker: ReadWriteLocker;
protected readonly expiration: number;
/**
* @param locker - Instance of ResourceLocker to use for acquiring a lock.
* @param expiration - Time in ms after which the lock expires.
*/
public constructor(locker: ResourceLocker, expiration: number) {
public constructor(locker: ReadWriteLocker, expiration: number) {
this.locker = locker;
this.expiration = expiration;
}

View File

@ -8,10 +8,11 @@ import type { ResourceStore } from '../../src/storage/ResourceStore';
import { APPLICATION_OCTET_STREAM } from '../../src/util/ContentTypes';
import { InternalServerError } from '../../src/util/errors/InternalServerError';
import { SingleRootIdentifierStrategy } from '../../src/util/identifiers/SingleRootIdentifierStrategy';
import type { ExpiringResourceLocker } from '../../src/util/locking/ExpiringResourceLocker';
import type { ResourceLocker } from '../../src/util/locking/ResourceLocker';
import { EqualReadWriteLocker } from '../../src/util/locking/EqualReadWriteLocker';
import type { ExpiringReadWriteLocker } from '../../src/util/locking/ExpiringReadWriteLocker';
import type { ReadWriteLocker } from '../../src/util/locking/ReadWriteLocker';
import { SingleThreadedResourceLocker } from '../../src/util/locking/SingleThreadedResourceLocker';
import { WrappedExpiringResourceLocker } from '../../src/util/locking/WrappedExpiringResourceLocker';
import { WrappedExpiringReadWriteLocker } from '../../src/util/locking/WrappedExpiringReadWriteLocker';
import { guardedStreamFrom } from '../../src/util/StreamUtil';
import { BASE } from './Config';
@ -20,8 +21,8 @@ jest.useFakeTimers();
describe('A LockingResourceStore', (): void => {
let path: string;
let store: LockingResourceStore;
let locker: ResourceLocker;
let expiringLocker: ExpiringResourceLocker;
let locker: ReadWriteLocker;
let expiringLocker: ExpiringReadWriteLocker;
let source: ResourceStore;
let getRepresentationSpy: jest.SpyInstance;
@ -36,8 +37,8 @@ describe('A LockingResourceStore', (): void => {
const initializer = new RootContainerInitializer({ store: source, baseUrl: BASE });
await initializer.handleSafe();
locker = new SingleThreadedResourceLocker();
expiringLocker = new WrappedExpiringResourceLocker(locker, 1000);
locker = new EqualReadWriteLocker(new SingleThreadedResourceLocker());
expiringLocker = new WrappedExpiringReadWriteLocker(locker, 1000);
store = new LockingResourceStore(source, expiringLocker);

View File

@ -4,7 +4,7 @@ import type { Representation } from '../../../src/ldp/representation/Representat
import type { ResourceIdentifier } from '../../../src/ldp/representation/ResourceIdentifier';
import { LockingResourceStore } from '../../../src/storage/LockingResourceStore';
import type { ResourceStore } from '../../../src/storage/ResourceStore';
import type { ExpiringResourceLocker } from '../../../src/util/locking/ExpiringResourceLocker';
import type { ExpiringReadWriteLocker } from '../../../src/util/locking/ExpiringReadWriteLocker';
import { guardedStreamFrom } from '../../../src/util/StreamUtil';
function emptyFn(): void {
@ -13,7 +13,7 @@ function emptyFn(): void {
describe('A LockingResourceStore', (): void => {
let store: LockingResourceStore;
let locker: ExpiringResourceLocker;
let locker: ExpiringReadWriteLocker;
let source: ResourceStore;
let order: string[];
let timeoutTrigger: EventEmitter;

View File

@ -0,0 +1,62 @@
import { EqualReadWriteLocker } from '../../../../src/util/locking/EqualReadWriteLocker';
import type { ResourceLocker } from '../../../../src/util/locking/ResourceLocker';
describe('An EqualReadWriteLocker', (): void => {
let sourceLocker: ResourceLocker;
let locker: EqualReadWriteLocker;
let emptyFn: () => void;
const identifier = { path: 'http://test.com/res' };
beforeEach(async(): Promise<void> => {
emptyFn = jest.fn();
sourceLocker = {
acquire: jest.fn(),
release: jest.fn(),
};
locker = new EqualReadWriteLocker(sourceLocker);
});
it('locks and unlocks a resource for a read lock.', async(): Promise<void> => {
const prom = locker.withReadLock(identifier, emptyFn);
expect(sourceLocker.acquire).toHaveBeenCalledTimes(1);
expect(sourceLocker.acquire).toHaveBeenLastCalledWith(identifier);
expect(emptyFn).toHaveBeenCalledTimes(0);
expect(sourceLocker.release).toHaveBeenCalledTimes(0);
await expect(prom).resolves.toBeUndefined();
expect(sourceLocker.acquire).toHaveBeenCalledTimes(1);
expect(emptyFn).toHaveBeenCalledTimes(1);
expect(sourceLocker.release).toHaveBeenCalledTimes(1);
expect(sourceLocker.release).toHaveBeenLastCalledWith(identifier);
});
it('locks and unlocks a resource for a write lock.', async(): Promise<void> => {
const prom = locker.withWriteLock(identifier, emptyFn);
expect(sourceLocker.acquire).toHaveBeenCalledTimes(1);
expect(sourceLocker.acquire).toHaveBeenLastCalledWith(identifier);
expect(emptyFn).toHaveBeenCalledTimes(0);
expect(sourceLocker.release).toHaveBeenCalledTimes(0);
await expect(prom).resolves.toBeUndefined();
expect(sourceLocker.acquire).toHaveBeenCalledTimes(1);
expect(emptyFn).toHaveBeenCalledTimes(1);
expect(sourceLocker.release).toHaveBeenCalledTimes(1);
expect(sourceLocker.release).toHaveBeenLastCalledWith(identifier);
});
it('unlocks correctly if the whileLocked function errors.', async(): Promise<void> => {
emptyFn = jest.fn().mockRejectedValue(new Error('bad data!'));
const prom = locker.withWriteLock(identifier, emptyFn);
expect(sourceLocker.acquire).toHaveBeenCalledTimes(1);
expect(sourceLocker.acquire).toHaveBeenLastCalledWith(identifier);
expect(emptyFn).toHaveBeenCalledTimes(0);
expect(sourceLocker.release).toHaveBeenCalledTimes(0);
await expect(prom).rejects.toThrow('bad data!');
expect(sourceLocker.acquire).toHaveBeenCalledTimes(1);
expect(emptyFn).toHaveBeenCalledTimes(1);
expect(sourceLocker.release).toHaveBeenCalledTimes(1);
expect(sourceLocker.release).toHaveBeenLastCalledWith(identifier);
});
});

View File

@ -1,62 +1,71 @@
import { InternalServerError } from '../../../../src/util/errors/InternalServerError';
import { SingleThreadedResourceLocker } from '../../../../src/util/locking/SingleThreadedResourceLocker';
describe('A SingleThreadedResourceLocker', (): void => {
let locker: SingleThreadedResourceLocker;
const identifier = { path: 'path' };
let syncCb: () => string;
let asyncCb: () => Promise<string>;
const identifier = { path: 'http://test.com/foo' };
beforeEach(async(): Promise<void> => {
locker = new SingleThreadedResourceLocker();
syncCb = jest.fn((): string => 'sync');
asyncCb = jest.fn(async(): Promise<string> => new Promise((resolve): void => {
setImmediate((): void => resolve('async'));
}));
});
it('can run simple functions with a read lock.', async(): Promise<void> => {
let prom = locker.withReadLock(identifier, syncCb);
await expect(prom).resolves.toBe('sync');
expect(syncCb).toHaveBeenCalledTimes(1);
prom = locker.withReadLock(identifier, asyncCb);
await expect(prom).resolves.toBe('async');
expect(asyncCb).toHaveBeenCalledTimes(1);
it('can lock and unlock a resource.', async(): Promise<void> => {
await expect(locker.acquire(identifier)).resolves.toBeUndefined();
await expect(locker.release(identifier)).resolves.toBeUndefined();
});
it('can run simple functions with a write lock.', async(): Promise<void> => {
let prom = locker.withWriteLock(identifier, syncCb);
await expect(prom).resolves.toBe('sync');
expect(syncCb).toHaveBeenCalledTimes(1);
prom = locker.withWriteLock(identifier, asyncCb);
await expect(prom).resolves.toBe('async');
expect(asyncCb).toHaveBeenCalledTimes(1);
it('can lock a resource again after it was unlocked.', async(): Promise<void> => {
await expect(locker.acquire(identifier)).resolves.toBeUndefined();
await expect(locker.release(identifier)).resolves.toBeUndefined();
await expect(locker.acquire(identifier)).resolves.toBeUndefined();
});
it('errors when unlocking a resource that was not locked.', async(): Promise<void> => {
await expect(locker.acquire(identifier)).resolves.toBeUndefined();
await expect(locker.release(identifier)).resolves.toBeUndefined();
await expect(locker.release(identifier)).rejects.toThrow(InternalServerError);
});
/* eslint-disable jest/valid-expect-in-promise */
it('blocks lock acquisition until they are released.', async(): Promise<void> => {
const results: number[] = [];
const lock1 = locker.acquire(identifier);
const lock2 = locker.acquire(identifier);
const lock3 = locker.acquire(identifier);
const promSlow = locker.withWriteLock(identifier, async(): Promise<void> =>
new Promise((resolve): void => {
setImmediate((): void => {
results.push(1);
resolve();
});
}));
const promFast = locker.withWriteLock(identifier, (): void => {
// Note the different order of calls
const prom2 = lock2.then(async(): Promise<void> => {
results.push(2);
return locker.release(identifier);
});
const prom3 = lock3.then(async(): Promise<void> => {
results.push(3);
return locker.release(identifier);
});
const prom1 = lock1.then(async(): Promise<void> => {
results.push(1);
return locker.release(identifier);
});
await Promise.all([ prom2, prom3, prom1 ]);
expect(results).toEqual([ 1, 2, 3 ]);
});
await Promise.all([ promFast, promSlow ]);
expect(results).toEqual([ 1, 2 ]);
it('can acquire different keys simultaneously.', async(): Promise<void> => {
const results: number[] = [];
const lock1 = locker.acquire({ path: 'path1' });
const lock2 = locker.acquire({ path: 'path2' });
const lock3 = locker.acquire({ path: 'path3' });
await lock2.then(async(): Promise<void> => {
results.push(2);
return locker.release({ path: 'path2' });
});
it('propagates errors.', async(): Promise<void> => {
asyncCb = jest.fn(async(): Promise<string> => new Promise((resolve, reject): void => {
setImmediate((): void => reject(new Error('test')));
}));
const prom = locker.withReadLock(identifier, asyncCb);
await expect(prom).rejects.toThrow('test');
await lock3.then(async(): Promise<void> => {
results.push(3);
return locker.release({ path: 'path3' });
});
await lock1.then(async(): Promise<void> => {
results.push(1);
return locker.release({ path: 'path1' });
});
expect(results).toEqual([ 2, 3, 1 ]);
});
});

View File

@ -1,15 +1,15 @@
import type { ResourceIdentifier } from '../../../../src/ldp/representation/ResourceIdentifier';
import type { ResourceLocker } from '../../../../src/util/locking/ResourceLocker';
import { WrappedExpiringResourceLocker } from '../../../../src/util/locking/WrappedExpiringResourceLocker';
import type { ReadWriteLocker } from '../../../../src/util/locking/ReadWriteLocker';
import { WrappedExpiringReadWriteLocker } from '../../../../src/util/locking/WrappedExpiringReadWriteLocker';
jest.useFakeTimers();
describe('A WrappedExpiringResourceLocker', (): void => {
describe('A WrappedExpiringReadWriteLocker', (): void => {
const identifier = { path: 'path' };
let syncCb: () => string;
let asyncCb: () => Promise<string>;
let wrappedLocker: ResourceLocker;
let locker: WrappedExpiringResourceLocker;
let wrappedLocker: ReadWriteLocker;
let locker: WrappedExpiringReadWriteLocker;
const expiration = 1000;
beforeEach(async(): Promise<void> => {
@ -25,7 +25,7 @@ describe('A WrappedExpiringResourceLocker', (): void => {
setImmediate((): void => resolve('async'));
}));
locker = new WrappedExpiringResourceLocker(wrappedLocker, expiration);
locker = new WrappedExpiringReadWriteLocker(wrappedLocker, expiration);
});
it('calls the wrapped locker for locking.', async(): Promise<void> => {