parallel

README

parallel

A parallel processing package for Go.

Usage

Two structures implement the Process interface; FixedProcess and VariableProcess. Both perform a set of parallel operations on either a fixed or varying number of goroutines.

FixedProcess

FixedProcess types execute their set of operations on a fixed number of goroutines specified upon initialization.

// Create a process with two goroutines.
p := parallel.NewFixedProcess(2)

// Execute 100 operations in parallel.
p.Execute(100, func (i int) {
  // Perform the ith operation.
})

VariableProcess

VariableProcess types execute their set of operations on a variable number of goroutines by utilizing a PID control loop to maximize CPU throughput. You configure the PID controller by creating a ControllerConfiguration struct and passing to the NewVariableProcess function.

// Create a variable process with optimization interval, initial routine count,
// max goroutine count, and PID controller configuration.
c := NewControllerConfiguration(2.0, 0.0, 1.0, 0.1, 1.0)
p := parallel.NewVariableProcess(500 * time.Millisecond, 1, 2 * runtime.NumCPU(), c, false)

// Execute 100 operations in parallel.
p.Execute(100, func(i int) {
  // Perform the ith operation.
})

You can't change the number of goroutines directly in an variable process, but you can modify its optimization parameters while it's executing.

// Execute the process on its own goroutine
go p.Execute(100, func(i int) {
  // Perform the ith operation.
})

// Modify optimization parameters while the process is executing...
p.SetOptimizationInterval(time.Second)
p.SetMaxRoutines(50)

c := p.GetControllerConfiguration()
// Modify config...
p.SetControllerConfiguration(c)

Tuning the PID Controller

A variable process has four probes that are activated by setting the probeController parameter to true upon initialization. They monitor and keep a record of the following values.

• CPU throughput
• PID input error
• PID output signal
• Number of goroutines

// Create a variable process with probeController set to true.
c := NewControllerConfiguration(2.0, 0.0, 1.0, 0.1, 1.0)
p := parallel.NewVariableProcess(500 * time.Millisecond, 1, 2 * runtime.NumCPU(), c, true)

// Execute 100 operations in parallel.
p.Execute(100, func(i int) {
  // Perform the ith operation.
})

// Examine the PID output signal programmatically
s := p.PIDProbe.Signal()

Stopping a Process

A process can be stopped at any time by calling the Stop() method. The process will stop after any operations that have already begun finish executing.

p.Execute(100, func(i int) {
  // Perform the ith operation with an error condition...
  if err != nil {
    p.Stop()
  }
})

Documentation

Overview

Package parallel provides objects for executing parallel operations on either a fixed or variable number of goroutines.

Index

• type ControllerConfiguration
  • func NewControllerConfiguration(kp float64, ki float64, kd float64, errorResponse float64, ...) *ControllerConfiguration
  • func (c *ControllerConfiguration) Copy() *ControllerConfiguration
• type FixedProcess
  • func NewFixedProcess(numRoutines int) *FixedProcess
  • func (p *FixedProcess) Execute(iterations int, operation Operation)
  • func (p *FixedProcess) NumRoutines() int
  • func (p *FixedProcess) Stop()
• type Operation
• type Process
• type VariableProcess
  • func NewVariableProcess(interval time.Duration, initialRoutines int, maxRoutines int, ...) *VariableProcess
  • func (p *VariableProcess) Execute(iterations int, operation Operation)
  • func (p *VariableProcess) GetControllerConfiguration() *ControllerConfiguration
  • func (p *VariableProcess) GetMaxRoutines() int
  • func (p *VariableProcess) GetOptimizationInterval() time.Duration
  • func (p *VariableProcess) NumRoutines() int
  • func (p *VariableProcess) SetControllerConfiguration(configuration *ControllerConfiguration)
  • func (p *VariableProcess) SetMaxRoutines(n int)
  • func (p *VariableProcess) SetOptimizationInterval(interval time.Duration)
  • func (p *VariableProcess) Stop()

Types

type ControllerConfiguration

type ControllerConfiguration struct {
	// The proportional term coefficient.
	Kp float64

	// The integral term coefficient.
	Ki float64

	// The derivative term coefficient.
	Kd float64

	// The error function response.
	ErrorResponse float64

	// The output signal response.
	OutputResponse float64
}

