recipes

Documentation

Index

• Constants
• type AfterConnectionEstablished
  • func (c *AfterConnectionEstablished) Future() *promise.Future
• type AtomicStats
• type AtomicValue
• type ChildrenSorter
  • func (s ChildrenSorter) Len() int
  • func (s ChildrenSorter) Less(i, j int) bool
  • func (s ChildrenSorter) Swap(i, j int)
• type DistributedAtomicNumber
• type DistributedAtomicValue
  • func NewDistributedAtomicValue(client curator.CuratorFramework, path string, retryPolicy curator.RetryPolicy) (DistributedAtomicValue, error)
  • func NewDistributedAtomicValueWithLock(client curator.CuratorFramework, path string, retryPolicy curator.RetryPolicy, ...) (DistributedAtomicValue, error)
• type InterProcessLock
• type InterProcessMutex
  • func NewInterProcessMutex(client curator.CuratorFramework, path string) (*InterProcessMutex, error)
  • func NewInterProcessMutexWithDriver(client curator.CuratorFramework, path string, driver LockInternalsDriver) (*InterProcessMutex, error)
  • func (m *InterProcessMutex) Acquire() (bool, error)
  • func (m *InterProcessMutex) AcquireTimeout(expires time.Duration) (bool, error)
  • func (m *InterProcessMutex) IsAcquiredInThisProcess() bool
  • func (m *InterProcessMutex) Release() error
• type LockInternalsDriver
• type LockInternalsSorter
• type PredicateResults
• type PromotedToLock
• type Revocable
• type RevocationListener
• type StandardLockInternalsDriver
  • func NewStandardLockInternalsDriver() *StandardLockInternalsDriver
  • func (d *StandardLockInternalsDriver) CreatesTheLock(client curator.CuratorFramework, path string, lockNodeBytes []byte) (string, error)
  • func (d *StandardLockInternalsDriver) FixForSorting(str, lockName string) string
  • func (d *StandardLockInternalsDriver) GetsTheLock(client curator.CuratorFramework, children []string, sequenceNodeName string, ...) (*PredicateResults, error)

Constants

const LockPrefix = "lock-"

Types

type AfterConnectionEstablished

type AfterConnectionEstablished struct {
	Client  curator.CuratorFramework
	Timeout time.Duration
}

Utility class to allow execution of logic once a ZooKeeper connection becomes available.

func (*AfterConnectionEstablished) Future

func (c *AfterConnectionEstablished) Future() *promise.Future

Spawns a new new background thread that will block until a connection is available and then execute the 'runAfterConnection' logic

type AtomicStats

type AtomicStats struct {
	//  the number of optimistic locks used to perform the operation
	OptimisticTries int

	// the number of mutex locks used to perform the operation
	PromotedTries int

	// the time spent trying the operation with optimistic locks
	OptimisticTime time.Duration

	// the time spent trying the operation with mutex locks
	PromotedTime time.Duration
}

Debugging stats about operations

type AtomicValue

type AtomicValue interface {
	// MUST be checked.
	// Returns true if the operation succeeded. If false is returned,
	// the operation failed and the atomic was not updated.
	Succeeded() bool

	// Returns the value of the counter prior to the operation
	PreValue() []byte

	// Returns the value of the counter after to the operation
	PostValue() []byte

	// Returns debugging stats about the operation
	Stats() *AtomicStats
}

Abstracts a value returned from one of the Atomics

type ChildrenSorter

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

func (ChildrenSorter) Len

func (s ChildrenSorter) Len() int

func (ChildrenSorter) Less

func (s ChildrenSorter) Less(i, j int) bool

func (ChildrenSorter) Swap

func (s ChildrenSorter) Swap(i, j int)

type DistributedAtomicNumber

type DistributedAtomicNumber interface {
	// Add 1 to the current value and return the new value information.
	// Remember to always check AtomicValue.Succeeded().
	Increment() (AtomicValue, error)

	// Subtract 1 from the current value and return the new value information.
	// Remember to always check AtomicValue.Succeeded().
	Decrement() (AtomicValue, error)

	// Add delta to the current value and return the new value information.
	// Remember to always check AtomicValue.Succeeded().
	Add(delta []byte) (AtomicValue, error)

	// Subtract delta from the current value and return the new value information.
	// Remember to always check AtomicValue.Succeeded().
	Subtract(delta []byte) (AtomicValue, error)
}

type DistributedAtomicValue

