mutex

README

This mutex workflow demos an ability to lock/unlock a particular resource within a particular Temporal namespace so that other workflows within the same namespace would wait until a resource lock is released. This is useful when we want to avoid race conditions or parallel mutually exclusive operations on the same resource.

One way of coordinating parallel processing is to use Temporal signals with SignalWithStartWorkflow and make sure signals are getting processed sequentially, however the logic might become too complex if we need to lock two or more resources at the same time. Mutex workflow pattern can simplify that.

This example enqueues two long running SampleWorkflowWithMutex workflows in parallel. And each of the workflows has a Mutex section. When SampleWorkflowWithMutex reaches Mutex section, it starts a mutex workflow via local activity, and blocks until acquire-lock-event is received. Once acquire-lock-event is received, it enters critical section, and finally releases the lock once processing is over by sending releaseLock a signal to the MutexWorkflow.

Steps to run this sample:

1. You need a Temporal service running. See details in README.md
2. Run the following command to start the worker

go run mutex/worker/main.go

1. Run the following command to start the example

go run mutex/starter/main.go

You should see that second workflow critical section is executed when first workflow critical operation is finished.

Documentation

Index

• Constants
• func MockMutexLock(env *testsuite.TestWorkflowEnvironment, resourceID string, mockError error)
• func MutexWorkflow(ctx workflow.Context, namespace string, resourceID string, ...) error
• func SampleWorkflowWithMutex(ctx workflow.Context, resourceID string) error
• func SignalWithStartMutexWorkflowActivity(ctx context.Context, namespace string, resourceID string, ...) (*workflow.Execution, error)
• type ContextKey
• type Mutex
  • func NewMutex(currentWorkflowID string, lockNamespace string) *Mutex
  • func (s *Mutex) Lock(ctx workflow.Context, resourceID string, unlockTimeout time.Duration) (UnlockFunc, error)
• type UnlockFunc

Constants

const (
	// AcquireLockSignalName signal channel name for lock acquisition
	AcquireLockSignalName = "acquire-lock-event"
	// RequestLockSignalName channel name for request lock
	RequestLockSignalName = "request-lock-event"

	ClientContextKey ContextKey = "Client"
)

Functions

func MockMutexLock

func MockMutexLock(env *testsuite.TestWorkflowEnvironment, resourceID string, mockError error)

MockMutexLock stubs mutex.Lock call

func MutexWorkflow

func MutexWorkflow(
	ctx workflow.Context,
	namespace string,
	resourceID string,
	unlockTimeout time.Duration,
) error

MutexWorkflow used for locking a resource

func SampleWorkflowWithMutex

func SampleWorkflowWithMutex(
	ctx workflow.Context,
	resourceID string,
) error

func SignalWithStartMutexWorkflowActivity

func SignalWithStartMutexWorkflowActivity(
	ctx context.Context,
	namespace string,
	resourceID string,
	senderWorkflowID string,
	unlockTimeout time.Duration,
) (*workflow.Execution, error)

SignalWithStartMutexWorkflowActivity ...

Types

type ContextKey

type ContextKey string

type Mutex

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

func NewMutex

func NewMutex(currentWorkflowID string, lockNamespace string) *Mutex

NewMutex initializes mutex

func (*Mutex) Lock

func (s *Mutex) Lock(ctx workflow.Context,
	resourceID string, unlockTimeout time.Duration) (UnlockFunc, error)

Lock - locks mutex

type UnlockFunc

type UnlockFunc func() error