Version: v3.0.1 Latest Latest

This package is not in the latest version of its module.

Go to latest
Published: Feb 23, 2021 License: Apache-2.0 Imports: 9 Imported by: 0




This section is empty.


This section is empty.


This section is empty.


type GetSyncLimit

type GetSyncLimit func(string) (int, error)

type IsWorkflowDeleted

type IsWorkflowDeleted func(string) bool

type LockKind

type LockKind string
const (
	LockKindConfigMap LockKind = "ConfigMap"
	LockKindMutex     LockKind = "Mutex"

type LockName

type LockName struct {
	Namespace    string
	ResourceName string
	Key          string
	Kind         LockKind

func DecodeLockName

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

func GetLockName

func GetLockName(sync *v1alpha1.Synchronization, namespace string) (*LockName, error)

func NewLockName

func NewLockName(namespace, resourceName, lockKey string, kind LockKind) *LockName

func (*LockName) EncodeName

func (ln *LockName) EncodeName() string

func (*LockName) Validate

func (ln *LockName) Validate() error

func (*LockName) ValidateEncoding

func (ln *LockName) ValidateEncoding(encoding string) 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 (cm *Manager) CheckWorkflowExistence()

func (*Manager) Initialize

func (cm *Manager) Initialize(wfs []wfv1.Workflow)

func (*Manager) Release

func (cm *Manager) Release(wf *wfv1.Workflow, nodeName string, syncRef *wfv1.Synchronization)

func (*Manager) ReleaseAll

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

func (*Manager) TryAcquire

func (cm *Manager) TryAcquire(wf *wfv1.Workflow, nodeName string, syncLockRef *wfv1.Synchronization) (bool, bool, 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 and any error encountered

type NextWorkflow

type NextWorkflow func(string)

type PriorityMutex

type PriorityMutex struct {
	// contains filtered or unexported fields

func NewMutex

func NewMutex(name string, nextWorkflow NextWorkflow) *PriorityMutex

NewMutex creates new mutex lock object name of the mutex callbackFunc is a release notification function.

type PrioritySemaphore

type PrioritySemaphore struct {
	// contains filtered or unexported fields

func NewSemaphore

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

type Semaphore

type Semaphore interface {
	// contains filtered or unexported methods

type Throttler

type Throttler interface {
	Add(key string, priority int32, creationTime time.Time)
	// Admin returns if the item should be processed.
	Admit(key string) bool
	// Remove notifies throttler that item processing is no longer needed
	Remove(key string)

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 NewThrottler

func NewThrottler(parallelism int, queue func(key string)) Throttler

NewThrottler returns a throttle that only runs `parallelism` items at once. When an item may need processing, `queue` is invoked.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL