redis-semaphore

Mutex and Semaphore implementations based on Redis ready for distributed systems

Features

• Fail-safe (all actions performed by LUA scripts (atomic))

Usage

Installation

1npm install --save redis-semaphore ioredis
2# or
3yarn add redis-semaphore ioredis

ioredis is the officially supported Redis client. This library's test code runs on it.

Users of other Redis clients should ensure ioredis-compatible API (see src/types.ts) when creating lock objects.

Mutex

See RedisLabs: Locks with timeouts

new Mutex(redisClient, key [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8, identifier = crypto.randomUUID() }])

• redisClient - required, configured redis client
• key - required, key for locking resource (final key in redis: mutex:<key>)
• options - optional
  • lockTimeout - optional ms, time after mutex will be auto released (expired)
  • acquireTimeout - optional ms, max timeout for .acquire() call
  • acquireAttemptsLimit - optional max number of attempts to be made in .acquire() call
  • retryInterval - optional ms, time between acquire attempts if resource locked
  • refreshInterval - optional ms, auto-refresh interval; to disable auto-refresh behaviour set 0
  • identifier - optional uuid, custom mutex identifier. Must be unique between parallel executors, otherwise multiple locks with same identifier can be treated as the same lock holder. Override only if you know what you are doing (see acquiredExternally option).
  • acquiredExternally - optional true, If identifier provided and acquiredExternally is true then _refresh will be used instead of _acquire in .tryAcquire()/.acquire(). Useful for lock sharing between processes: acquire in scheduler, refresh and release in handler.
  • onLockLost - optional function, called when lock loss is detected due refresh cycle; default onLockLost throws unhandled LostLockError

Example

1const Mutex = require('redis-semaphore').Mutex
2const Redis = require('ioredis')
3
4// TypeScript
5// import { Mutex } from 'redis-semaphore'
6// import Redis from 'ioredis'
7
8const redisClient = new Redis()
9
10async function doSomething() {
11  const mutex = new Mutex(redisClient, 'lockingResource')
12  await mutex.acquire()
13  try {
14    // critical code
15  } finally {
16    await mutex.release()
17  }
18}

Example with lost lock handling

1async function doSomething() {
2  const mutex = new Mutex(redisClient, 'lockingResource', {
3    // By default onLockLost throws unhandled LostLockError
4    onLockLost(err) {
5      console.error(err)
6    }
7  })
8  await mutex.acquire()
9  try {
10    while (mutex.isAcquired) {
11      // critical cycle iteration
12    }
13  } finally {
14    // It's safe to always call release, because if lock is no longer belongs to this mutex, .release() will have no effect
15    await mutex.release()
16  }
17}

Example with optional lock

1async function doSomething() {
2  const mutex = new Mutex(redisClient, 'lockingResource', {
3    acquireAttemptsLimit: 1
4  })
5  const lockAcquired = await mutex.tryAcquire()
6  if (!lockAcquired) {
7    return
8  }
9  try {
10    while (mutex.isAcquired) {
11      // critical cycle iteration
12    }
13  } finally {
14    // It's safe to always call release, because if lock is no longer belongs to this mutex, .release() will have no effect
15    await mutex.release()
16  }
17}

Example with temporary refresh

1async function doSomething() {
2  const mutex = new Mutex(redisClient, 'lockingResource', {
3    lockTimeout: 120000,
4    refreshInterval: 15000
5  })
6  const lockAcquired = await mutex.tryAcquire()
7  if (!lockAcquired) {
8    return
9  }
10  try {
11    // critical cycle iteration
12  } finally {
13    // We want to let lock expire over time after operation is finished
14    await mutex.stopRefresh()
15  }
16}

Example with dynamically adjusting existing lock

1const Mutex = require('redis-semaphore').Mutex
2const Redis = require('ioredis')
3
4// TypeScript
5// import { Mutex } from 'redis-semaphore'
6// import Redis from 'ioredis'
7
8const redisClient = new Redis()
9
10// This creates an original lock
11const preMutex = new Mutex(redisClient, 'lockingResource', {
12  lockTimeout: 10 * 1e3, // lock for 10s
13  refreshInterval: 0
14});
15
16// This modifies lock with a new TTL and starts refresh
17const mutex = new Mutex(redisClient, 'lockingResource', {
18  identifier: preMutex.identifier,
19  acquiredExternally: true, // required in this case
20  lockTimeout: 30 * 60 * 1e3, // lock for 30min
21  refreshInterval: 60 * 1e3
22});
23

Example with shared lock between scheduler and handler apps

1const Mutex = require('redis-semaphore').Mutex
2const Redis = require('ioredis')
3
4// TypeScript
5// import { Mutex } from 'redis-semaphore'
6// import Redis from 'ioredis'
7
8const redisClient = new Redis()
9
10// scheduler app code
11async function every10MinutesCronScheduler() {
12  const mutex = new Mutex(redisClient, 'lockingResource', {
13    lockTimeout: 30 * 60 * 1e3, // lock for 30min
14    refreshInterval: 0
15  })
16  if (await mutex.tryAcquire()) {
17    someQueue.publish({ mutexIdentifier: mutex.identifier })
18  } else {
19    logger.info('Job already scheduled. Do nothing in current cron cycle')
20  }
21}
22
23// handler app code
24async function queueHandler(queueMessageData) {
25  const { mutexIdentifier } = queueMessageData
26  const mutex = new Mutex(redisClient, 'lockingResource', {
27    lockTimeout: 10 * 1e3, // 10sec
28    identifier: mutexIdentifier,
29    acquiredExternally: true // required in this case
30  })
31
32  // actually will do `refresh` with new lockTimeout instead of acquire
33  // if mutex was locked by another process or lock was expired - exception will be thrown (default refresh behavior)
34  await mutex.acquire()
35
36  try {
37    // critical code
38  } finally {
39    await mutex.release()
40  }
41}

Semaphore

See RedisLabs: Basic counting sempahore

This implementation is slightly different from the algorithm described in the book, but the main idea has not changed.

zrank check replaced with zcard, so now it is fair as RedisLabs: Fair semaphore (see tests).

In edge cases (node time difference is greater than lockTimeout) both algorithms are not fair due cleanup stage (removing expired members from sorted set), so FairSemaphore API has been removed (it's safe to replace it with Semaphore).

Most reliable way to use: lockTimeout is greater than possible node clock differences, refreshInterval is not 0 and is less enough than lockTimeout (by default is lockTimeout * 0.8)

new Semaphore(redisClient, key, maxCount [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])

• redisClient - required, configured redis client
• key - required, key for locking resource (final key in redis: semaphore:<key>)
• maxCount - required, maximum simultaneously resource usage count
• options optional See Mutex options

Example

1const Semaphore = require('redis-semaphore').Semaphore
2const Redis = require('ioredis')
3
4// TypeScript
5// import { Semaphore } from 'redis-semaphore'
6// import Redis from 'ioredis'
7
8const redisClient = new Redis()
9
10async function doSomething() {
11  const semaphore = new Semaphore(redisClient, 'lockingResource', 5)
12  await semaphore.acquire()
13  try {
14    // maximum 5 simultaneously executions
15  } finally {
16    await semaphore.release()
17  }
18}

MultiSemaphore

Same as Semaphore with one difference - MultiSemaphore will try to acquire multiple permits instead of one.

MultiSemaphore and Semaphore shares same key namespace and can be used together (see test/src/RedisMultiSemaphore.test.ts).

new MultiSemaphore(redisClient, key, maxCount, permits [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])

• redisClient - required, configured redis client
• key - required, key for locking resource (final key in redis: semaphore:<key>)
• maxCount - required, maximum simultaneously resource usage count
• permits - required, number of acquiring permits
• options optional See Mutex options

Example

1const MultiSemaphore = require('redis-semaphore').MultiSemaphore
2const Redis = require('ioredis')
3
4// TypeScript
5// import { MultiSemaphore } from 'redis-semaphore'
6// import Redis from 'ioredis'
7
8const redisClient = new Redis()
9
10async function doSomething() {
11  const semaphore = new MultiSemaphore(redisClient, 'lockingResource', 5, 2)
12
13  await semaphore.acquire()
14  try {
15    // make 2 parallel calls to remote service which allow only 5 simultaneously calls
16  } finally {
17    await semaphore.release()
18  }
19}

RedlockMutex

Distributed Mutex version

See The Redlock algorithm

new RedlockMutex(redisClients, key [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])

• redisClients - required, array of configured redis client connected to independent nodes
• key - required, key for locking resource (final key in redis: mutex:<key>)
• options optional See Mutex options

Example

1const RedlockMutex = require('redis-semaphore').RedlockMutex
2const Redis = require('ioredis')
3
4// TypeScript
5// import { RedlockMutex } from 'redis-semaphore'
6// import Redis from 'ioredis'
7
8const redisClients = [
9  new Redis('127.0.0.1:6377'),
10  new Redis('127.0.0.1:6378'),
11  new Redis('127.0.0.1:6379')
12] // "Those nodes are totally independent, so we don’t use replication or any other implicit coordination system."
13
14async function doSomething() {
15  const mutex = new RedlockMutex(redisClients, 'lockingResource')
16  await mutex.acquire()
17  try {
18    // critical code
19  } finally {
20    await mutex.release()
21  }
22}

RedlockSemaphore

Distributed Semaphore version

See The Redlock algorithm

new RedlockSemaphore(redisClients, key, maxCount [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])

• redisClients - required, array of configured redis client connected to independent nodes
• key - required, key for locking resource (final key in redis: semaphore:<key>)
• maxCount - required, maximum simultaneously resource usage count
• options optional See Mutex options

Example

1const RedlockSemaphore = require('redis-semaphore').RedlockSemaphore
2const Redis = require('ioredis')
3
4// TypeScript
5// import { RedlockSemaphore } from 'redis-semaphore'
6// import Redis from 'ioredis'
7
8const redisClients = [
9  new Redis('127.0.0.1:6377'),
10  new Redis('127.0.0.1:6378'),
11  new Redis('127.0.0.1:6379')
12] // "Those nodes are totally independent, so we don’t use replication or any other implicit coordination system."
13
14async function doSomething() {
15  const semaphore = new Semaphore(redisClients, 'lockingResource', 5)
16  await semaphore.acquire()
17  try {
18    // maximum 5 simultaneously executions
19  } finally {
20    await semaphore.release()
21  }
22}

RedlockMultiSemaphore

Distributed MultiSemaphore version

See The Redlock algorithm

new RedlockMultiSemaphore(redisClients, key, maxCount, permits [, { lockTimeout = 10000, acquireTimeout = 10000, acquireAttemptsLimit = Number.POSITIVE_INFINITY, retryInterval = 10, refreshInterval = lockTimeout * 0.8 }])

• redisClients - required, array of configured redis client connected to independent nodes
• key - required, key for locking resource (final key in redis: semaphore:<key>)
• maxCount - required, maximum simultaneously resource usage count
• permits - required, number of acquiring permits
• options optional See Mutex options

Example

1const RedlockMultiSemaphore = require('redis-semaphore').RedlockMultiSemaphore
2const Redis = require('ioredis')
3
4// TypeScript
5// import { RedlockMultiSemaphore } from 'redis-semaphore'
6// import Redis from 'ioredis'
7
8const redisClients = [
9  new Redis('127.0.0.1:6377'),
10  new Redis('127.0.0.1:6378'),
11  new Redis('127.0.0.1:6379')
12] // "Those nodes are totally independent, so we don’t use replication or any other implicit coordination system."
13
14async function doSomething() {
15  const semaphore = new RedlockMultiSemaphore(
16    redisClients,
17    'lockingResource',
18    5,
19    2
20  )
21
22  await semaphore.acquire()
23  try {
24    // make 2 parallel calls to remote service which allow only 5 simultaneously calls
25  } finally {
26    await semaphore.release()
27  }
28}

Development

1yarn --immutable
2./setup-redis-servers.sh
3yarn dev

License

MIT