started refactoring worker

client/client.go

@@ -46,7 +46,6 @@ func New(net, addr string) (client *Client, err error) {
 	if err = client.connect(); err != nil {
 		return
 	}
-	client.isConn = true
 	go client.readLoop()
 	go client.processLoop()
 	return
@@ -57,6 +56,10 @@ func New(net, addr string) (client *Client, err error) {
 //
 func (client *Client) connect() (err error) {
 	client.conn, err = net.Dial(client.net, client.addr)
+	if err != nil {
+		return
+	}
+	client.isConn = true
 	return
 }
 

client/error.go

@@ -28,8 +28,6 @@ var (
 	ErrConnClosed    = errors.New("Connection closed")
 )
 
-func DisablePanic() { recover() }
-
 // Extract the error message
 func GetError(data []byte) (err error) {
 	rel := bytes.SplitN(data, []byte{'\x00'}, 2)

client/response.go

@@ -106,14 +106,6 @@ func decodeResponse(data []byte) (resp *Response, l int, err error) {
 	return
 }
 
-func (resp *Response) IsEcho() bool {
-	return resp.DataType == ECHO_RES
-}
-
-func (resp *Response) IsStatus() bool {
-	return resp.DataType == STATUS_RES
-}
-
 // status handler
 func (resp *Response) Status() (status *Status, err error) {
 	data := bytes.SplitN(resp.Data, []byte{'\x00'}, 4)

gearman.go

@@ -11,6 +11,5 @@ package gearman
 
 import (
 	_ "github.com/mikespook/gearman-go/client"
-	_ "github.com/mikespook/gearman-go/common"
 	_ "github.com/mikespook/gearman-go/worker"
 )

gearman_test.go

@@ -23,19 +23,20 @@ const (
 	GEARMAND = "127.0.0.1:4730"
 )
 
-func ToUpper(job *worker.Job) ([]byte, error) {
+func ToUpper(job worker.Job) ([]byte, error) {
-	data := []byte(strings.ToUpper(string(job.Data)))
+	data := job.Data()
+	data = []byte(strings.ToUpper(string(data)))
 	return data, nil
 }
 
-func Sleep(job *worker.Job) ([]byte, error) {
+func Sleep(job worker.Job) ([]byte, error) {
 	time.Sleep(time.Second * 5)
 	return nil, nil
 }
 
 func TestJobs(t *testing.T) {
 	w := worker.New(worker.Unlimited)
-	if err := w.AddServer(GEARMAND); err != nil {
+	if err := w.AddServer("tcp4", GEARMAND); err != nil {
 		t.Error(err)
 		return
 	}
@@ -50,7 +51,7 @@ func TestJobs(t *testing.T) {
 		return
 	}
 	t.Log("Functions added...")
-	w.ErrHandler = func(e error) {
+	w.ErrorHandler = func(e error) {
 		t.Error(e)
 	}
 	go w.Work()

worker/agent.go

@@ -14,184 +14,110 @@ type agent struct {
 	conn   net.Conn
 	worker *Worker
 	in     chan []byte
-	out    chan *Job
+	net, addr   string
-	addr   string
+	isConn bool
 }
 
 // Create the agent of job server.
-func newAgent(addr string, worker *Worker) (a *agent, err error) {
+func newAgent(net, addr string, worker *Worker) (a *agent, err error) {
-	conn, err := net.Dial(common.NETWORK, addr)
-	if err != nil {
-		return
-	}
 	a = &agent{
-		conn:   conn,
+		net: net,
-		worker: worker,
 		addr:   addr,
-		in:     make(chan []byte, common.QUEUE_SIZE),
+		worker: worker,
-		out:    make(chan *Job, common.QUEUE_SIZE),
+		in:     make(chan []byte, QUEUE_SIZE),
 	}
-	// reset abilities
-	a.WriteJob(newJob(common.REQ, common.RESET_ABILITIES, nil))
 	return
 }
 
-// outputing loop
+func (a *agent) Connect() (err error) {
-func (a *agent) outLoop() {
+	a.conn, err = net.Dial(a.net, a.addr)
-	ok := true
+	if err != nil {
-	var job *Job
+		return
-	for a.worker.running && ok {
-		if job, ok = <-a.out; ok {
-			if err := a.write(job.Encode()); err != nil {
-				a.worker.err(err)
 	}
+	a.isConn = true
+	return
+}
+
+func (a *agent) Work() {
+	go a.readLoop()
+
+	var resp *Response
+	var l int
+	var err error
+	var data, leftdata []byte
+	for data = range a.in {
+		if len(leftdata) > 0 { // some data left for processing
+			data = append(leftdata, data...)
+		}
+		l = len(data)
+		if l < MIN_PACKET_LEN { // not enough data
+			leftdata = data
+			continue
+		}
+		if resp, l, err = decodeResponse(data); err != nil {
+			a.worker.err(err)
+			continue
+		}
+		leftdata = nil
+		resp.agentId = a.net + a.addr
+		a.worker.in <- resp
+		if len(data) > l {
+			leftdata = data[l:]
 		}
 	}
 }
 
-// inputing loop
+// read data from socket
-func (a *agent) inLoop() {
+func (a *agent) readLoop() {
-	defer func() {
+	var data []byte
-		if r := recover(); r != nil {
+	var err error
-			a.worker.err(common.Errorf("Exiting: %s", r))
+	for a.isConn {
-		}
+		if data, err = a.read(BUFFER_SIZE); err != nil {
-		close(a.in)
+			if err == ErrConnClosed {
-		close(a.out)
-		a.worker.removeAgent(a)
-	}()
-	for a.worker.running {
-		a.WriteJob(newJob(common.REQ, common.PRE_SLEEP, nil))
-	RESTART:
-		// got noop msg and in queue is zero, grab job
-		rel, err := a.read()
-		if err != nil {
-			if err == common.ErrConnection {
-				for i := 0; i < 3 && a.worker.running; i++ {
-					if conn, err := net.Dial(common.NETWORK, a.addr); err != nil {
-						a.worker.err(common.Errorf("Reconnection: %d faild", i))
-						continue
-					} else {
-						a.conn = conn
-						goto RESTART
-					}
-				}
-				a.worker.err(err)
 				break
 			}
 			a.worker.err(err)
 			continue
 		}
-		job, err := decodeJob(rel)
+		a.in <- data
-		if err != nil {
-			a.worker.err(err)
-			continue
-		}
-		switch job.DataType {
-		case common.NOOP:
-			a.WriteJob(newJob(common.REQ, common.GRAB_JOB_UNIQ, nil))
-		case common.ERROR, common.ECHO_RES, common.JOB_ASSIGN_UNIQ, common.JOB_ASSIGN:
-			if a.worker.running {
-				if a.worker.limit != nil {
-					a.worker.limit <- true
-				}
-				job.agent = a
-				a.worker.in <- job
-			}
-		}
 	}
+	close(a.in)
 }
 
 func (a *agent) Close() {
 	a.conn.Close()
 }
 
-func (a *agent) Work() {
+// read length bytes from the socket
-	go a.outLoop()
+func (a *agent) read(length int) (data []byte, err error) {
-	go a.inLoop()
-}
-
-func (a *agent) readData(length int) (data []byte, err error) {
 	n := 0
-	buf := make([]byte, common.BUFFER_SIZE)
+	buf := getBuffer(BUFFER_SIZE)
 	// read until data can be unpacked
-	for i := length; i > 0 || len(data) < common.PACKET_LEN; i -= n {
+	for i := length; i > 0 || len(data) < MIN_PACKET_LEN; i -= n {
 		if n, err = a.conn.Read(buf); err != nil {
-			if err == io.EOF && n == 0 {
+			if !a.isConn {
-				if data == nil {
+				err = ErrConnClosed
-					err = common.ErrConnection
 				return
 			}
-				return data, nil
+			if err == io.EOF && n == 0 {
+				if data == nil {
+					err = ErrConnection
+				}
 			}
 			return
 		}
 		data = append(data, buf[0:n]...)
-		if n < common.BUFFER_SIZE {
+		if n < BUFFER_SIZE {
 			break
 		}
 	}
 	return
 }
 
-func (a *agent) unpack(data []byte) ([]byte, int, bool) {
-	tl := len(data)
-	start := 0
-	for i := 0; i < tl+1-common.PACKET_LEN; i++ {
-		if start+common.PACKET_LEN > tl { // too few data to unpack, read more
-			return nil, common.PACKET_LEN, false
-		}
-		if string(data[start:start+4]) == common.RES_STR {
-			l := int(common.BytesToUint32([4]byte{data[start+8],
-				data[start+9], data[start+10], data[start+11]}))
-			total := l + common.PACKET_LEN
-			if total == tl { // data is what we want
-				return data, common.PACKET_LEN, true
-			} else if total < tl { // data[:total] is what we want, data[total:] is the more
-				a.in <- data[total:]
-				data = data[start:total]
-				return data, common.PACKET_LEN, true
-			} else { // ops! It won't be possible.
-				return nil, total - tl, false
-			}
-		} else { // flag was not found, move to next step
-			start++
-		}
-	}
-	return nil, common.PACKET_LEN, false
-}
-
-func (a *agent) read() (rel []byte, err error) {
-	var data []byte
-	ok := false
-	l := common.PACKET_LEN
-	for !ok {
-		inlen := len(a.in)
-		if inlen > 0 {
-			// in queue is not empty
-			for i := 0; i < inlen; i++ {
-				data = append(data, <-a.in...)
-			}
-		} else {
-			var d []byte
-			d, err = a.readData(l)
-			if err != nil {
-				return
-			}
-			data = append(data, d...)
-		}
-		rel, l, ok = a.unpack(data)
-	}
-	return
-}
-
-// Send a job to the job server.
-func (a *agent) WriteJob(job *Job) {
-	a.out <- job
-}
-
 // Internal write the encoded job.
-func (a *agent) write(buf []byte) (err error) {
+func (a *agent) write(req *request) (err error) {
 	var n int
+	buf := req.Encode()
 	for i := 0; i < len(buf); i += n {
 		n, err = a.conn.Write(buf[i:])
 		if err != nil {

worker/common.go

@@ -54,3 +54,9 @@ const (
 	SUBMIT_JOB_LOW     = 33
 	SUBMIT_JOB_LOW_BG  = 34
 )
+
+func getBuffer(l int) (buf []byte) {
+	// TODO add byte buffer pool
+	buf = make([]byte, l)
+	return
+}

worker/error.go

@@ -28,8 +28,6 @@ var (
 	ErrConnClosed    = errors.New("Connection closed")
 )
 
-func DisablePanic() { recover() }
-
 // Extract the error message
 func GetError(data []byte) (err error) {
 	rel := bytes.SplitN(data, []byte{'\x00'}, 2)

worker/func.go

@@ -5,13 +5,27 @@ import (
 	"runtime"
 )
 
+// Job handler
+type JobHandler func(Job) error
+
+type JobFunc func(Job) ([]byte, error)
+
+// The definition of the callback function.
+type jobFunc struct {
+	f       JobFunc
+	timeout uint32
+}
+
+// Map for added function.
+type JobFuncs map[string]*jobFunc
+
 type systemInfo struct {
 	GOOS, GOARCH, GOROOT, Version string
 	NumCPU, NumGoroutine          int
 	NumCgoCall                    int64
 }
 
-func SysInfo(job *Job) ([]byte, error) {
+func SysInfo(job Job) ([]byte, error) {
 	return json.Marshal(&systemInfo{
 		GOOS:         runtime.GOOS,
 		GOARCH:       runtime.GOARCH,
@@ -25,7 +39,7 @@ func SysInfo(job *Job) ([]byte, error) {
 
 var memState runtime.MemStats
 
-func MemInfo(job *Job) ([]byte, error) {
+func MemInfo(job Job) ([]byte, error) {
 	runtime.ReadMemStats(&memState)
 	return json.Marshal(&memState)
 }

worker/job.go

@@ -0,0 +1,68 @@
+package worker
+
+import (
+	"strconv"
+)
+
+type Job interface {
+	Data() []byte
+	SendWarning(data []byte)
+	SendData(data []byte)
+	UpdateStatus(numerator, denominator int)
+}
+
+type _job struct {
+	a *agent
+	Handle string
+	data []byte
+}
+
+func getJob() *_job {
+	return &_job{}
+}
+
+func (j *_job) Data() []byte {
+	return j.data
+}
+
+// Send some datas to client.
+// Using this in a job's executing.
+func (j *_job) SendData(data []byte) {
+	req := getRequest()
+	req.DataType = WORK_DATA
+	hl := len(j.Handle)
+	l := hl + len(data) + 1
+	req.Data = getBuffer(l)
+	copy(req.Data, []byte(j.Handle))
+	copy(req.Data[hl + 1:], data)
+	j.a.write(req)
+}
+
+func (j *_job) SendWarning(data []byte) {
+	req := getRequest()
+	req.DataType = WORK_WARNING
+	hl := len(j.Handle)
+	l := hl + len(data) + 1
+	req.Data = getBuffer(l)
+	copy(req.Data, []byte(j.Handle))
+	copy(req.Data[hl + 1:], data)
+	j.a.write(req)
+}
+
+// Update status.
+// Tall client how many percent job has been executed.
+func (j *_job) UpdateStatus(numerator, denominator int) {
+	n := []byte(strconv.Itoa(numerator))
+	d := []byte(strconv.Itoa(denominator))
+	req := getRequest()
+	req.DataType = WORK_STATUS
+	hl := len(j.Handle)
+	nl := len(n)
+	dl := len(d)
+	req.Data = getBuffer(hl + nl + dl + 3)
+	copy(req.Data, []byte(j.Handle))
+	copy(req.Data[hl+1:], n)
+	copy(req.Data[hl+nl+2:], d)
+	j.a.write(req)
+}
+

worker/request.go

@@ -6,128 +6,45 @@
 package worker
 
 import (
-	"bytes"
+	"encoding/binary"
-	"strconv"
 )
 
 // Worker side job
-type Job struct {
+type request struct {
+	DataType             uint32
 	Data                 []byte
 	Handle, UniqueId, Fn string
-	agent                *agent
-	magicCode, DataType  uint32
-	c                    chan bool
 }
 
-// Create a new job
+func getRequest() (req *request) {
-func newJob(magiccode, datatype uint32, data []byte) (job *Job) {
+	// TODO pool
-	return &Job{magicCode: magiccode,
+	return &request{}
-		DataType: datatype,
-		Data:     data,
-		c:        make(chan bool)}
-}
-
-// Decode job from byte slice
-func decodeJob(data []byte) (job *Job, err error) {
-	if len(data) < 12 {
-		return nil, common.Errorf("Invalid data: %V", data)
-	}
-	datatype := common.BytesToUint32([4]byte{data[4], data[5], data[6], data[7]})
-	l := common.BytesToUint32([4]byte{data[8], data[9], data[10], data[11]})
-	if len(data[12:]) != int(l) {
-		return nil, common.Errorf("Invalid data: %V", data)
-	}
-	data = data[12:]
-	job = &Job{magicCode: common.RES, DataType: datatype, c: make(chan bool)}
-	switch datatype {
-	case common.JOB_ASSIGN:
-		s := bytes.SplitN(data, []byte{'\x00'}, 3)
-		if len(s) == 3 {
-			job.Handle = string(s[0])
-			job.Fn = string(s[1])
-			data = s[2]
-		}
-	case common.JOB_ASSIGN_UNIQ:
-		s := bytes.SplitN(data, []byte{'\x00'}, 4)
-		if len(s) == 4 {
-			job.Handle = string(s[0])
-			job.Fn = string(s[1])
-			job.UniqueId = string(s[2])
-			data = s[3]
-		}
-	}
-	job.Data = data
-	return
 }
 
 // Encode a job to byte slice
-func (job *Job) Encode() (data []byte) {
+func (req *request) Encode() (data []byte) {
 	var l int
-	if job.DataType == common.WORK_FAIL {
+	if req.DataType == WORK_FAIL {
-		l = len(job.Handle)
+		l = len(req.Handle)
 	} else {
-		l = len(job.Data)
+		l = len(req.Data)
-		if job.Handle != "" {
+		if req.Handle != "" {
-			l += len(job.Handle) + 1
+			l += len(req.Handle) + 1
 		}
 	}
-	data = make([]byte, 0, l+12)
+	data = getBuffer(l + MIN_PACKET_LEN)
-
+	binary.BigEndian.PutUint32(data[:4], REQ)
-	magiccode := common.Uint32ToBytes(job.magicCode)
+	binary.BigEndian.PutUint32(data[4:8], req.DataType)
-	datatype := common.Uint32ToBytes(job.DataType)
+	binary.BigEndian.PutUint32(data[8:MIN_PACKET_LEN], uint32(l))
-	datalength := common.Uint32ToBytes(uint32(l))
+	i := MIN_PACKET_LEN
-
+	if req.Handle != "" {
-	data = append(data, magiccode[:]...)
+		hi := len(req.Handle) + i
-	data = append(data, datatype[:]...)
+		copy(data[i:hi], []byte(req.Handle))
-	data = append(data, datalength[:]...)
+		if req.DataType != WORK_FAIL {
-	if job.Handle != "" {
+			data[hi] = '\x00'
-		data = append(data, []byte(job.Handle)...)
-		if job.DataType != common.WORK_FAIL {
-			data = append(data, 0)
 		}
+		i = i + hi
 	}
-	data = append(data, job.Data...)
+	copy(data[i:], req.Data)
 	return
 }
-
-// Send some datas to client.
-// Using this in a job's executing.
-func (job *Job) UpdateData(data []byte, iswarning bool) {
-	result := append([]byte(job.Handle), 0)
-	result = append(result, data...)
-	var datatype uint32
-	if iswarning {
-		datatype = common.WORK_WARNING
-	} else {
-		datatype = common.WORK_DATA
-	}
-	job.agent.WriteJob(newJob(common.REQ, datatype, result))
-}
-
-// Update status.
-// Tall client how many percent job has been executed.
-func (job *Job) UpdateStatus(numerator, denominator int) {
-	n := []byte(strconv.Itoa(numerator))
-	d := []byte(strconv.Itoa(denominator))
-	result := append([]byte(job.Handle), '\x00')
-	result = append(result, n...)
-	result = append(result, '\x00')
-	result = append(result, d...)
-	job.agent.WriteJob(newJob(common.REQ, common.WORK_STATUS, result))
-}
-
-// close the job
-func (job *Job) Close() {
-	close(job.c)
-}
-
-// cancel the job executing
-func (job *Job) cancel() {
-	defer func() { recover() }()
-	job.c <- true
-}
-
-// When a job was canceled, return a true form a channel
-func (job *Job) Canceled() <-chan bool {
-	return job.c
-}

worker/response.go

@@ -7,127 +7,56 @@ package worker
 
 import (
 	"bytes"
-	"strconv"
+	"fmt"
+	"encoding/binary"
 )
 
 // Worker side job
-type Job struct {
+type Response struct {
+	DataType  uint32
 	Data                 []byte
 	Handle, UniqueId, Fn string
-	agent                *agent
+	agentId	string
-	magicCode, DataType  uint32
-	c                    chan bool
 }
 
 // Create a new job
-func newJob(magiccode, datatype uint32, data []byte) (job *Job) {
+func getResponse() (resp *Response) {
-	return &Job{magicCode: magiccode,
+	return &Response{}
-		DataType: datatype,
-		Data:     data,
-		c:        make(chan bool)}
 }
 
 // Decode job from byte slice
-func decodeJob(data []byte) (job *Job, err error) {
+func decodeResponse(data []byte) (resp *Response, l int, err error) {
-	if len(data) < 12 {
+	if len(data) < MIN_PACKET_LEN { // valid package should not less 12 bytes
-		return nil, common.Errorf("Invalid data: %V", data)
+		err = fmt.Errorf("Invalid data: %V", data)
+		return
 	}
-	datatype := common.BytesToUint32([4]byte{data[4], data[5], data[6], data[7]})
+	dl := int(binary.BigEndian.Uint32(data[8:12]))
-	l := common.BytesToUint32([4]byte{data[8], data[9], data[10], data[11]})
+	dt := data[MIN_PACKET_LEN : dl+MIN_PACKET_LEN]
-	if len(data[12:]) != int(l) {
+	if len(dt) != int(dl) { // length not equal
-		return nil, common.Errorf("Invalid data: %V", data)
+		err = fmt.Errorf("Invalid data: %V", data)
+		return
 	}
-	data = data[12:]
+	resp = getResponse()
-	job = &Job{magicCode: common.RES, DataType: datatype, c: make(chan bool)}
+	resp.DataType = binary.BigEndian.Uint32(data[4:8])
-	switch datatype {
+	switch resp.DataType {
-	case common.JOB_ASSIGN:
+	case JOB_ASSIGN:
-		s := bytes.SplitN(data, []byte{'\x00'}, 3)
+		s := bytes.SplitN(dt, []byte{'\x00'}, 3)
 		if len(s) == 3 {
-			job.Handle = string(s[0])
+			resp.Handle = string(s[0])
-			job.Fn = string(s[1])
+			resp.Fn = string(s[1])
-			data = s[2]
+			resp.Data = s[2]
 		}
-	case common.JOB_ASSIGN_UNIQ:
+	case JOB_ASSIGN_UNIQ:
-		s := bytes.SplitN(data, []byte{'\x00'}, 4)
+		s := bytes.SplitN(dt, []byte{'\x00'}, 4)
 		if len(s) == 4 {
-			job.Handle = string(s[0])
+			resp.Handle = string(s[0])
-			job.Fn = string(s[1])
+			resp.Fn = string(s[1])
-			job.UniqueId = string(s[2])
+			resp.UniqueId = string(s[2])
-			data = s[3]
+			resp.Data = s[3]
 		}
+	default:
+		resp.Data = dt
 	}
-	job.Data = data
+	l = dl + MIN_PACKET_LEN
 	return
 }
-
-// Encode a job to byte slice
-func (job *Job) Encode() (data []byte) {
-	var l int
-	if job.DataType == common.WORK_FAIL {
-		l = len(job.Handle)
-	} else {
-		l = len(job.Data)
-		if job.Handle != "" {
-			l += len(job.Handle) + 1
-		}
-	}
-	data = make([]byte, 0, l+12)
-
-	magiccode := common.Uint32ToBytes(job.magicCode)
-	datatype := common.Uint32ToBytes(job.DataType)
-	datalength := common.Uint32ToBytes(uint32(l))
-
-	data = append(data, magiccode[:]...)
-	data = append(data, datatype[:]...)
-	data = append(data, datalength[:]...)
-	if job.Handle != "" {
-		data = append(data, []byte(job.Handle)...)
-		if job.DataType != common.WORK_FAIL {
-			data = append(data, 0)
-		}
-	}
-	data = append(data, job.Data...)
-	return
-}
-
-// Send some datas to client.
-// Using this in a job's executing.
-func (job *Job) UpdateData(data []byte, iswarning bool) {
-	result := append([]byte(job.Handle), 0)
-	result = append(result, data...)
-	var datatype uint32
-	if iswarning {
-		datatype = common.WORK_WARNING
-	} else {
-		datatype = common.WORK_DATA
-	}
-	job.agent.WriteJob(newJob(common.REQ, datatype, result))
-}
-
-// Update status.
-// Tall client how many percent job has been executed.
-func (job *Job) UpdateStatus(numerator, denominator int) {
-	n := []byte(strconv.Itoa(numerator))
-	d := []byte(strconv.Itoa(denominator))
-	result := append([]byte(job.Handle), '\x00')
-	result = append(result, n...)
-	result = append(result, '\x00')
-	result = append(result, d...)
-	job.agent.WriteJob(newJob(common.REQ, common.WORK_STATUS, result))
-}
-
-// close the job
-func (job *Job) Close() {
-	close(job.c)
-}
-
-// cancel the job executing
-func (job *Job) cancel() {
-	defer func() { recover() }()
-	job.c <- true
-}
-
-// When a job was canceled, return a true form a channel
-func (job *Job) Canceled() <-chan bool {
-	return job.c
-}

worker/worker.go

@@ -5,7 +5,10 @@
 package worker
 
 import (
+	"fmt"
 	"time"
+	"sync"
+	"encoding/binary"
 )
 
 const (
@@ -15,24 +18,6 @@ const (
 	Immediately = 0
 )
 
-var (
-	ErrConnection = common.ErrConnection
-)
-
-// Job handler
-type JobHandler func(*Job) error
-
-type JobFunc func(*Job) ([]byte, error)
-
-// The definition of the callback function.
-type jobFunc struct {
-	f       JobFunc
-	timeout uint32
-}
-
-// Map for added function.
-type JobFuncs map[string]*jobFunc
-
 /*
 Worker side api for gearman
 
@@ -52,24 +37,25 @@ func foobar(job *Job) (data []byte, err os.Error) {
 }
 */
 type Worker struct {
-	agents  []*agent
+	agents  map[string]*agent
 	funcs   JobFuncs
-	in      chan *Job
+	in      chan *Response
 	running bool
 	limit   chan bool
 
 	Id string
 	// assign a ErrFunc to handle errors
-	ErrHandler common.ErrorHandler
+	ErrorHandler ErrorHandler
 	JobHandler JobHandler
+	mutex sync.Mutex
 }
 
 // Get a new worker
 func New(l int) (worker *Worker) {
 	worker = &Worker{
-		agents: make([]*agent, 0),
+		agents: make(map[string]*agent, QUEUE_SIZE),
 		funcs:  make(JobFuncs),
-		in:     make(chan *Job, common.QUEUE_SIZE),
+		in:     make(chan *Response, QUEUE_SIZE),
 	}
 	if l != Unlimited {
 		worker.limit = make(chan bool, l)
@@ -79,29 +65,29 @@ func New(l int) (worker *Worker) {
 
 //
 func (worker *Worker) err(e error) {
-	if worker.ErrHandler != nil {
+	if worker.ErrorHandler != nil {
-		worker.ErrHandler(e)
+		worker.ErrorHandler(e)
 	}
 }
 
 // Add a server. The addr should be 'host:port' format.
 // The connection is established at this time.
-func (worker *Worker) AddServer(addr string) (err error) {
+func (worker *Worker) AddServer(net, addr string) (err error) {
 	// Create a new job server's client as a agent of server
-	server, err := newAgent(addr, worker)
+	a, err := newAgent(net, addr, worker)
 	if err != nil {
 		return err
 	}
-	worker.agents = append(worker.agents, server)
+	worker.agents[net + addr] = a
 	return
 }
 
 // Write a job to job server.
 // Here, the job's mean is not the oraginal mean.
 // Just looks like a network package for job's result or tell job server, there was a fail.
-func (worker *Worker) broadcast(job *Job) {
+func (worker *Worker) broadcast(req *request) {
 	for _, v := range worker.agents {
-		v.WriteJob(job)
+		v.write(req)
 	}
 }
 
@@ -110,11 +96,12 @@ func (worker *Worker) broadcast(job *Job) {
 // The API will tell every connected job server that 'I can do this'
 func (worker *Worker) AddFunc(funcname string,
 	f JobFunc, timeout uint32) (err error) {
+	worker.mutex.Lock()
+	defer worker.mutex.Unlock()
 	if _, ok := worker.funcs[funcname]; ok {
-		return common.Errorf("The function already exists: %s", funcname)
+		return fmt.Errorf("The function already exists: %s", funcname)
 	}
 	worker.funcs[funcname] = &jobFunc{f: f, timeout: timeout}
-
 	if worker.running {
 		worker.addFunc(funcname, timeout)
 	}
@@ -123,27 +110,27 @@ func (worker *Worker) AddFunc(funcname string,
 
 // inner add function
 func (worker *Worker) addFunc(funcname string, timeout uint32) {
-	var datatype uint32
+	req := getRequest()
-	var data []byte
 	if timeout == 0 {
-		datatype = common.CAN_DO
+		req.DataType = CAN_DO
-		data = []byte(funcname)
+		req.Data = []byte(funcname)
 	} else {
-		datatype = common.CAN_DO_TIMEOUT
+		req.DataType = CAN_DO_TIMEOUT
-		data = []byte(funcname + "\x00")
+		l := len(funcname)
-		t := common.Uint32ToBytes(timeout)
+		req.Data = getBuffer(l + 5)
-		data = append(data, t[:]...)
+		copy(req.Data, []byte(funcname))
+		req.Data[l] = '\x00'
+		binary.BigEndian.PutUint32(req.Data[l + 1:], timeout)
 	}
-	job := newJob(common.REQ, datatype, data)
+	worker.broadcast(req)
-	worker.broadcast(job)
-
 }
 
 // Remove a function.
-// Tell job servers 'I can not do this now' at the same time.
 func (worker *Worker) RemoveFunc(funcname string) (err error) {
+	worker.mutex.Lock()
+	defer worker.mutex.Unlock()
 	if _, ok := worker.funcs[funcname]; !ok {
-		return common.Errorf("The function does not exist: %s", funcname)
+		return fmt.Errorf("The function does not exist: %s", funcname)
 	}
 	delete(worker.funcs, funcname)
 	if worker.running {
@@ -154,27 +141,27 @@ func (worker *Worker) RemoveFunc(funcname string) (err error) {
 
 // inner remove function
 func (worker *Worker) removeFunc(funcname string) {
-	job := newJob(common.REQ, common.CANT_DO, []byte(funcname))
+	req := getRequest()
-	worker.broadcast(job)
+	req.DataType = CANT_DO
+	req.Data = []byte(funcname)
+	worker.broadcast(req)
 }
 
-func (worker *Worker) dealJob(job *Job) {
+func (worker *Worker) dealResp(resp *Response) {
 	defer func() {
-		job.Close()
 		if worker.running && worker.limit != nil {
 			<-worker.limit
 		}
 	}()
-	switch job.DataType {
+	switch resp.DataType {
-	case common.ERROR:
+	case ERROR:
-		_, err := common.GetError(job.Data)
+		worker.err(GetError(resp.Data))
-		worker.err(err)
+	case JOB_ASSIGN, JOB_ASSIGN_UNIQ:
-	case common.JOB_ASSIGN, common.JOB_ASSIGN_UNIQ:
+		if err := worker.exec(resp); err != nil {
-		if err := worker.exec(job); err != nil {
 			worker.err(err)
 		}
 	default:
-		worker.handleJob(job)
+		worker.handleResponse(resp)
 	}
 }
 
@@ -187,23 +174,29 @@ func (worker *Worker) Work() {
 	}()
 	worker.running = true
 	for _, v := range worker.agents {
+		v.Connect()
 		go v.Work()
 	}
+	worker.Reset()
 	for funcname, f := range worker.funcs {
 		worker.addFunc(funcname, f.timeout)
 	}
-	ok := true
+	var resp *Response
-	for ok {
+	for resp = range worker.in {
-		var job *Job
+		fmt.Println(resp)
-		if job, ok = <-worker.in; ok {
+		go worker.dealResp(resp)
-			go worker.dealJob(job)
-		}
 	}
 }
 
 // job handler
-func (worker *Worker) handleJob(job *Job) {
+func (worker *Worker) handleResponse(resp *Response) {
 	if worker.JobHandler != nil {
+		job := getJob()
+		job.a = worker.agents[resp.agentId]
+		job.Handle = resp.Handle
+		if resp.DataType == ECHO_RES {
+			job.data = resp.Data
+		}
 		if err := worker.JobHandler(job); err != nil {
 			worker.err(err)
 		}
@@ -221,85 +214,79 @@ func (worker *Worker) Close() {
 
 // Send a something out, get the samething back.
 func (worker *Worker) Echo(data []byte) {
-	job := newJob(common.REQ, common.ECHO_REQ, data)
+	req := getRequest()
-	worker.broadcast(job)
+	req.DataType = ECHO_REQ
+	req.Data = data
+	worker.broadcast(req)
 }
 
 // Remove all of functions.
 // Both from the worker or job servers.
 func (worker *Worker) Reset() {
-	job := newJob(common.REQ, common.RESET_ABILITIES, nil)
+	req := getRequest()
-	worker.broadcast(job)
+	req.DataType = RESET_ABILITIES
+	worker.broadcast(req)
 	worker.funcs = make(JobFuncs)
 }
 
 // Set the worker's unique id.
 func (worker *Worker) SetId(id string) {
 	worker.Id = id
-	job := newJob(common.REQ, common.SET_CLIENT_ID, []byte(id))
+	req := getRequest()
-	worker.broadcast(job)
+	req.DataType = SET_CLIENT_ID
+	req.Data = []byte(id)
+	worker.broadcast(req)
 }
 
 // Execute the job. And send back the result.
-func (worker *Worker) exec(job *Job) (err error) {
+func (worker *Worker) exec(resp *Response) (err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			if e, ok := r.(error); ok {
 				err = e
 			} else {
-				err = common.ErrUnknown
+				err = ErrUnknown
 			}
 		}
 	}()
-	f, ok := worker.funcs[job.Fn]
+	f, ok := worker.funcs[resp.Fn]
 	if !ok {
-		return common.Errorf("The function does not exist: %s", job.Fn)
+		return fmt.Errorf("The function does not exist: %s", resp.Fn)
 	}
 	var r *result
+	job := getJob()
+	job.a = worker.agents[resp.agentId]
+	job.Handle = resp.Handle
 	if f.timeout == 0 {
 		d, e := f.f(job)
 		r = &result{data: d, err: e}
 	} else {
 		r = execTimeout(f.f, job, time.Duration(f.timeout)*time.Second)
 	}
-	var datatype uint32
+	req := getRequest()
 	if r.err == nil {
-		datatype = common.WORK_COMPLETE
+		req.DataType = WORK_COMPLETE
 	} else {
 		if r.data == nil {
-			datatype = common.WORK_FAIL
+			req.DataType = WORK_FAIL
 		} else {
-			datatype = common.WORK_EXCEPTION
+			req.DataType = WORK_EXCEPTION
 		}
 		err = r.err
 	}
-
+	req.Data = r.data
-	job.magicCode = common.REQ
-	job.DataType = datatype
-	job.Data = r.data
 	if worker.running {
-		job.agent.WriteJob(job)
+		job.a.write(req)
 	}
 	return
 }
 
-func (worker *Worker) removeAgent(a *agent) {
-	for k, v := range worker.agents {
-		if v == a {
-			worker.agents = append(worker.agents[:k], worker.agents[k+1:]...)
-		}
-	}
-	if len(worker.agents) == 0 {
-		worker.err(common.ErrNoActiveAgent)
-	}
-}
-
 type result struct {
 	data []byte
 	err  error
 }
 
-func execTimeout(f JobFunc, job *Job, timeout time.Duration) (r *result) {
+func execTimeout(f JobFunc, job Job, timeout time.Duration) (r *result) {
 	rslt := make(chan *result)
 	defer close(rslt)
 	go func() {
@@ -310,8 +297,7 @@ func execTimeout(f JobFunc, job *Job, timeout time.Duration) (r *result) {
 	select {
 	case r = <-rslt:
 	case <-time.After(timeout):
-		go job.cancel()
+		return &result{err: ErrTimeOut}
-		return &result{err: common.ErrTimeOut}
 	}
 	return r
 }

worker/worker_test.go

@@ -10,7 +10,7 @@ func init() {
 
 func TestWorkerAddServer(t *testing.T) {
 	t.Log("Add local server 127.0.0.1:4730.")
-	if err := worker.AddServer("127.0.0.1:4730"); err != nil {
+	if err := worker.AddServer("tcp4", "127.0.0.1:4730"); err != nil {
 		t.Error(err)
 	}
 
@@ -20,7 +20,7 @@ func TestWorkerAddServer(t *testing.T) {
 	}
 }
 
-func foobar(job *Job) ([]byte, error) {
+func foobar(job Job) ([]byte, error) {
 	return nil, nil
 }
 
@@ -42,3 +42,11 @@ func TestWorkerRemoveFunc(t *testing.T) {
 		t.Error(err)
 	}
 }
+
+func TestWork(t *testing.T) {
+	go worker.Work()
+}
+
+func TestWorkerClose(t *testing.T) {
+	worker.Close()
+}