dbmutex

README

go-dbmutex

Package dbmutex implements a DB-based mutex that can be used to synchronize work among multiple processes.

Features

• works with mysql and postgres
• should work with database failovers (haven't verified this)
• mutex lock and unlock operations can timeout if desired
• callers can be notified when mutexes are unlocked
• minimal dependencies

Install

go get github.com/dynata/go-dbmutex

Documentation

Full godoc style documentation for the package can be viewed online without installing this package by using the GoDoc site.

Disclaimer

YOU ARE USING THIS APPLICATION AT YOUR OWN RISK. DYNATA MAKES NO WARRANTIES OR REPRESENTATIONS ABOUT THIS APPLICATION. THIS APPLICATION IS PROVIDED TO YOU “AS IS”. DYNATA HEREBY DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED WITH RESPECT TO THE APPLICATION, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY, FITNESS OF PURPOSE, NON-INFRINGEMENT AND ANY IMPLIED WARRANTIES ARISING OUT OF A COURSE OF PERFORMANCE, DEALING, OR TRADE USAGE. TO THE EXTENT DYNATA MAY NOT, AS A MATTER OF APPLICABLE LAW, DISCLAIM ANY WARRANTY, THE SCOPE AND DURATION OF SUCH WARRANTY SHALL BE LIMITED TO THE MINIMUM PERMITTED UNDER SUCH APPLICABLE LAW. YOU WILL INDEMNIFY, DEFEND AND HOLD HARMLESS DYNATA AND ITS AFFILIATES, EMPLOYEES, OFFICERS AND CONTRACTORS FROM ANY THIRD PARTY CLAIM ARISING FROM YOUR USE OF THIS APPLICATION.

Documentation

Overview

Package dbmutex implements a DB-based mutex that can be used to synchronize work among multiple processes.

Features

- minimal dependencies

- works with mysql and postgres

- works with database failovers

- mutex lock and unlock operations can timeout if desired

- callers can be notified when mutexes are unlocked

Usage

See the examples.

Index

• Constants
• func IgnoreLogError(error) error
• func LogError(e error) error
• type ErrorNotifier
• type Identity
• type Mutex
  • func New(ctx context.Context, db *sql.DB, options ...MutexOption) (*Mutex, error)
  • func (m *Mutex) Identity() Identity
  • func (m *Mutex) Lock(ctx context.Context) (context.Context, error)
  • func (m *Mutex) Unlock(ctx context.Context) error
• type MutexOption
  • func WithCreateMissingTable(b bool) MutexOption
  • func WithDriver(d driver.Driver) MutexOption
  • func WithErrorNotifier(f ErrorNotifier) MutexOption
  • func WithExpiration(t time.Duration) MutexOption
  • func WithFailFast(b bool) MutexOption
  • func WithLockErrorNotifier(f ErrorNotifier) MutexOption
  • func WithMutexName(name string) MutexOption
  • func WithMutexTableName(name string) MutexOption
  • func WithPollInterval(d time.Duration) MutexOption
  • func WithRefresh(t time.Duration) MutexOption
  • func WithRefreshErrorNotifier(f ErrorNotifier) MutexOption
  • func WithUnlockErrorNotifier(f ErrorNotifier) MutexOption

Examples

• Mutex.Lock
• Mutex.Lock (FailFast)
• Mutex.Lock (FastPoll)
• Mutex.Lock (WithTimeout)

Constants

const (
	// DefaultMutexTableName is the default table name that will be used for storing mutexes unless overridden
	// via WithMutexTableName.
	DefaultMutexTableName = "dbmutex"

	// DefaultMutexName is the default mutex name that will be used unless overridden via WithMutexName. A unique
	// mutex name should be used for each locking scenario. It is advised to always use WithMutexName.
	DefaultMutexName = "mutex"

	// DefaultRefresh is the default time to wait between refreshing locked Mutexes. See WithRefresh.
	DefaultRefresh = 1 * time.Second

	// DefaultExpiration is the default time after which a mutex will expire if it has not been automatically
	// refreshed.
	DefaultExpiration = 5 * time.Minute

	// DefaultPollInterval is the default time that will be used to poll a locked mutex when attempting to
	// lock a mutex. Override it via WithPollInterval
	DefaultPollInterval = 1 * time.Second
)

Functions

func IgnoreLogError

func IgnoreLogError(error) error

IgnoreLogError is an ErrorNotifier that does nothing with passed errors.

func LogError

func LogError(e error) error

LogError is an ErrorNotifier that simply logs errors using the standard logger.

Types

type ErrorNotifier

type ErrorNotifier func(error) error

An ErrorNotifier is called to notify when an error occurs while locking, unlocking and refreshing Mutexes. The function should typically return null so that the retries will occur. However, if the function returns non-nil, then the calling code will exit any retry loop. Normally, an ErrorNotifier can be used to simply log the fact that a transient error occurred and then return nil. See WithErrorNotifier.

type Identity

type Identity struct {
	TableName string
	MutexName string
	Hostname  string
	Pid       int
	LockerId  string
}

An Identity uniquely identifies a Mutex. Note that two different Mutex Identities can still be used for mutual exclusion because only TableName, MutexName and DB server are used when determining exclusion. The other data elements like Hostname, Pid and LockerId are additional information.

type Mutex

type Mutex struct {
	// contains filtered or unexported fields
}

A Mutex is used to provide mutual exclusion among multiple processes that have access to a shared database. In order to provide mutual exclusion, the same underlying database table and mutex name (which ties to a row) should be used. (See WithMutexTableName and WithMutexName.) Multiple mutexes can be stored in the same table but different names should be used for different application specific locking scenarios.

Behaviour can be be customized via the MutexOption parameters passed to New and the LockOption parameters passed to Lock.

func New

func New(
	ctx context.Context,
	db *sql.DB,
	options ...MutexOption,
) (*Mutex, error)

New creates a new Mutex. The passed db is used for all database interactions. Behaviour can be customized by passing options. The option that should almost always be passed is the lock name (See WithMutexName.)

func (*Mutex) Identity

func (m *Mutex) Identity() Identity

Identity returns the Identity of this Mutex.

func (*Mutex) Lock

func (m *Mutex) Lock(ctx context.Context) (context.Context, error)

The Lock operation attempts to acquire the Mutex by updating the "locked" column in the underlying database table. If unable to update the row, Lock will poll attempting to acquire the mutex by updating the row. Any database related errors that occur during Lock are reported via an ErrorNotifier (see WithErrorNotifier) but are typically ignored in order to complete the Lock operation. Lock can be timed out by using a ctx parameter that has a deadline (see examples). The poll interval and ability to fail fast after a database error can be controlled via MutexOption passed to New. If the mutex is acquired, a nil error and a context that expires when the Mutex is unlocked are returned. Because polling is used even after database generated errors, Lock will typically not return unless the Mutex is acquired or the ctx expires.

Example

In this simple scenario we wait forever for a lock to be acquired. Note that the lock name is application specific and should be specific to the area of code that needs to be protected.

// Initialization
// Typically the Mutex is created outside of the func that needs to use Lock...
var db *sql.DB // acquire DB as normal
dbm, err := New(context.Background(), db, WithMutexName("application specific"))
if err != nil {
	panic(err)
}

// In the func that needs to protect a section of code...
// Wait until the lock is acquired.
_, err = dbm.Lock(context.Background())
if err != nil {
	// Unable to acquire lock. Because lock acquisition is retried (even in the case
	// of errors), a non-nil err will typically not be returned.
}
defer func() { _ = dbm.Unlock(context.Background()) }()
// Do "critical section" application logic here...

Example (FailFast)

In this scenario we want to acquire the lock or immediately fail if unable to do so.

// Initialization
// Typically the Mutex is created outside of the func that needs to use Lock...
var db *sql.DB // acquire DB as normal
dbm, err := New(
	context.Background(),
	db,
	WithMutexName("application specific"),
	WithFailFast(true),
)
if err != nil {
	panic(err)
}

// In the func that needs to protect a section of code...
// Use background context here but could have gotten context from an input parameter.
ctx := context.Background()
_, err = dbm.Lock(ctx)
var e *dbmerr.LockFailFastError
if errors.As(err, &e) {
	// Failed to immediately acquire the lock.
} else if err != nil {
	// Some other error when attempting to acquire lock.
}
defer func() { _ = dbm.Unlock(context.Background()) }()
// Do "critical section" application logic here...

Example (FastPoll)

In this scenario we want to acquire the lock with a fast poll. Note that if you use a fast poll, you may cause performance problems for the database server.

// Initialization
// Typically the Mutex is created outside of the func that needs to use Lock...
var db *sql.DB // acquire DB as normal
dbm, err := New(
	context.Background(),
	db,
	WithMutexName("application specific"),
	WithPollInterval(time.Millisecond),
)
if err != nil {
	panic(err)
}

// In the func that needs to protect a section of code...
// Use background context here but could have gotten context from an input parameter.
ctx := context.Background()
// This will cause a 1 millisecond poll interval - which will probably cause DB performance problems.
_, err = dbm.Lock(ctx)
if err != nil {
	// Some error when attempting to acquire lock.
}
defer func() { _ = dbm.Unlock(context.Background()) }()
// Do "critical section" application logic here...

Example (WithTimeout)

In this scenario we wait for a maximum of 10 seconds to acquire the lock.

// In the func that needs to protect a section of code...
var dbm Mutex // Mutex normally acquired via New()
// Use background context here but could have gotten context from an input parameter.
ctx := context.Background()
// Wait until the lock is acquired or 10 seconds passes.
deadlineCtx, cancelFunc := context.WithDeadline(ctx, time.Now().Add(10*time.Second))
defer cancelFunc()
_, err := dbm.Lock(deadlineCtx)
if err == context.DeadlineExceeded {
	// Timed out waiting to acquire lock.
} else if err != nil {
	// Some other error when attempting to acquire lock.
}
defer func() { _ = dbm.Unlock(context.Background()) }()
// Do "critical section" application logic here...

func (*Mutex) Unlock

func (m *Mutex) Unlock(ctx context.Context) error

The Unlock operation attempts to release the Mutex by updating the "locked" column in the underlying database table. If unable to update the row, Lock will poll attempting to release the mutex by updating the row. Any database related errors that occur during Unlock are reported via an ErrorNotifier (see WithErrorNotifier) but are typically ignored in order to complete the Unlock operation. Unlock can be timed out by using a ctx parameter that has a deadline (see examples). The poll interval and ability to fail fast after a database error can be controlled via MutexOption passed to New. Because polling is used even after database generated errors, Unlock will typically not return unless the Mutex is released or the ctx expires.

type MutexOption

type MutexOption func(options *mutexOptions)

MutexOption is used to customize Mutex behaviour.

func WithCreateMissingTable

func WithCreateMissingTable(b bool) MutexOption

WithCreateMissingTable allows customization of table creation if it does not exist. Pass false in order to override the default behaviour and not create a missing table.

func WithDriver

func WithDriver(d driver.Driver) MutexOption

WithDriver allows explict setting of the Driver. Normally, mysql or postgres is automatically detected.

func WithErrorNotifier

func WithErrorNotifier(f ErrorNotifier) MutexOption

WithErrorNotifier allows customization of the ErrorNotifier that will be used when acquiring the lock, releasing the lock and refreshing the lock. This is a convenience call that sets all three ErrorNotifiers to the same value

func WithExpiration

func WithExpiration(t time.Duration) MutexOption

WithExpiration allows customization of the duration after which a lock will expire if it is not refreshed. Normally, as long as the locking process continues to run and can reach the database, refreshes happen automatically. If a process cannot update the database, then the lock will expire after t Duration.

func WithFailFast

func WithFailFast(b bool) MutexOption

WithFailFast, when passed true, causes (Un)Lock calls to immediately return if unable to interact with the underlying database. Normally (Un)Lock operations will retry until the passed context expires.

func WithLockErrorNotifier

func WithLockErrorNotifier(f ErrorNotifier) MutexOption

WithLockErrorNotifier allows customization of the ErrorNotifier that will be used when acquiring the lock in the Lock call.

func WithMutexName

func WithMutexName(name string) MutexOption

WithMutexName uses name as the name of lock. Normally this option should be passed in order to define an application-specific scope of locking that can be used across multiple processes. Examples:

WithMutexName("order 234")
WithMutexName("customer 9854")

If len(name) > driver.MaxMutexNameLength name will be silently truncated.

func WithMutexTableName

func WithMutexTableName(name string) MutexOption

WithMutexTableName allows customization of the lock table name. Normally this is not needed since multiple named locks can be used in the same lock table.

func WithPollInterval

func WithPollInterval(d time.Duration) MutexOption

WithPollInterval will change the interval at which (Un)Lock tries to acquire or release the lock. Normally, DefaultPollInterval is used. Be careful when using a small poll interval because you can potentially cause increased load on the database server. Duration must be > 0 or else it will be set to DefaultPollInterval.

func WithRefresh

func WithRefresh(t time.Duration) MutexOption

WithRefresh allows customization of the interval at which lock refreshes are performed.

func WithRefreshErrorNotifier

func WithRefreshErrorNotifier(f ErrorNotifier) MutexOption

WithRefreshErrorNotifier allows customization of the ErrorNotifier that will be used when keeping the Mutex refreshed.

func WithUnlockErrorNotifier

func WithUnlockErrorNotifier(f ErrorNotifier) MutexOption

WithUnlockErrorNotifier allows customization of the ErrorNotifier that will be used when releasing the lock in the Unlock call.