ytlock

Documentation

Overview

Package ytlock is high level interface for yt lock.

Index

• Variables
• func AbortExclusiveLock(ctx context.Context, yc yt.Client, path ypath.Path) error
• func RunLocked(ctx context.Context, lock *Lock, job Job, notify Notify) error
• type Job
• type Lock
  • func NewLock(yc yt.Client, path ypath.Path) (l *Lock)
  • func NewLockOptions(yc yt.Client, path ypath.Path, opts Options) (l *Lock)
  • func (l *Lock) Acquire(ctx context.Context) (lost <-chan struct{}, err error)
  • func (l *Lock) AcquireTx(ctx context.Context) (lockTx yt.Tx, err error)
  • func (l *Lock) IsLocked() bool
  • func (l *Lock) Release(ctx context.Context) error
• type Notify
• type Options
• type WinnerTx
  • func FindConflictWinner(err error) *WinnerTx

Variables

var ErrLockLost = xerrors.NewSentinel("lock lost")

ErrLockLost is an error propagated to notify in RunLocked when an acquired lock is lost.

Functions

func AbortExclusiveLock

func AbortExclusiveLock(ctx context.Context, yc yt.Client, path ypath.Path) error

AbortExclusiveLock cancels transaction under exclusive lock specified by the path.

There can not be more than one exclusive lock acquired for the path at the time.

In case if node has not acquired an exclusive lock nil is returned.

func RunLocked

func RunLocked(ctx context.Context, lock *Lock, job Job, notify Notify) error

RunLocked infinitely (with constant back-off) reacquires the lock and reruns the job.

Stops when the job finishes with no error or when the context is closed.

Calls notify on each failed attempt. In case of lost lock notify will have an error that wraps ErrLockLost.

Types

type Job

type Job func(ctx context.Context) error

Job is function cancellable via ctx.

type Lock

type Lock struct {
	Path    ypath.Path
	Options Options
	Yc      yt.Client
	// contains filtered or unexported fields
}

Lock object represents cypress lock.

func NewLock

func NewLock(yc yt.Client, path ypath.Path) (l *Lock)

NewLock creates Lock object using default Options.

func NewLockOptions

func NewLockOptions(yc yt.Client, path ypath.Path, opts Options) (l *Lock)

NewLockOptions creates Lock object with specified options.

func (*Lock) Acquire

func (l *Lock) Acquire(ctx context.Context) (lost <-chan struct{}, err error)

Acquire acquires cypress lock.

Returned lost channel is closed when lock is lost because of some external event, e.g. transaction being aborted remotely, network partition or coordination service downtime.

If lock is lost, there is no need to call Release explicitly.

Lock is automatically released when provided ctx is canceled.

func (*Lock) AcquireTx

func (l *Lock) AcquireTx(ctx context.Context) (lockTx yt.Tx, err error)

AcquireTx is the same as Acquire, but returns new tx that is holding the lock.

func (*Lock) IsLocked

func (l *Lock) IsLocked() bool

IsLocked returns true if lock is in acquired state.

func (*Lock) Release

func (l *Lock) Release(ctx context.Context) error

Release releases distributed lock by aborting transaction.

Error might indicate, that we failed to receive acknowledgement from the master, but lock will be released eventually.

type Notify

type Notify func(error, time.Duration)

Notify is a notify-on-error function.

It receives a job error and backoff delay if the job failed.

type Options

type Options struct {
	// Create map node if path is missing.
	CreateIfMissing bool
	LockMode        yt.LockMode
	LockChild       string

	TxAttributes map[string]any
}

type WinnerTx

type WinnerTx struct {
	ID        yt.TxID       `yson:"id"`
	Owner     string        `yson:"owner"`
	StartTime yson.Time     `yson:"start_time"`
	Timeout   yson.Duration `yson:"duration"`
}

func FindConflictWinner

func FindConflictWinner(err error) *WinnerTx

FindConflictWinner returns information about a process holding the lock that caused the conflict.