func (worker *Worker) isShuttingDown() bool {
	worker.Lock()
	defer worker.Unlock()
	return worker.shuttingDown
}
IsShutdown checks to see if the worker is in the process of being shutdown.
package worker
import (
	"encoding/binary"
	"fmt"
	"sync"
	"time"
	rt "github.com/appscode/g2/pkg/runtime"
)
const (
	Unlimited = iota
	OneByOne
)
type Worker struct {
	sync.Mutex
	agents  []*agent
	funcs   jobFuncs
	in      chan *inPack
	running bool
	ready   bool
	shuttingDown bool
	activeJobs sync.WaitGroup
	once sync.Once
	Id           string
	ErrorHandler ErrorHandler
	JobHandler   JobHandler
	limit        chan bool
}
func New(limit int) (worker *Worker) {
	worker = &Worker{
		agents: make([]*agent, 0, limit),
		funcs:  make(jobFuncs),
		in:     make(chan *inPack, rt.QueueSize),
	}
	if limit != Unlimited {
		worker.limit = make(chan bool, limit-1)
	}
	return
}
func (worker *Worker) err(e error) {
	if worker.ErrorHandler != nil {
		worker.ErrorHandler(e)
	}
}
func (worker *Worker) AddServer(net, addr string) (err error) {
	a, err := newAgent(net, addr, worker)
	if err != nil {
		return err
	}
	worker.agents = append(worker.agents, a)
	return
}
func (worker *Worker) broadcast(outpack *outPack) {
	for _, v := range worker.agents {
		v.write(outpack)
	}
}
func (worker *Worker) AddFunc(funcname string,
	f JobFunc, timeout uint32) (err error) {
	worker.Lock()
	defer worker.Unlock()
	if _, ok := worker.funcs[funcname]; ok {
		return fmt.Errorf("The function already exists: %s", funcname)
	}
	worker.funcs[funcname] = &jobFunc{f: f, timeout: timeout}
	if worker.running {
		worker.addFunc(funcname, timeout)
	}
	return
}
func (worker *Worker) addFunc(funcname string, timeout uint32) {
	outpack := prepFuncOutpack(funcname, timeout)
	worker.broadcast(outpack)
}
func prepFuncOutpack(funcname string, timeout uint32) *outPack {
	outpack := getOutPack()
	if timeout == 0 {
		outpack.dataType = rt.PT_CanDo
		outpack.data = []byte(funcname)
	} else {
		outpack.dataType = rt.PT_CanDoTimeout
		l := len(funcname)
		outpack.data = rt.NewBuffer(l + 5)
		copy(outpack.data, []byte(funcname))
		outpack.data[l] = '\x00'
		binary.BigEndian.PutUint32(outpack.data[l+1:], timeout)
	}
	return outpack
}
func (worker *Worker) RemoveFunc(funcname string) (err error) {
	worker.Lock()
	defer worker.Unlock()
	if _, ok := worker.funcs[funcname]; !ok {
		return fmt.Errorf("The function does not exist: %s", funcname)
	}
	delete(worker.funcs, funcname)
	if worker.running {
		worker.removeFunc(funcname)
	}
	return
}
func (worker *Worker) removeFunc(funcname string) {
	outpack := getOutPack()
	outpack.dataType = rt.PT_CantDo
	outpack.data = []byte(funcname)
	worker.broadcast(outpack)
}
func (worker *Worker) handleInPack(inpack *inPack) {
	switch inpack.dataType {
	case rt.PT_NoJob:
		inpack.a.PreSleep()
	case rt.PT_Noop:
		if !worker.isShuttingDown() {
			inpack.a.Grab()
		}
	case rt.PT_JobAssign, rt.PT_JobAssignUniq:
		go func() {
			if err := worker.exec(inpack); err != nil {
				worker.err(err)
			}
		}()
		if worker.limit != nil {
			worker.limit <- true
		}
		if !worker.isShuttingDown() {
			inpack.a.Grab()
		}
	case rt.PT_Error:
		worker.err(inpack.Err())
		fallthrough
	case rt.PT_EchoRes:
		fallthrough
	default:
		worker.customHandler(inpack)
	}
}
func (worker *Worker) Ready() (err error) {
	if len(worker.agents) == 0 {
		return ErrNoneAgents
	}
	if len(worker.funcs) == 0 {
		return ErrNoneFuncs
	}
	for _, a := range worker.agents {
		if err = a.Connect(); err != nil {
			return
		}
	}
	worker.once.Do(func() {
		for funcname, f := range worker.funcs {
			worker.addFunc(funcname, f.timeout)
		}
	})
	worker.ready = true
	return
}
func (worker *Worker) Work() {
	if !worker.ready {
		err := worker.Ready()
		if err != nil {
			panic(err)
		}
	}
	worker.running = true
	for _, a := range worker.agents {
		a.Grab()
	}
	var inpack *inPack
	for inpack = range worker.in {
		worker.handleInPack(inpack)
	}
}
func (worker *Worker) customHandler(inpack *inPack) {
	if worker.JobHandler != nil {
		if err := worker.JobHandler(inpack); err != nil {
			worker.err(err)
		}
	}
}
func (worker *Worker) Close() {
	worker.Lock()
	defer worker.Unlock()
	if worker.running == true {
		for _, a := range worker.agents {
			a.Close()
		}
		worker.running = false
		close(worker.in)
	}
}
func (worker *Worker) Reconnect() error {
	worker.Lock()
	defer worker.Unlock()
	if worker.running == true {
		for _, a := range worker.agents {
			if err := a.reconnect(); err != nil {
				return err
			}
		}
	}
	return nil
}
func (worker *Worker) Echo(data []byte) {
	outpack := getOutPack()
	outpack.dataType = rt.PT_EchoReq
	outpack.data = data
	worker.broadcast(outpack)
}
func (worker *Worker) Reset() {
	outpack := getOutPack()
	outpack.dataType = rt.PT_ResetAbilities
	worker.broadcast(outpack)
	worker.funcs = make(jobFuncs)
}
func (worker *Worker) SetId(id string) {
	worker.Id = id
	outpack := getOutPack()
	outpack.dataType = rt.PT_SetClientId
	outpack.data = []byte(id)
	worker.broadcast(outpack)
}
func (worker *Worker) exec(inpack *inPack) (err error) {
	defer func() {
		worker.Lock()
		worker.activeJobs.Done()
		worker.Unlock()
		if worker.limit != nil {
			<-worker.limit
		}
		if r := recover(); r != nil {
			if e, ok := r.(error); ok {
				err = e
			} else {
				err = ErrUnknown
			}
		}
	}()
	worker.activeJobs.Add(1)
	if worker.isShuttingDown() {
		return
	}
	f, ok := worker.funcs[inpack.fn]
	if !ok {
		return fmt.Errorf("The function does not exist: %s", inpack.fn)
	}
	var r *result
	if f.timeout == 0 {
		d, e := f.f(inpack)
		r = &result{data: d, err: e}
	} else {
		r = execTimeout(f.f, inpack, time.Duration(f.timeout)*time.Second)
	}
	if worker.running {
		outpack := getOutPack()
		if r.err == nil {
			outpack.dataType = rt.PT_WorkComplete
		} else {
			if len(r.data) == 0 {
				outpack.dataType = rt.PT_WorkFail
			} else {
				outpack.dataType = rt.PT_WorkException
			}
			err = r.err
		}
		outpack.handle = inpack.handle
		outpack.data = r.data
		inpack.a.Write(outpack)
	}
	return
}
func (worker *Worker) reRegisterFuncsForAgent(a *agent) {
	worker.Lock()
	defer worker.Unlock()
	for funcname, f := range worker.funcs {
		outpack := prepFuncOutpack(funcname, f.timeout)
		a.write(outpack)
	}
}
func (worker *Worker) Shutdown() {
	worker.Lock()
	worker.shuttingDown = true
	worker.Unlock()
	worker.activeJobs.Wait()
	worker.Close()
}