concurrency

README

concurrency

go concurrency library

• 

Installation

go get -u github.com/iTrellis/concurrency

Usage

concurrency repo

// ConcurrencyRepo functions for go routings to run tasks
type ConcurrencyRepo interface {
	// Invoke tasks: task must be functions
	Invoke(tasks []interface{}) ([]Runner, error)
	InvokeDuration(tasks []interface{}, timeout time.Duration) ([]Runner, error)
}

new and input a namespace's transaction

	c := concurrency.New(100)

	f := func(i int) (n int) {
		fmt.Println(i, time.Now())
		return i
	}

	var tasks []interface{}
	for i := 0; i < 50; i++ {
		tasks = append(tasks, f)
	}

	runners, err := c.Invoke(tasks)

	for i := 0; i < 50; i++ {
		fmt.Println(runners[i].Get.MapV(func(n int){fmt.Println("%3.d", n)}))
	}

Documentation

Index

• Variables
• type Repo
  • func New(number int) Repo
• type Result
  • func (p *Result) MapV(params interface{}) error
• type Runner
• type RunnerStack
  • func (p *RunnerStack) Length() int64
  • func (p *RunnerStack) Pop() Runner
  • func (p *RunnerStack) PopAll() (rs []Runner)
  • func (p *RunnerStack) Push(r Runner)
• type Task
  • func (p *Task) Cancel() bool
  • func (p *Task) Get() (result *Result)
  • func (p *Task) GetDuration(d time.Duration) (result *Result)
  • func (p *Task) IsCancelled() bool
  • func (p *Task) IsDone() bool
  • func (p *Task) IsRunning() bool
  • func (p *Task) Run()

Variables

var (
	ErrInterruptedAddTask = errors.New("interrupted to add task")
	ErrFailedGetResult    = errors.New("failed get result")
	ErrLastValueIsError   = errors.New("last value is error")
)

concurrency errors

Types

type Repo

type Repo interface {
	// Invoke tasks: task must be functions
	Invoke(tasks []interface{}) ([]Runner, error)
	InvokeDuration(tasks []interface{}, timeout time.Duration) ([]Runner, error)
}

Repo functions for go routings to run tasks

func New

func New(number int) Repo

New return a concurrency pool with number

type Result

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

Result run task return values

func (*Result) MapV

func (p *Result) MapV(params interface{}) error

MapV map values into function parmas

type Runner

type Runner interface {
	// get execute result
	Get() *Result
	// get execute result in duration
	GetDuration(d time.Duration) (result *Result)
	// judge runner is done
	IsDone() bool
	// judge runner is running
	IsRunning() bool
	// cancel runner
	Cancel() bool
	// is cancelled
	IsCancelled() bool
	// runner to run
	Run()
}

Runner functions for a runner

type RunnerStack

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

RunnerStack runner stack

func (*RunnerStack) Length

func (p *RunnerStack) Length() int64

Length stack length

func (*RunnerStack) Pop

func (p *RunnerStack) Pop() Runner

Pop pop a runner from stack

func (*RunnerStack) PopAll

func (p *RunnerStack) PopAll() (rs []Runner)

PopAll pop all runners from stack

func (*RunnerStack) Push

func (p *RunnerStack) Push(r Runner)

Push push a runner to stack

type Task

type Task struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Task task

func (*Task) Cancel

func (p *Task) Cancel() bool

Cancel cancel a task

func (*Task) Get

func (p *Task) Get() (result *Result)

Get get result from context

func (*Task) GetDuration

func (p *Task) GetDuration(d time.Duration) (result *Result)

GetDuration get result

func (*Task) IsCancelled

func (p *Task) IsCancelled() bool

IsCancelled judge a task if was cancelled return true

func (*Task) IsDone

func (p *Task) IsDone() bool

IsDone judge a task if was done return true

func (*Task) IsRunning

func (p *Task) IsRunning() bool

IsRunning judge a task if running return true

func (*Task) Run

func (p *Task) Run()

Run run task