sync

Documentation

Index

• func DecodeLockName(name string) (*lockName, error)
• func NewMutex(name string, nextWorkflow NextWorkflow) *prioritySemaphore
• func NewSemaphore(name string, limit int, nextWorkflow NextWorkflow, lockType string) *prioritySemaphore
• type GetSyncLimit
• type IsWorkflowDeleted
• type Key
• type Manager
  • func NewLockManager(getSyncLimit GetSyncLimit, nextWorkflow NextWorkflow, ...) *Manager
  • func (sm *Manager) CheckWorkflowExistence()
  • func (sm *Manager) Initialize(ctx context.Context, wfs []wfv1.Workflow)
  • func (sm *Manager) Release(ctx context.Context, wf *wfv1.Workflow, nodeName string, ...)
  • func (sm *Manager) ReleaseAll(wf *wfv1.Workflow) bool
  • func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName string, ...) (bool, bool, string, string, error)
• type NextWorkflow
• type QueueFunc
• type SyncLevelType
• type Throttler
  • func NewMultiThrottler(parallelism int, namespaceParallelismLimit int, queue QueueFunc) Throttler

Functions

func DecodeLockName

func DecodeLockName(name string) (*lockName, error)

func NewMutex

func NewMutex(name string, nextWorkflow NextWorkflow) *prioritySemaphore

NewMutex creates a size 1 semaphore

func NewSemaphore

func NewSemaphore(name string, limit int, nextWorkflow NextWorkflow, lockType string) *prioritySemaphore

Types

type GetSyncLimit

type GetSyncLimit func(string) (int, error)

type IsWorkflowDeleted

type IsWorkflowDeleted func(string) bool

type Key

type Key = string

type Manager

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

func NewLockManager

func NewLockManager(getSyncLimit GetSyncLimit, nextWorkflow NextWorkflow, isWFDeleted IsWorkflowDeleted) *Manager

func (*Manager) CheckWorkflowExistence

func (sm *Manager) CheckWorkflowExistence()

func (*Manager) Initialize

func (sm *Manager) Initialize(ctx context.Context, wfs []wfv1.Workflow)

func (*Manager) Release

func (sm *Manager) Release(ctx context.Context, wf *wfv1.Workflow, nodeName string, syncRef *wfv1.Synchronization)

func (*Manager) ReleaseAll

func (sm *Manager) ReleaseAll(wf *wfv1.Workflow) bool

func (*Manager) TryAcquire

func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName string, syncLockRef *wfv1.Synchronization) (bool, bool, string, string, error)

TryAcquire tries to acquire the lock from semaphore. It returns status of acquiring a lock , status of Workflow status updated, waiting message if lock is not available, the failed lock, and any error encountered

type NextWorkflow

type NextWorkflow func(string)

type QueueFunc

type QueueFunc func(Key)

type SyncLevelType

type SyncLevelType int

const (
	WorkflowLevel SyncLevelType = 1
	TemplateLevel SyncLevelType = 2
	ErrorLevel    SyncLevelType = 3
)

type Throttler

type Throttler interface {
	Init(wfs []wfv1.Workflow) error
	Add(key Key, priority int32, creationTime time.Time)
	// Admit returns if the item should be processed.
	Admit(key Key) bool
	// Remove notifies throttler that item processing is no longer needed
	Remove(key Key)
}

Throttler allows the controller to limit number of items it is processing in parallel. Items are processed in priority order, and one processing starts, other items (including higher-priority items) will be kept pending until the processing is complete. Implementations should be idempotent.

func NewMultiThrottler

func NewMultiThrottler(parallelism int, namespaceParallelismLimit int, queue QueueFunc) Throttler

NewMultiThrottler creates a new multi throttler for throttling both namespace and global parallelism, a parallelism value of zero disables throttling