syncx

README

syncx

More Advanced sync primitives.

Implemented some Go sync primitives.

Token

provides token implementation.

Only the one thats owns the Token can do stuff and then it can handoffs the token to others.

Batch

provides batch implementation.

It is like errgroup and can return all errors results of each task.

Any

provides partial batch implementation.

You can wait some tasks have finished and returns.

Stentor

provides pub-sub(observer) pattern.

Examples

use syncx to solve leetcode concurrency problems

• 1115 Print FooBar Alternately: solution
• 1116 Print Zero Even Odd: solution
• 1195 Fizz Buzz Multithreaded: solution

Other advanced sync primitives

• singleflight: provides a duplicate function call suppression
• errgroup: provides synchronization, error propagation, and Context cancelation for groups of goroutines working on subtasks of a common task
• semaphore: provides a weighted semaphore implementation

Documentation

Index

• type Any
  • func NewAny(ctx context.Context, n int, least int) *Any
  • func NewAnyWithParallel(ctx context.Context, n int, p uint32, least int) *Any
  • func (b *Any) Go(i int, task func(ctx context.Context) error)
  • func (b *Any) Wait() (errCount uint64, errs []error)
• type Batch
  • func NewBatch(ctx context.Context, n int) *Batch
  • func NewBatchWithParallel(ctx context.Context, n int, p uint32) *Batch
  • func (b *Batch) Go(i int, task func(ctx context.Context) error)
  • func (b *Batch) Wait() (errCount uint64, errs []error)
• type G
• type Stentor
  • func NewStentor(bufSize int) *Stentor
  • func (s *Stentor) Broadcast(g G)
  • func (s *Stentor) Count() int
  • func (s *Stentor) Reset()
  • func (s *Stentor) Subscribe() <-chan G
  • func (s *Stentor) Unsubscribe(sr <-chan G) bool
• type Token
  • func NewToken(n int) *Token
  • func (t *Token) Accquire(ctx context.Context, selfIdx int) error
  • func (t *Token) Handoff(ctx context.Context, otherIdx int) error
  • func (t *Token) Rand(ctx context.Context) error

Types

type Any

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

Any is used to execute batch tasks anf wait some tasks finish. It can't be reused.

func NewAny

func NewAny(ctx context.Context, n int, least int) *Any

NewAny creates a batch to execute n tasks and wait least tasks finish.

func NewAnyWithParallel

func NewAnyWithParallel(ctx context.Context, n int, p uint32, least int) *Any

NewAnyWithParallel creates a batch to execute n tasks in max p goroutine and wait least tasks finish.

func (*Any) Go

func (b *Any) Go(i int, task func(ctx context.Context) error)

Go executes the ith task. Only call this method for one index one time.

func (*Any) Wait

func (b *Any) Wait() (errCount uint64, errs []error)

Wait blocks until least tasks have been done or canceled. if some tasks return errors, this method returns the count of errors and result of each task.

Must be called after call Go methods.

type Batch

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

Batch is used to execute batch tasks. It can't be reused.

func NewBatch

func NewBatch(ctx context.Context, n int) *Batch

NewBatch creates a batch to execute n tasks.

func NewBatchWithParallel

func NewBatchWithParallel(ctx context.Context, n int, p uint32) *Batch

NewBatchWithParallel creates a batch to execute n tasks in max p goroutine.

func (*Batch) Go

func (b *Batch) Go(i int, task func(ctx context.Context) error)

Go executes the ith task. Only call this method for one index one time.

func (*Batch) Wait

func (b *Batch) Wait() (errCount uint64, errs []error)

Wait blocks until all tasks have been done or canceled. if some tasks return errors, this method returns the count of errors and result of each task.

Must be called after call Go methods.

type G

type G interface{}

G represents intefrace{}. Since Go has not implemented generic so we use it to mock generic.

type Stentor

type Stentor struct {
	FailureCallback func(<-chan G, G)
	// contains filtered or unexported fields
}

Stentor implments pub-sub pattern.

func NewStentor

func NewStentor(bufSize int) *Stentor

NewStentor creates a Stentor.

func (*Stentor) Broadcast

func (s *Stentor) Broadcast(g G)

Broadcast broadcasts an event.

func (*Stentor) Count

func (s *Stentor) Count() int

Count returns count of subscribers.

func (*Stentor) Reset

func (s *Stentor) Reset()

Reset clears all subscribers.

func (*Stentor) Subscribe

func (s *Stentor) Subscribe() <-chan G

Subscribe subscribes this Stentor.

func (*Stentor) Unsubscribe

func (s *Stentor) Unsubscribe(sr <-chan G) bool

Unsubscribe unsubscribes this Stentor.

type Token

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

Token is a token shared by n participants, and only one participant can do stuff at the same time. Participants must accquire the otoen before they do stuff. They will be blocked if the token by another participant. After a participant has finished its work, it can handoff the token to an determined participant.

func NewToken

func NewToken(n int) *Token

NewToken creates a new Token with n participants.

func (*Token) Accquire

func (t *Token) Accquire(ctx context.Context, selfIdx int) error

Accquire accquires the token by current participant(selfIdx).

func (*Token) Handoff

func (t *Token) Handoff(ctx context.Context, otherIdx int) error

Handoff passes the token to another participant(otherIdx).

func (*Token) Rand

func (t *Token) Rand(ctx context.Context) error

Rand gives the token to a participant randomly.