ControllerConfiguration types contain the information necessary to calculate values from a PID controller.

func NewControllerConfiguration

func NewControllerConfiguration(kp float64, ki float64, kd float64, errorResponse float64, outputResponse float64) *ControllerConfiguration

NewControllerConfiguration creates and returns a new controller configuration.

func (*ControllerConfiguration) Copy

func (c *ControllerConfiguration) Copy() *ControllerConfiguration

Copy returns a copy of the configuration.

type FixedProcess

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

FixedProcess types execute a specified number of operations on a given number of goroutines.

func NewFixedProcess

func NewFixedProcess(numRoutines int) *FixedProcess

NewFixedProcess creates and returns a new parallel process with the specified number of goroutines.

func (*FixedProcess) Execute

func (p *FixedProcess) Execute(iterations int, operation Operation)

Execute executes the fixed process for the specified number of operations.

func (*FixedProcess) NumRoutines

func (p *FixedProcess) NumRoutines() int

NumRoutines returns the number of routines that the synced processes was initialized with.

func (*FixedProcess) Stop

func (p *FixedProcess) Stop()

Stop stops the fixed process after all of the current operations have finished executing.

type Operation

type Operation func(i int)

Operation types represent a single operation in a parallel process. Responders should perform the i-th operation.

type Process

type Process interface {

	// Execute executes a parallel process for the given number of iterations
	// using the provided operation function.
	Execute(iterations int, operation Operation)

	// Stop stops the process if it is currently executing.
	Stop()

	// NumRoutines returns the number of routines that are currently executing in
	// the parallel process.
	NumRoutines() int
}

Process types execute a specified number of operations on a given number of goroutines.

type VariableProcess

type VariableProcess struct {

	// The CPU probe.
	CPUProbe *probes.Probe

	// The error probe.
	ErrorProbe *probes.Probe

	// The PID output probe.
	PIDProbe *probes.Probe

	// The routine probe.
	RoutineProbe *probes.Probe
	// contains filtered or unexported fields
}

VariableProcess types execute a specified number of operations on a variable number of goroutines.

func NewVariableProcess

func NewVariableProcess(interval time.Duration, initialRoutines int, maxRoutines int, controllerConfiguration *ControllerConfiguration, probeController bool) *VariableProcess

NewVariableProcess creates and returns a new parallel process with the specified optimization interval.

func (*VariableProcess) Execute

func (p *VariableProcess) Execute(iterations int, operation Operation)

Execute executes the parallel process for the specified number of operations while optimizing every interval iterations.

func (*VariableProcess) GetControllerConfiguration

func (p *VariableProcess) GetControllerConfiguration() *ControllerConfiguration

GetControllerConfiguration gets the PID controller configuration.

func (*VariableProcess) GetMaxRoutines

func (p *VariableProcess) GetMaxRoutines() int

GetMaxRoutines returns the maximum number of goroutines to use when optimizing.

func (*VariableProcess) GetOptimizationInterval

func (p *VariableProcess) GetOptimizationInterval() time.Duration

GetOptimizationInterval returns the interval of the process' ticker.

func (*VariableProcess) NumRoutines

func (p *VariableProcess) NumRoutines() int

NumRoutines returns the number of routines that the variable processes is currently using.

func (*VariableProcess) SetControllerConfiguration

func (p *VariableProcess) SetControllerConfiguration(configuration *ControllerConfiguration)

SetControllerConfiguration sets the PID controller coefficients.

func (*VariableProcess) SetMaxRoutines

func (p *VariableProcess) SetMaxRoutines(n int)

SetMaxRoutines sets the maximum number of goroutines to use when optimizing. Must be greater than 0.

func (*VariableProcess) SetOptimizationInterval

func (p *VariableProcess) SetOptimizationInterval(interval time.Duration)

SetOptimizationInterval sets the optimization interval and restarts the process' ticker.

func (*VariableProcess) Stop

func (p *VariableProcess) Stop()

Stop stops the variable process after all of the current operations have finished executing.