type DistributedAtomicValue interface {
	// Returns the current value of the counter.
	Get() (AtomicValue, error)

	// Atomically sets the value to the given updated value
	// if the current value == the expected value.
	// Remember to always check AtomicValue.Succeeded().
	CompareAndSet(expectedValue, newValue []byte) (AtomicValue, error)

	// Attempt to atomically set the value to the given value.
	// Remember to always check AtomicValue.Succeeded().
	TrySet(newValue []byte) (AtomicValue, error)

	// Forcibly sets the value of the counter without any guarantees of atomicity.
	ForceSet(newValue []byte) error

	// Atomic values are initially set to the equivalent of <code>NULL</code> in a database.
	// Use this method to initialize the value.
	// The value will be set if and only iff the node does not exist.
	Initialize(value []byte) (bool, error)
}

func NewDistributedAtomicValue

func NewDistributedAtomicValue(client curator.CuratorFramework, path string, retryPolicy curator.RetryPolicy) (DistributedAtomicValue, error)

func NewDistributedAtomicValueWithLock

func NewDistributedAtomicValueWithLock(client curator.CuratorFramework, path string, retryPolicy curator.RetryPolicy, promotedToLock *PromotedToLock) (DistributedAtomicValue, error)

type InterProcessLock

type InterProcessLock interface {
	// Acquire the mutex - blocking until it's available.
	// Each call to acquire must be balanced by a call to Release()
	Acquire() (bool, error)

	// Acquire the mutex - blocks until it's available or the given time expires.
	AcquireTimeout(expires time.Duration) (bool, error)

	// Perform one release of the mutex.
	Release() error

	// Returns true if the mutex is acquired by a go-routine in this process
	IsAcquiredInThisProcess() bool
}

type InterProcessMutex

type InterProcessMutex struct {
	LockNodeBytes []byte
	// contains filtered or unexported fields
}

A re-entrant mutex that works across processes. Uses Zookeeper to hold the lock. All processes that use the same lock path will achieve an inter-process critical section. Further, this mutex is "fair" - each user will get the mutex in the order requested (from ZK's point of view)

func NewInterProcessMutex

func NewInterProcessMutex(client curator.CuratorFramework, path string) (*InterProcessMutex, error)

func NewInterProcessMutexWithDriver

func NewInterProcessMutexWithDriver(client curator.CuratorFramework, path string, driver LockInternalsDriver) (*InterProcessMutex, error)

func (*InterProcessMutex) Acquire

func (m *InterProcessMutex) Acquire() (bool, error)

func (*InterProcessMutex) AcquireTimeout

func (m *InterProcessMutex) AcquireTimeout(expires time.Duration) (bool, error)

func (*InterProcessMutex) IsAcquiredInThisProcess

func (m *InterProcessMutex) IsAcquiredInThisProcess() bool

func (*InterProcessMutex) Release

func (m *InterProcessMutex) Release() error

type LockInternalsDriver

type LockInternalsDriver interface {
	LockInternalsSorter

	GetsTheLock(client curator.CuratorFramework, children []string, sequenceNodeName string, maxLeases int) (*PredicateResults, error)

	CreatesTheLock(client curator.CuratorFramework, path string, lockNodeBytes []byte) (string, error)
}

type LockInternalsSorter

type LockInternalsSorter interface {
	FixForSorting(str, lockName string) string
}

type PredicateResults

type PredicateResults struct {
	GetsTheLock bool
	PathToWatch string
}

type PromotedToLock

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

type Revocable

type Revocable interface {
	// Make the lock revocable.
	// Your listener will get called when another process/thread wants you to release the lock. Revocation is cooperative.
	MakeRevocable(listener RevocationListener)
}

Specifies locks that can be revoked

type RevocationListener

type RevocationListener interface {
	// Called when a revocation request has been received.
	// You should release the lock as soon as possible. Revocation is cooperative.
	RevocationRequested(forLock InterProcessMutex)
}

type StandardLockInternalsDriver

type StandardLockInternalsDriver struct{}

func NewStandardLockInternalsDriver

func NewStandardLockInternalsDriver() *StandardLockInternalsDriver

func (*StandardLockInternalsDriver) CreatesTheLock

func (d *StandardLockInternalsDriver) CreatesTheLock(client curator.CuratorFramework, path string, lockNodeBytes []byte) (string, error)

func (*StandardLockInternalsDriver) FixForSorting

func (d *StandardLockInternalsDriver) FixForSorting(str, lockName string) string

func (*StandardLockInternalsDriver) GetsTheLock

func (d *StandardLockInternalsDriver) GetsTheLock(client curator.CuratorFramework, children []string, sequenceNodeName string, maxLeases int) (*PredicateResults, error)