Security News
Research
Data Theft Repackaged: A Case Study in Malicious Wrapper Packages on npm
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
async-mutex
Advanced tools
The async-mutex package provides a mutual exclusion mechanism (mutex) for asynchronous operations in JavaScript. This is useful when you need to ensure that only one piece of code can access a particular resource or execute a specific task at a time, even in a concurrent or asynchronous environment.
Mutex
The Mutex class allows you to create a lock that can be used to protect a critical section of code so that only one caller can execute it at a time.
const { Mutex } = require('async-mutex');
const mutex = new Mutex();
async function criticalSection() {
const release = await mutex.acquire();
try {
// Perform the critical task that must not be interrupted
} finally {
release();
}
}
Semaphore
The Semaphore class is similar to Mutex but allows multiple concurrent accesses to a resource, with the maximum number of concurrent accesses specified during the creation of the Semaphore.
const { Semaphore } = require('async-mutex');
const semaphore = new Semaphore(2);
async function limitedAccessFunction() {
const [value, release] = await semaphore.acquire();
try {
// Perform a task that should only allow 2 concurrent accesses
} finally {
release();
}
}
The semaphore package provides a similar functionality to async-mutex, offering both semaphore and mutex implementations. It allows for limiting concurrent access to a resource, but the API and implementation details may differ.
This package also offers semaphore functionality for synchronizing asynchronous operations. It is similar to async-mutex but focuses solely on semaphores and does not provide a mutex implementation.
Bottleneck is a rate limiter that can also be used to control access to resources. While it is more focused on rate limiting rather than mutual exclusion, it can be configured to act similarly to a semaphore from async-mutex.
This package implements primitives for synchronizing asynchronous operations in Javascript.
The term "mutex" usually refers to a data structure used to synchronize concurrent processes running on different threads. For example, before accessing a non-threadsafe resource, a thread will lock the mutex. This is guaranteed to block the thread until no other thread holds a lock on the mutex and thus enforces exclusive access to the resource. Once the operation is complete, the thread releases the lock, allowing other threads to acquire a lock and access the resource.
While Javascript is strictly single-threaded, the asynchronous nature of its execution model allows for race conditions that require similar synchronization primitives. Consider for example a library communicating with a web worker that needs to exchange several subsequent messages with the worker in order to achieve a task. As these messages are exchanged in an asynchronous manner, it is perfectly possible that the library is called again during this process. Depending on the way state is handled during the async process, this will lead to race conditions that are hard to fix and even harder to track down.
This library solves the problem by applying the concept of mutexes to Javascript. Locking the mutex will return a promise that resolves once the mutex becomes available. Once the async process is complete (usually taking multiple spins of the event loop), a callback supplied to the caller should be called in order to release the mutex, allowing the next scheduled worker to execute.
Imagine a situation where you need to control access to several instances of a shared resource. For example, you might want to distribute images between several worker processes that perform transformations, or you might want to create a web crawler that performs a defined number of requests in parallel.
A semaphore is a data structure that is initialized with an arbitrary integer value and that can be locked multiple times. As long as the semaphore value is positive, locking it will return the current value and the locking process will continue execution immediately; the semaphore will be decremented upon locking. Releasing the lock will increment the semaphore again.
Once the semaphore has reached zero, the next process that attempts to acquire a lock will be suspended until another process releases its lock and this increments the semaphore again.
This library provides a semaphore implementation for Javascript that is similar to the mutex implementation described above.
You can install the library into your project via npm
npm install async-mutex
The library is written in TypeScript and will work in any environment that
supports ES5, ES6 promises and Array.isArray
. On ancient browsers,
a shim can be used (e.g. core-js).
No external typings are required for using this library with
TypeScript (version >= 2).
Starting with Node 12.16 and 13.7, native ES6 style imports are supported.
WARNING: Node 13 versions < 13.2.0 fail to import this package correctly. Node 12 and earlier are fine, as are newer versions of Node 13.
CommonJS:
var Mutex = require('async-mutex').Mutex;
var Semaphore = require('async-mutex').Semaphore;
var withTimeout = require('async-mutex').withTimeout;
ES6:
import {Mutex, Semaphore, withTimeout} from 'async-mutex';
TypeScript:
import {Mutex, MutexInterface, Semaphore, SemaphoreInterface, withTimeout} from 'async-mutex';
With the latest version of Node, native ES6 style imports are supported.
const mutex = new Mutex();
Create a new mutex.
Promise style:
mutex
.runExclusive(() => {
// ...
})
.then((result) => {
// ...
});
async/await:
await mutex.runExclusive(async () => {
// ...
});
runExclusive
schedules the supplied callback to be run once the mutex is unlocked.
The function may return a promise. Once the promise is resolved or rejected (or immediately after
execution if an immediate value was returned),
the mutex is released. runExclusive
returns a promise that adopts the state of the function result.
The mutex is released and the result rejected if an exception occurs during execution of the callback.
Promise style:
mutex
.acquire()
.then(function(release) {
// ...
release();
});
async/await:
const release = await mutex.acquire();
try {
// ...
} finally {
release();
}
acquire
returns an (ES6) promise that will resolve as soon as the mutex is
available. The promise resolves with a function release
that
must be called once the mutex should be released again. The release
callback
is idempotent.
IMPORTANT: Failure to call release
will hold the mutex locked and will
likely deadlock the application. Make sure to call release
under all circumstances
and handle exceptions accordingly.
As an alternative to calling the release
callback returned by acquire
, the mutex
can be released by calling release
directly on it:
mutex.release();
mutex.isLocked();
Pending locks can be cancelled by calling cancel()
on the mutex. This will reject
all pending locks with E_CANCELED
:
Promise style:
import {E_CANCELED} from 'async-mutex';
mutex
.runExclusive(() => {
// ...
})
.then(() => {
// ...
})
.catch(e => {
if (e === E_CANCELED) {
// ...
}
});
async/await:
import {E_CANCELED} from 'async-mutex';
try {
await mutex.runExclusive(() => {
// ...
});
} catch (e) {
if (e === E_CANCELED) {
// ...
}
}
This works with acquire
, too:
if acquire
is used for locking, the resulting promise will reject with E_CANCELED
.
The error that is thrown can be customized by passing a different error to the Mutex
constructor:
const mutex = new Mutex(new Error('fancy custom error'));
Note that while all pending locks are cancelled, a currently held lock will not be
revoked. In consequence, the mutex may not be available even after cancel()
has been called.
You can wait until the mutex is available without locking it by calling waitForUnlock()
.
This will return a promise that resolve once the mutex can be acquired again. This operation
will not lock the mutex, and there is no guarantee that the mutex will still be available
once an async barrier has been encountered.
Promise style:
mutex
.waitForUnlock()
.then(() => {
// ...
});
Async/await:
await mutex.waitForUnlock();
// ...
const semaphore = new Semaphore(initialValue);
Creates a new semaphore. initialValue
is an arbitrary integer that defines the
initial value of the semaphore.
Promise style:
semaphore
.runExclusive(function(value) {
// ...
})
.then(function(result) {
// ...
});
async/await:
await semaphore.runExclusive(async (value) => {
// ...
});
runExclusive
schedules the supplied callback to be run once the semaphore is available.
The callback will receive the current value of the semaphore as its argument.
The function may return a promise. Once the promise is resolved or rejected (or immediately after
execution if an immediate value was returned),
the semaphore is released. runExclusive
returns a promise that adopts the state of the function result.
The semaphore is released and the result rejected if an exception occurs during execution of the callback.
runExclusive
accepts a first optional argument weight
. Specifying a weight
will decrement the
semaphore by the specified value, and the callback will only be invoked once the semaphore's
value greater or equal to weight
.
runExclusive
accepts a second optional argument priority
. Specifying a greater value for priority
tells the scheduler to run this task before other tasks. priority
can be any real number. The default
is zero.
Promise style:
semaphore
.acquire()
.then(function([value, release]) {
// ...
release();
});
async/await:
const [value, release] = await semaphore.acquire();
try {
// ...
} finally {
release();
}
acquire
returns an (ES6) promise that will resolve as soon as the semaphore is
available. The promise resolves to an array with the
first entry being the current value of the semaphore, and the second value a
function that must be called to release the semaphore once the critical operation
has completed. The release
callback is idempotent.
IMPORTANT: Failure to call release
will hold the semaphore locked and will
likely deadlock the application. Make sure to call release
under all circumstances
and handle exceptions accordingly.
acquire
accepts a first optional argument weight
. Specifying a weight
will decrement the
semaphore by the specified value, and the semaphore will only be acquired once its
value is greater or equal to weight
.
acquire
accepts a second optional argument priority
. Specifying a greater value for priority
tells the scheduler to release the semaphore to the caller before other callers. priority
can be
any real number. The default is zero.
As an alternative to calling the release
callback returned by acquire
, the semaphore
can be released by calling release
directly on it:
semaphore.release();
release
accepts an optional argument weight
and increments the semaphore accordingly.
IMPORTANT: Releasing a previously acquired semaphore with the releaser that was
returned by acquire will automatically increment the semaphore by the correct weight. If
you release by calling the unscoped release
you have to supply the correct weight
yourself!
semaphore.getValue()
semaphore.isLocked();
The semaphore is considered to be locked if its value is either zero or negative.
The value of a semaphore can be set directly to a desired value. A positive value will cause the semaphore to schedule any pending waiters accordingly.
semaphore.setValue();
Pending locks can be cancelled by calling cancel()
on the semaphore. This will reject
all pending locks with E_CANCELED
:
Promise style:
import {E_CANCELED} from 'async-mutex';
semaphore
.runExclusive(() => {
// ...
})
.then(() => {
// ...
})
.catch(e => {
if (e === E_CANCELED) {
// ...
}
});
async/await:
import {E_CANCELED} from 'async-mutex';
try {
await semaphore.runExclusive(() => {
// ...
});
} catch (e) {
if (e === E_CANCELED) {
// ...
}
}
This works with acquire
, too:
if acquire
is used for locking, the resulting promise will reject with E_CANCELED
.
The error that is thrown can be customized by passing a different error to the Semaphore
constructor:
const semaphore = new Semaphore(2, new Error('fancy custom error'));
Note that while all pending locks are cancelled, any currently held locks will not be
revoked. In consequence, the semaphore may not be available even after cancel()
has been called.
You can wait until the semaphore is available without locking it by calling waitForUnlock()
.
This will return a promise that resolve once the semaphore can be acquired again. This operation
will not lock the semaphore, and there is no guarantee that the semaphore will still be available
once an async barrier has been encountered.
Promise style:
semaphore
.waitForUnlock()
.then(() => {
// ...
});
Async/await:
await semaphore.waitForUnlock();
// ...
waitForUnlock
accepts optional arguments weight
and priority
. The promise will resolve as soon
as it is possible to acquire
the semaphore with the given weight and priority. Scheduled tasks with
the greatest priority
values execute first.
Sometimes it is desirable to limit the time a program waits for a mutex or
semaphore to become available. The withTimeout
decorator can be applied
to both semaphores and mutexes and changes the behavior of acquire
and
runExclusive
accordingly.
import {withTimeout, E_TIMEOUT} from 'async-mutex';
const mutexWithTimeout = withTimeout(new Mutex(), 100);
const semaphoreWithTimeout = withTimeout(new Semaphore(5), 100);
The API of the decorated mutex or semaphore is unchanged.
The second argument of withTimeout
is the timeout in milliseconds. After the
timeout is exceeded, the promise returned by acquire
and runExclusive
will
reject with E_TIMEOUT
. The latter will not run the provided callback in case
of an timeout.
The third argument of withTimeout
is optional and can be used to
customize the error with which the promise is rejected.
const mutexWithTimeout = withTimeout(new Mutex(), 100, new Error('new fancy error'));
const semaphoreWithTimeout = withTimeout(new Semaphore(5), 100, new Error('new fancy error'));
A shortcut exists for the case where you do not want to wait for a lock to
be available at all. The tryAcquire
decorator can be applied to both mutexes
and semaphores and changes the behavior of acquire
and runExclusive
to
immediately throw E_ALREADY_LOCKED
if the mutex is not available.
Promise style:
import {tryAcquire, E_ALREADY_LOCKED} from 'async-mutex';
tryAcquire(semaphoreOrMutex)
.runExclusive(() => {
// ...
})
.then(() => {
// ...
})
.catch(e => {
if (e === E_ALREADY_LOCKED) {
// ...
}
});
async/await:
import {tryAcquire, E_ALREADY_LOCKED} from 'async-mutex';
try {
await tryAcquire(semaphoreOrMutex).runExclusive(() => {
// ...
});
} catch (e) {
if (e === E_ALREADY_LOCKED) {
// ...
}
}
Again, the error can be customized by providing a custom error as second argument to
tryAcquire
.
tryAcquire(semaphoreOrMutex, new Error('new fancy error'))
.runExclusive(() => {
// ...
});
Feel free to use this library under the conditions of the MIT license.
FAQs
A mutex for guarding async workflows
The npm package async-mutex receives a total of 1,506,005 weekly downloads. As such, async-mutex popularity was classified as popular.
We found that async-mutex demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.
Security News
The Ultralytics' PyPI Package was compromised four times in one weekend through GitHub Actions cache poisoning and failure to rotate previously compromised API tokens.