sync

Documentation

Index

• type ShardJob
• type ShardWorkerPool
  • func NewShardWorkerPool(size int, numShards int, workerFunc WorkerFunc) (ShardWorkerPool, error)
• type WorkerFunc
• type WorkerPool
  • func NewWorkerPool(size int) WorkerPool

Types

type ShardJob

type ShardJob interface {
	// Source get the job identifier for sharding.
	Source() uint32
}

ShardJob represents a job with its shard source.

type ShardWorkerPool

type ShardWorkerPool interface {
	// Init initializes the pool.
	Init()

	// Shard get the real shard of giving source, this may helps in case like global object accessing.
	Shard(source uint32) uint32

	// Offer puts the job into the corresponding shard and execute it.
	Offer(job ShardJob, block bool)
}

ShardWorkerPool provides a pool for goroutines, the actual goroutines themselves are assumed permanent running and waiting for the incoming jobs from job channel. Its behaviour is specified by the :ref:`WorkerFunc<sync.WorkerFunc>`.

In order to prevent excessive lock contention, the ShardWorkerPool also implements sharding of its underlying worker jobs. Source value, which can be used to calculate the actual shard(goroutine), is required when jobs are committed to the ShardWorkerPool. Jobs in the same shard are serial executed in FIFO order.

shard goroutine respawn is performed while panic during the WorkerFunc execution.

func NewShardWorkerPool

func NewShardWorkerPool(size int, numShards int, workerFunc WorkerFunc) (ShardWorkerPool, error)

NewShardWorkerPool creates a new shard worker pool.

type WorkerFunc

type WorkerFunc func(shard int, jobCh <-chan interface{})

WorkerFunc is called by the goroutine of the ShardWorkerPool and assumed never return in normal case.

type WorkerPool

type WorkerPool interface {

	// Schedule try to acquire pooled worker goroutine to execute the specified task,
	// this method would block if no worker goroutine is available
	Schedule(task func())

	// Schedule try to acquire pooled worker goroutine to execute the specified task first,
	// but would not block if no worker goroutine is available. A temp goroutine will be created for task execution.
	ScheduleAlways(task func())

	ScheduleAuto(task func())
}

WorkerPool provides a pool for goroutines

func NewWorkerPool

func NewWorkerPool(size int) WorkerPool

NewWorkerPool create a worker pool