GaoreGo/gearman-go
8
0
Fork 0
forked from yuxh/gearman-go
Code Issues Pull Requests Releases 1 Wiki Activity

Upgrade to weekly.2011-12-14

Browse Source
This commit is contained in:
mikespook 2011-12-28 17:59:44 +08:00
parent 5c13b0923b
commit 85a02820fc
9 changed files with 754 additions and 754 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
example/client.go
View File

@ -22,12 +22,12 @@ func main() {
log.Fatalln(err) log.Fatalln(err)
} else { } else {
log.Println(handle) log.Println(handle)
/*job := <-client.JobQueue job := <-client.JobQueue
if data, err := job.Result(); err != nil { if data, err := job.Result(); err != nil {
log.Fatalln(err) log.Fatalln(err)
} else { } else {
log.Println(string(data)) log.Println(string(data))
}*/ }
} }
known, running, numerator, denominator, err := client.Status(handle) known, running, numerator, denominator, err := client.Status(handle)

75
example/worker.go
View File

@ -1,48 +1,47 @@
package main package main
import ( import (
"gearman" "fmt"
"fmt" "gearman"
"log" "log"
"os" "strings"
"strings"
) )
func ToUpper(job *gearman.WorkerJob) ([]byte, os.Error) { func ToUpper(job *gearman.WorkerJob) ([]byte, error) {
data := []byte(strings.ToUpper(string(job.Data))) data := []byte(strings.ToUpper(string(job.Data)))
return data, nil return data, nil
} }
func main() { func main() {
worker := gearman.NewWorker() worker := gearman.NewWorker()
worker.AddServer("127.0.0.1:4730") worker.AddServer("127.0.0.1:4730")
worker.AddFunction("ToUpper", ToUpper, 0) worker.AddFunction("ToUpper", ToUpper, 0)
worker.AddFunction("ToUpperTimeOut5", ToUpper, 5) worker.AddFunction("ToUpperTimeOut5", ToUpper, 5)
go func() { go func() {
log.Println("start worker") log.Println("start worker")
for { for {
print("cmd: ") print("cmd: ")
var str string var str string
fmt.Scan(&str) fmt.Scan(&str)
switch str { switch str {
case "echo": case "echo":
worker.Echo([]byte("Hello world!")) worker.Echo([]byte("Hello world!"))
var job *gearman.WorkerJob var job *gearman.WorkerJob
for job = <-worker.JobQueue; job.DataType != gearman.ECHO_RES; job = <-worker.JobQueue { for job = <-worker.JobQueue; job.DataType != gearman.ECHO_RES; job = <-worker.JobQueue {
log.Println(job) log.Println(job)
} }
log.Println(string(job.Data)) log.Println(string(job.Data))
case "quit": case "quit":
worker.Close() worker.Close()
return return
case "result": case "result":
job := <-worker.JobQueue job := <-worker.JobQueue
log.Println(string(job.Data)) log.Println(string(job.Data))
default: default:
log.Println("Unknown command") log.Println("Unknown command")
} }
} }
}() }()
worker.Work() worker.Work()
} }

415
src/gearman/client.go
View File

@ -5,12 +5,13 @@
package gearman package gearman
import ( import (
"os" "errors"
"net" "io"
"sync" "net"
// "log" "sync"
"strconv" // "log"
"bytes" "bytes"
"strconv"
) )
/* /*
@ -23,97 +24,97 @@ usage:
*/ */
type Client struct { type Client struct {
mutex sync.Mutex mutex sync.Mutex
conn net.Conn conn net.Conn
JobQueue chan *ClientJob JobQueue chan *ClientJob
incoming chan []byte incoming chan []byte
UId uint32 UId uint32
} }
// Create a new client. // Create a new client.
func NewClient() (client *Client) { func NewClient() (client *Client) {
client = &Client{JobQueue: make(chan *ClientJob, QUEUE_CAP), client = &Client{JobQueue: make(chan *ClientJob, QUEUE_CAP),
incoming: make(chan []byte, QUEUE_CAP), incoming: make(chan []byte, QUEUE_CAP),
UId: 1} UId: 1}
return return
} }
// Add a server. // Add a server.
// In this version, one client connect to one job server. // In this version, one client connect to one job server.
// Sample is better. Plz do the load balancing by your self. // Sample is better. Plz do the load balancing by your self.
func (client *Client) AddServer(addr string) (err os.Error) { func (client *Client) AddServer(addr string) (err error) {
conn, err := net.Dial(TCP, addr) conn, err := net.Dial(TCP, addr)
if err != nil { if err != nil {
return return
} }
client.conn = conn client.conn = conn
return return
} }
// Internal read // Internal read
func (client *Client) read() (data []byte, err os.Error) { func (client *Client) read() (data []byte, err error) {
if len(client.incoming) > 0 { if len(client.incoming) > 0 {
// incoming queue is not empty // incoming queue is not empty
data = <-client.incoming data = <-client.incoming
} else { } else {
// empty queue, read data from socket // empty queue, read data from socket
for { for {
buf := make([]byte, BUFFER_SIZE) buf := make([]byte, BUFFER_SIZE)
var n int var n int
if n, err = client.conn.Read(buf); err != nil { if n, err = client.conn.Read(buf); err != nil {
if err == os.EOF && n == 0 { if err == io.EOF && n == 0 {
break break
} }
return return
} }
data = append(data, buf[0:n]...) data = append(data, buf[0:n]...)
if n < BUFFER_SIZE { if n < BUFFER_SIZE {
break break
} }
} }
} }
// split package // split package
start, end := 0, 4 start, end := 0, 4
tl := len(data) tl := len(data)
for i := 0; i < tl; i++ { for i := 0; i < tl; i++ {
if string(data[start:end]) == RES_STR { if string(data[start:end]) == RES_STR {
l := int(byteToUint32([4]byte{data[start+8], data[start+9], data[start+10], data[start+11]})) l := int(byteToUint32([4]byte{data[start+8], data[start+9], data[start+10], data[start+11]}))
total := l + 12 total := l + 12
if total == tl { if total == tl {
return return
} else { } else {
client.incoming <- data[total:] client.incoming <- data[total:]
data = data[:total] data = data[:total]
return return
} }
} else { } else {
start++ start++
end++ end++
} }
} }
err = os.NewError("Invalid data struct.") err = errors.New("Invalid data struct.")
return return
} }
// Read a job from job server. // Read a job from job server.
// This function will return the job, and add it to the job queue. // This function will return the job, and add it to the job queue.
func (client *Client) ReadJob() (job *ClientJob, err os.Error) { func (client *Client) ReadJob() (job *ClientJob, err error) {
var rel []byte var rel []byte
if rel, err = client.read(); err != nil { if rel, err = client.read(); err != nil {
return return
} }
if job, err = DecodeClientJob(rel); err != nil { if job, err = DecodeClientJob(rel); err != nil {
return return
} else { } else {
switch job.DataType { switch job.DataType {
case ERROR: case ERROR:
_, err = getError(job.Data) _, err = getError(job.Data)
return return
case WORK_DATA, WORK_WARNING, WORK_STATUS, WORK_COMPLETE, WORK_FAIL, WORK_EXCEPTION: case WORK_DATA, WORK_WARNING, WORK_STATUS, WORK_COMPLETE, WORK_FAIL, WORK_EXCEPTION:
client.JobQueue <- job client.JobQueue <- job
} }
} }
return return
} }
// Do the function. // Do the function.
@ -122,157 +123,157 @@ func (client *Client) ReadJob() (job *ClientJob, err os.Error) {
// flag set the job type, include running level: JOB_LOW, JOB_NORMAL, JOB_HIGH, // flag set the job type, include running level: JOB_LOW, JOB_NORMAL, JOB_HIGH,
// and if it is background job: JOB_BG. // and if it is background job: JOB_BG.
// JOB_LOW | JOB_BG means the job is running with low level in background. // JOB_LOW | JOB_BG means the job is running with low level in background.
func (client *Client) Do(funcname string, data []byte, flag byte) (handle string, err os.Error) { func (client *Client) Do(funcname string, data []byte, flag byte) (handle string, err error) {
var datatype uint32 var datatype uint32
if flag&JOB_LOW == JOB_LOW { if flag&JOB_LOW == JOB_LOW {
if flag&JOB_BG == JOB_BG { if flag&JOB_BG == JOB_BG {
datatype = SUBMIT_JOB_LOW_BG datatype = SUBMIT_JOB_LOW_BG
} else { } else {
datatype = SUBMIT_JOB_LOW datatype = SUBMIT_JOB_LOW
} }
} else if flag&JOB_HIGH == JOB_HIGH { } else if flag&JOB_HIGH == JOB_HIGH {
if flag&JOB_BG == JOB_BG { if flag&JOB_BG == JOB_BG {
datatype = SUBMIT_JOB_HIGH_BG datatype = SUBMIT_JOB_HIGH_BG
} else { } else {
datatype = SUBMIT_JOB_HIGH datatype = SUBMIT_JOB_HIGH
} }
} else if flag&JOB_BG == JOB_BG { } else if flag&JOB_BG == JOB_BG {
datatype = SUBMIT_JOB_BG datatype = SUBMIT_JOB_BG
} else { } else {
datatype = SUBMIT_JOB datatype = SUBMIT_JOB
} }
rel := make([]byte, 0, 1024*64) rel := make([]byte, 0, 1024*64)
rel = append(rel, []byte(funcname)...) rel = append(rel, []byte(funcname)...)
rel = append(rel, '\x00') rel = append(rel, '\x00')
client.mutex.Lock() client.mutex.Lock()
uid := strconv.Itoa(int(client.UId)) uid := strconv.Itoa(int(client.UId))
client.UId++ client.UId++
rel = append(rel, []byte(uid)...) rel = append(rel, []byte(uid)...)
client.mutex.Unlock() client.mutex.Unlock()
rel = append(rel, '\x00') rel = append(rel, '\x00')
rel = append(rel, data...) rel = append(rel, data...)
if err = client.WriteJob(NewClientJob(REQ, datatype, rel)); err != nil { if err = client.WriteJob(NewClientJob(REQ, datatype, rel)); err != nil {
return return
} }
var job *ClientJob var job *ClientJob
if job, err = client.readLastJob(JOB_CREATED); err != nil { if job, err = client.readLastJob(JOB_CREATED); err != nil {
return return
} }
handle = string(job.Data) handle = string(job.Data)
go func() { go func() {
if flag&JOB_BG != JOB_BG { if flag&JOB_BG != JOB_BG {
for { for {
if job, err = client.ReadJob(); err != nil { if job, err = client.ReadJob(); err != nil {
return return
} }
switch job.DataType { switch job.DataType {
case WORK_DATA, WORK_WARNING: case WORK_DATA, WORK_WARNING:
case WORK_STATUS: case WORK_STATUS:
case WORK_COMPLETE, WORK_FAIL, WORK_EXCEPTION: case WORK_COMPLETE, WORK_FAIL, WORK_EXCEPTION:
return return
} }
} }
} }
}() }()
return return
} }
// Internal read last job // Internal read last job
func (client *Client) readLastJob(datatype uint32) (job *ClientJob, err os.Error) { func (client *Client) readLastJob(datatype uint32) (job *ClientJob, err error) {
for { for {
if job, err = client.ReadJob(); err != nil { if job, err = client.ReadJob(); err != nil {
return return
} }
if job.DataType == datatype { if job.DataType == datatype {
break break
} }
} }
if job.DataType != datatype { if job.DataType != datatype {
err = os.NewError("No job got.") err = errors.New("No job got.")
} }
return return
} }
// Get job status from job server. // Get job status from job server.
// !!!Not fully tested.!!! // !!!Not fully tested.!!!
func (client *Client) Status(handle string) (known, running bool, numerator, denominator uint, err os.Error) { func (client *Client) Status(handle string) (known, running bool, numerator, denominator uint64, err error) {
if err = client.WriteJob(NewClientJob(REQ, GET_STATUS, []byte(handle))); err != nil { if err = client.WriteJob(NewClientJob(REQ, GET_STATUS, []byte(handle))); err != nil {
return return
} }
var job *ClientJob var job *ClientJob
if job, err = client.readLastJob(STATUS_RES); err != nil { if job, err = client.readLastJob(STATUS_RES); err != nil {
return return
} }
data := bytes.SplitN(job.Data, []byte{'\x00'}, 5) data := bytes.SplitN(job.Data, []byte{'\x00'}, 5)
if len(data) != 5 { if len(data) != 5 {
err = os.NewError("Data Error.") err = errors.New("Data Error.")
return return
} }
if handle != string(data[0]) { if handle != string(data[0]) {
err = os.NewError("Invalid handle.") err = errors.New("Invalid handle.")
return return
} }
known = data[1][0] == '1' known = data[1][0] == '1'
running = data[2][0] == '1' running = data[2][0] == '1'
if numerator, err = strconv.Atoui(string(data[3][0])); err != nil { if numerator, err = strconv.ParseUint(string(data[3][0]), 10, 0); err != nil {
return return
} }
if denominator, err = strconv.Atoui(string(data[4][0])); err != nil { if denominator, err = strconv.ParseUint(string(data[4][0]), 10, 0); err != nil {
return return
} }
return return
} }
// Send a something out, get the samething back. // Send a something out, get the samething back.
func (client *Client) Echo(data []byte) (echo []byte, err os.Error) { func (client *Client) Echo(data []byte) (echo []byte, err error) {
if err = client.WriteJob(NewClientJob(REQ, ECHO_REQ, data)); err != nil { if err = client.WriteJob(NewClientJob(REQ, ECHO_REQ, data)); err != nil {
return return
} }
var job *ClientJob var job *ClientJob
if job, err = client.readLastJob(ECHO_RES); err != nil { if job, err = client.readLastJob(ECHO_RES); err != nil {
return return
} }
echo = job.Data echo = job.Data
return return
} }
// Get the last job. // Get the last job.
// the job means a network package. // the job means a network package.
// Normally, it is the job executed result. // Normally, it is the job executed result.
func (client *Client) LastJob() (job *ClientJob) { func (client *Client) LastJob() (job *ClientJob) {
if l := len(client.JobQueue); l != 1 { if l := len(client.JobQueue); l != 1 {
if l == 0 { if l == 0 {
return return
} }
for i := 0; i < l-1; i++ { for i := 0; i < l-1; i++ {
<-client.JobQueue <-client.JobQueue
} }
} }
return <-client.JobQueue return <-client.JobQueue
} }
// Send the job to job server. // Send the job to job server.
func (client *Client) WriteJob(job *ClientJob) (err os.Error) { func (client *Client) WriteJob(job *ClientJob) (err error) {
return client.write(job.Encode()) return client.write(job.Encode())
} }
// Internal write // Internal write
func (client *Client) write(buf []byte) (err os.Error) { func (client *Client) write(buf []byte) (err error) {
var n int var n int
for i := 0; i < len(buf); i += n { for i := 0; i < len(buf); i += n {
n, err = client.conn.Write(buf[i:]) n, err = client.conn.Write(buf[i:])
if err != nil { if err != nil {
return return
} }
} }
return return
} }
// Close. // Close.
func (client *Client) Close() (err os.Error) { func (client *Client) Close() (err error) {
err = client.conn.Close() err = client.conn.Close()
close(client.JobQueue) close(client.JobQueue)
return return
} }

140
src/gearman/clientjob.go
View File

@ -5,95 +5,95 @@
package gearman package gearman
import ( import (
"os" "bytes"
"bytes" "errors"
// "log" // "log"
) )
// Client side job // Client side job
type ClientJob struct { type ClientJob struct {
Data []byte Data []byte
Handle, UniqueId string Handle, UniqueId string
magicCode, DataType uint32 magicCode, DataType uint32
} }
// Create a new job // Create a new job
func NewClientJob(magiccode, datatype uint32, data []byte) (job *ClientJob) { func NewClientJob(magiccode, datatype uint32, data []byte) (job *ClientJob) {
return &ClientJob{magicCode: magiccode, return &ClientJob{magicCode: magiccode,
DataType: datatype, DataType: datatype,
Data: data} Data: data}
} }
// Decode a job from byte slice // Decode a job from byte slice
func DecodeClientJob(data []byte) (job *ClientJob, err os.Error) { func DecodeClientJob(data []byte) (job *ClientJob, err error) {
if len(data) < 12 { if len(data) < 12 {
err = os.NewError("Data length is too small.") err = errors.New("Data length is too small.")
return return
} }
datatype := byteToUint32([4]byte{data[4], data[5], data[6], data[7]}) datatype := byteToUint32([4]byte{data[4], data[5], data[6], data[7]})
l := byteToUint32([4]byte{data[8], data[9], data[10], data[11]}) l := byteToUint32([4]byte{data[8], data[9], data[10], data[11]})
if len(data[12:]) != int(l) { if len(data[12:]) != int(l) {
err = os.NewError("Invalid data length.") err = errors.New("Invalid data length.")
return return
} }
data = data[12:] data = data[12:]
job = NewClientJob(RES, datatype, data) job = NewClientJob(RES, datatype, data)
return return
} }
// Encode a job to byte slice // Encode a job to byte slice
func (job *ClientJob) Encode() (data []byte) { func (job *ClientJob) Encode() (data []byte) {
magiccode := uint32ToByte(job.magicCode) magiccode := uint32ToByte(job.magicCode)
datatype := uint32ToByte(job.DataType) datatype := uint32ToByte(job.DataType)
data = make([]byte, 0, 1024*64) data = make([]byte, 0, 1024*64)
data = append(data, magiccode[:]...) data = append(data, magiccode[:]...)
data = append(data, datatype[:]...) data = append(data, datatype[:]...)
l := len(job.Data) l := len(job.Data)
datalength := uint32ToByte(uint32(l)) datalength := uint32ToByte(uint32(l))
data = append(data, datalength[:]...) data = append(data, datalength[:]...)
data = append(data, job.Data...) data = append(data, job.Data...)
return return
} }
// Extract the job's result. // Extract the job's result.
func (job *ClientJob) Result() (data []byte, err os.Error) { func (job *ClientJob) Result() (data []byte, err error) {
switch job.DataType { switch job.DataType {
case WORK_FAIL: case WORK_FAIL:
job.Handle = string(job.Data) job.Handle = string(job.Data)
err = os.NewError("Work fail.") err = errors.New("Work fail.")
return return
case WORK_EXCEPTION: case WORK_EXCEPTION:
err = os.NewError("Work exception.") err = errors.New("Work exception.")
fallthrough fallthrough
case WORK_COMPLETE: case WORK_COMPLETE:
s := bytes.SplitN(job.Data, []byte{'\x00'}, 2) s := bytes.SplitN(job.Data, []byte{'\x00'}, 2)
if len(s) != 2 { if len(s) != 2 {
err = os.NewError("Invalid data.") err = errors.New("Invalid data.")
return return
} }
job.Handle = string(s[0]) job.Handle = string(s[0])
data = s[1] data = s[1]
default: default:
err = os.NewError("The job is not a result.") err = errors.New("The job is not a result.")
} }
return return
} }
// Extract the job's update // Extract the job's update
func (job *ClientJob) Update() (data []byte, err os.Error) { func (job *ClientJob) Update() (data []byte, err error) {
if job.DataType != WORK_DATA && job.DataType != WORK_WARNING { if job.DataType != WORK_DATA && job.DataType != WORK_WARNING {
err = os.NewError("The job is not a update.") err = errors.New("The job is not a update.")
return return
} }
s := bytes.SplitN(job.Data, []byte{'\x00'}, 2) s := bytes.SplitN(job.Data, []byte{'\x00'}, 2)
if len(s) != 2 { if len(s) != 2 {
err = os.NewError("Invalid data.") err = errors.New("Invalid data.")
return return
} }
if job.DataType == WORK_WARNING { if job.DataType == WORK_WARNING {
err = os.NewError("Work warning") err = errors.New("Work warning")
} }
job.Handle = string(s[0]) job.Handle = string(s[0])
data = s[1] data = s[1]
return return
} }

159
src/gearman/gearman.go
View File

@ -10,101 +10,102 @@ The protocol was implemented by native way.
package gearman package gearman
import ( import (
"bytes" "bytes"
"os" "errors"
"syscall"
) )
const ( const (
// tcp4 is tested. You can modify this to 'tcp' for both ipv4 and ipv6, // tcp4 is tested. You can modify this to 'tcp' for both ipv4 and ipv6,
// or 'tcp6' only for ipv6. // or 'tcp6' only for ipv6.
TCP = "tcp4" TCP = "tcp4"
// the number limited for job servers. // the number limited for job servers.
WORKER_SERVER_CAP = 32 WORKER_SERVER_CAP = 32
// the number limited for functions. // the number limited for functions.
WORKER_FUNCTION_CAP = 512 WORKER_FUNCTION_CAP = 512
// queue size // queue size
QUEUE_CAP = 512 QUEUE_CAP = 512
// read buffer size // read buffer size
BUFFER_SIZE = 1024 BUFFER_SIZE = 1024
// \x00REQ // \x00REQ
REQ = 5391697 REQ = 5391697
REQ_STR = "\x00REQ" REQ_STR = "\x00REQ"
// \x00RES // \x00RES
RES = 5391699 RES = 5391699
RES_STR = "\x00RES" RES_STR = "\x00RES"
// package data type // package data type
CAN_DO = 1 CAN_DO = 1
CANT_DO = 2 CANT_DO = 2
RESET_ABILITIES = 3 RESET_ABILITIES = 3
PRE_SLEEP = 4 PRE_SLEEP = 4
NOOP = 6 NOOP = 6
JOB_CREATED = 8 JOB_CREATED = 8
GRAB_JOB = 9 GRAB_JOB = 9
NO_JOB = 10 NO_JOB = 10
JOB_ASSIGN = 11 JOB_ASSIGN = 11
WORK_STATUS = 12 WORK_STATUS = 12
WORK_COMPLETE = 13 WORK_COMPLETE = 13
WORK_FAIL = 14 WORK_FAIL = 14
GET_STATUS = 15 GET_STATUS = 15
ECHO_REQ = 16 ECHO_REQ = 16
ECHO_RES = 17 ECHO_RES = 17
ERROR = 19 ERROR = 19
STATUS_RES = 20 STATUS_RES = 20
SET_CLIENT_ID = 22 SET_CLIENT_ID = 22
CAN_DO_TIMEOUT = 23 CAN_DO_TIMEOUT = 23
WORK_EXCEPTION = 25 WORK_EXCEPTION = 25
WORK_DATA = 28 WORK_DATA = 28
WORK_WARNING = 29 WORK_WARNING = 29
GRAB_JOB_UNIQ = 30 GRAB_JOB_UNIQ = 30
JOB_ASSIGN_UNIQ = 31 JOB_ASSIGN_UNIQ = 31
SUBMIT_JOB = 7 SUBMIT_JOB = 7
SUBMIT_JOB_BG = 18 SUBMIT_JOB_BG = 18
SUBMIT_JOB_HIGH = 21 SUBMIT_JOB_HIGH = 21
SUBMIT_JOB_HIGH_BG = 32 SUBMIT_JOB_HIGH_BG = 32
SUBMIT_JOB_LOW = 33 SUBMIT_JOB_LOW = 33
SUBMIT_JOB_LOW_BG = 34 SUBMIT_JOB_LOW_BG = 34
// Job type // Job type
// JOB_NORMAL | JOB_BG means a normal level job run in background // JOB_NORMAL | JOB_BG means a normal level job run in background
// normal level // normal level
JOB_NORMAL = 0 JOB_NORMAL = 0
// background job // background job
JOB_BG = 1 JOB_BG = 1
// low level // low level
JOB_LOW = 2 JOB_LOW = 2
// high level // high level
JOB_HIGH = 4 JOB_HIGH = 4
) )
// Extract the error message // Extract the error message
func getError(data []byte) (eno os.Errno, err os.Error) { func getError(data []byte) (eno syscall.Errno, err error) {
rel := bytes.SplitN(data, []byte{'\x00'}, 2) rel := bytes.SplitN(data, []byte{'\x00'}, 2)
if len(rel) != 2 { if len(rel) != 2 {
err = os.NewError("The input is not a error data.") err = errors.New("The input is not a error data.")
return return
} }
l := len(rel[0]) l := len(rel[0])
eno = os.Errno(byteToUint32([4]byte{rel[0][l-4], rel[0][l-3], rel[0][l-2], rel[0][l-1]})) eno = syscall.Errno(byteToUint32([4]byte{rel[0][l-4], rel[0][l-3], rel[0][l-2], rel[0][l-1]}))
err = os.NewError(string(rel[1])) err = errors.New(string(rel[1]))
return return
} }
// Decode [4]byte to uint32 // Decode [4]byte to uint32
func byteToUint32(buf [4]byte) uint32 { func byteToUint32(buf [4]byte) uint32 {
return uint32(buf[0])<<24 + return uint32(buf[0])<<24 +
uint32(buf[1])<<16 + uint32(buf[1])<<16 +
uint32(buf[2])<<8 + uint32(buf[2])<<8 +
uint32(buf[3]) uint32(buf[3])
} }
// Encode uint32 to [4]byte // Encode uint32 to [4]byte
func uint32ToByte(i uint32) (data [4]byte) { func uint32ToByte(i uint32) (data [4]byte) {
data[0] = byte((i >> 24) & 0xff) data[0] = byte((i >> 24) & 0xff)
data[1] = byte((i >> 16) & 0xff) data[1] = byte((i >> 16) & 0xff)
data[2] = byte((i >> 8) & 0xff) data[2] = byte((i >> 8) & 0xff)
data[3] = byte(i & 0xff) data[3] = byte(i & 0xff)
return return
} }

202
src/gearman/jobagent.go
View File

@ -5,127 +5,129 @@
package gearman package gearman
import ( import (
"net" "errors"
"os" "io"
// "log" "net"
// "log"
) )
// The agent of job server. // The agent of job server.
type jobAgent struct { type jobAgent struct {
conn net.Conn conn net.Conn
worker *Worker worker *Worker
running bool running bool
incoming chan []byte incoming chan []byte
} }
// Create the agent of job server. // Create the agent of job server.
func newJobAgent(addr string, worker *Worker) (jobagent *jobAgent, err os.Error) { func newJobAgent(addr string, worker *Worker) (jobagent *jobAgent, err error) {
conn, err := net.Dial(TCP, addr) conn, err := net.Dial(TCP, addr)
if err != nil { if err != nil {
return nil, err return nil, err
} }
jobagent = &jobAgent{conn: conn, worker: worker, running: true, incoming: make(chan []byte, QUEUE_CAP)} jobagent = &jobAgent{conn: conn, worker: worker, running: true, incoming: make(chan []byte, QUEUE_CAP)}
return jobagent, err return jobagent, err
} }
// Internal read // Internal read
func (agent *jobAgent) read() (data []byte, err os.Error) { func (agent *jobAgent) read() (data []byte, err error) {
if len(agent.incoming) > 0 { if len(agent.incoming) > 0 {
// incoming queue is not empty // incoming queue is not empty
data = <-agent.incoming data = <-agent.incoming
} else { } else {
for { for {
buf := make([]byte, BUFFER_SIZE) buf := make([]byte, BUFFER_SIZE)
var n int var n int
if n, err = agent.conn.Read(buf); err != nil { if n, err = agent.conn.Read(buf); err != nil {
if err == os.EOF && n == 0 { if err == io.EOF && n == 0 {
err = nil err = nil
return return
} }
return return
} }
data = append(data, buf[0:n]...) data = append(data, buf[0:n]...)
if n < BUFFER_SIZE { if n < BUFFER_SIZE {
break break
} }
} }
} }
// split package // split package
start := 0 start := 0
tl := len(data) tl := len(data)
for i := 0; i < tl; i++ { for i := 0; i < tl; i++ {
if string(data[start:start+4]) == RES_STR { if string(data[start:start+4]) == RES_STR {
l := int(byteToUint32([4]byte{data[start+8], data[start+9], data[start+10], data[start+11]})) l := int(byteToUint32([4]byte{data[start+8], data[start+9], data[start+10], data[start+11]}))
total := l + 12 total := l + 12
if total == tl { if total == tl {
return return
} else { } else {
agent.incoming <- data[total:] agent.incoming <- data[total:]
data = data[:total] data = data[:total]
return return
} }
} else { } else {
start++ start++
} }
} }
err = os.NewError("Invalid data struct.") err = errors.New("Invalid data struct.")
return return
} }
// Main loop. // Main loop.
func (agent *jobAgent) Work() { func (agent *jobAgent) Work() {
noop := true noop := true
for agent.running { for agent.running {
// got noop msg and incoming queue is zero, grab job // got noop msg and incoming queue is zero, grab job
if noop && len(agent.incoming) == 0 { if noop && len(agent.incoming) == 0 {
agent.WriteJob(NewWorkerJob(REQ, GRAB_JOB, nil)) agent.WriteJob(NewWorkerJob(REQ, GRAB_JOB, nil))
} }
rel, err := agent.read() rel, err := agent.read()
if err != nil { if err != nil {
agent.worker.ErrQueue <- err agent.worker.ErrQueue <- err
continue continue
} }
job, err := DecodeWorkerJob(rel) job, err := DecodeWorkerJob(rel)
if err != nil { if err != nil {
agent.worker.ErrQueue <- err agent.worker.ErrQueue <- err
continue continue
} else { } else {
switch job.DataType { switch job.DataType {
case NOOP: case NOOP:
noop = true noop = true
case NO_JOB: case NO_JOB:
noop = false noop = false
agent.WriteJob(NewWorkerJob(REQ, PRE_SLEEP, nil)) agent.WriteJob(NewWorkerJob(REQ, PRE_SLEEP, nil))
case ECHO_RES, JOB_ASSIGN_UNIQ, JOB_ASSIGN: case ECHO_RES, JOB_ASSIGN_UNIQ, JOB_ASSIGN:
job.agent = agent job.agent = agent
agent.worker.incoming <- job agent.worker.incoming <- job
} }
} }
} }
return return
} }
// Send a job to the job server. // Send a job to the job server.
func (agent *jobAgent) WriteJob(job *WorkerJob) (err os.Error) { func (agent *jobAgent) WriteJob(job *WorkerJob) (err error) {
return agent.write(job.Encode()) return agent.write(job.Encode())
} }
// Internal write the encoded job. // Internal write the encoded job.
func (agent *jobAgent) write(buf []byte) (err os.Error) { func (agent *jobAgent) write(buf []byte) (err error) {
var n int var n int
for i := 0; i < len(buf); i += n { for i := 0; i < len(buf); i += n {
n, err = agent.conn.Write(buf[i:]) n, err = agent.conn.Write(buf[i:])
if err != nil { if err != nil {
return err return err
} }
} }
return return
} }
// Close. // Close.
func (agent *jobAgent) Close() (err os.Error) { func (agent *jobAgent) Close() (err error) {
agent.running = false agent.running = false
close(agent.incoming) close(agent.incoming)
err = agent.conn.Close() err = agent.conn.Close()
return return
} }

330
src/gearman/worker.go
View File

@ -5,13 +5,13 @@
package gearman package gearman
import ( import (
"os" "bytes"
"sync" "errors"
"bytes" "sync"
) )
// The definition of the callback function. // The definition of the callback function.
type JobFunction func(job *WorkerJob) ([]byte, os.Error) type JobFunction func(job *WorkerJob) ([]byte, error)
// Map for added function. // Map for added function.
type JobFunctionMap map[string]JobFunction type JobFunctionMap map[string]JobFunction
@ -34,124 +34,124 @@ func foobar(job *WorkerJob) (data []byte, err os.Error) {
} }
*/ */
type Worker struct { type Worker struct {
clients []*jobAgent clients []*jobAgent
functions JobFunctionMap functions JobFunctionMap
running bool running bool
incoming chan *WorkerJob incoming chan *WorkerJob
mutex sync.Mutex mutex sync.Mutex
JobQueue chan *WorkerJob JobQueue chan *WorkerJob
ErrQueue chan os.Error ErrQueue chan error
} }
// Get a new worker // Get a new worker
func NewWorker() (worker *Worker) { func NewWorker() (worker *Worker) {
worker = &Worker{ worker = &Worker{
// job server list // job server list
clients: make([]*jobAgent, 0, WORKER_SERVER_CAP), clients: make([]*jobAgent, 0, WORKER_SERVER_CAP),
// function list // function list
functions: make(JobFunctionMap), functions: make(JobFunctionMap),
incoming: make(chan *WorkerJob, QUEUE_CAP), incoming: make(chan *WorkerJob, QUEUE_CAP),
JobQueue: make(chan *WorkerJob, QUEUE_CAP), JobQueue: make(chan *WorkerJob, QUEUE_CAP),
ErrQueue: make(chan os.Error, QUEUE_CAP), ErrQueue: make(chan error, QUEUE_CAP),
running: true, running: true,
} }
return return
} }
// Add a server. The addr should be 'host:port' format. // Add a server. The addr should be 'host:port' format.
// The connection is established at this time. // The connection is established at this time.
func (worker *Worker) AddServer(addr string) (err os.Error) { func (worker *Worker) AddServer(addr string) (err error) {
worker.mutex.Lock() worker.mutex.Lock()
defer worker.mutex.Unlock() defer worker.mutex.Unlock()
if len(worker.clients) == cap(worker.clients) { if len(worker.clients) == cap(worker.clients) {
return os.NewError("To many servers added.") return errors.New("To many servers added.")
} }
// Create a new job server's client as a agent of server // Create a new job server's client as a agent of server
server, err := newJobAgent(addr, worker) server, err := newJobAgent(addr, worker)
if err != nil { if err != nil {
return err return err
} }
n := len(worker.clients) n := len(worker.clients)
worker.clients = worker.clients[0 : n+1] worker.clients = worker.clients[0 : n+1]
worker.clients[n] = server worker.clients[n] = server
return return
} }
// Add a function. // Add a function.
// Plz added job servers first, then functions. // Plz added job servers first, then functions.
// The API will tell every connected job server that 'I can do this' // The API will tell every connected job server that 'I can do this'
func (worker *Worker) AddFunction(funcname string, func (worker *Worker) AddFunction(funcname string,
f JobFunction, timeout uint32) (err os.Error) { f JobFunction, timeout uint32) (err error) {
if len(worker.clients) < 1 { if len(worker.clients) < 1 {
return os.NewError("Did not connect to Job Server.") return errors.New("Did not connect to Job Server.")
} }
worker.mutex.Lock() worker.mutex.Lock()
defer worker.mutex.Unlock() defer worker.mutex.Unlock()
worker.functions[funcname] = f worker.functions[funcname] = f
var datatype uint32 var datatype uint32
var data []byte var data []byte
if timeout == 0 { if timeout == 0 {
datatype = CAN_DO datatype = CAN_DO
data = []byte(funcname) data = []byte(funcname)
} else { } else {
datatype = CAN_DO_TIMEOUT datatype = CAN_DO_TIMEOUT
data = []byte(funcname + "\x00") data = []byte(funcname + "\x00")
t := uint32ToByte(timeout) t := uint32ToByte(timeout)
data = append(data, t[:]...) data = append(data, t[:]...)
} }
job := NewWorkerJob(REQ, datatype, data) job := NewWorkerJob(REQ, datatype, data)
worker.WriteJob(job) worker.WriteJob(job)
return return
} }
// Remove a function. // Remove a function.
// Tell job servers 'I can not do this now' at the same time. // Tell job servers 'I can not do this now' at the same time.
func (worker *Worker) RemoveFunction(funcname string) (err os.Error) { func (worker *Worker) RemoveFunction(funcname string) (err error) {
worker.mutex.Lock() worker.mutex.Lock()
defer worker.mutex.Unlock() defer worker.mutex.Unlock()
if worker.functions[funcname] == nil { if worker.functions[funcname] == nil {
return os.NewError("No function named: " + funcname) return errors.New("No function named: " + funcname)
} }
worker.functions[funcname] = nil, false delete(worker.functions, funcname)
job := NewWorkerJob(REQ, CANT_DO, []byte(funcname)) job := NewWorkerJob(REQ, CANT_DO, []byte(funcname))
worker.WriteJob(job) worker.WriteJob(job)
return return
} }
// Main loop // Main loop
func (worker *Worker) Work() { func (worker *Worker) Work() {
for _, v := range worker.clients { for _, v := range worker.clients {
go v.Work() go v.Work()
} }
for worker.running { for worker.running {
select { select {
case job := <-worker.incoming: case job := <-worker.incoming:
if job == nil { if job == nil {
break break
} }
switch job.DataType { switch job.DataType {
case NO_JOB: case NO_JOB:
// do nothing // do nothing
case ERROR: case ERROR:
_, err := getError(job.Data) _, err := getError(job.Data)
worker.ErrQueue <- err worker.ErrQueue <- err
case JOB_ASSIGN, JOB_ASSIGN_UNIQ: case JOB_ASSIGN, JOB_ASSIGN_UNIQ:
go func() { go func() {
if err := worker.exec(job); err != nil { if err := worker.exec(job); err != nil {
worker.ErrQueue <- err worker.ErrQueue <- err
} }
}() }()
default: default:
worker.JobQueue <- job worker.JobQueue <- job
} }
} }
} }
} }
// Get the last job in queue. // Get the last job in queue.
@ -159,98 +159,98 @@ func (worker *Worker) Work() {
// the last one will be returned, // the last one will be returned,
// the others will be lost. // the others will be lost.
func (worker *Worker) LastJob() (job *WorkerJob) { func (worker *Worker) LastJob() (job *WorkerJob) {
if l := len(worker.JobQueue); l != 1 { if l := len(worker.JobQueue); l != 1 {
if l == 0 { if l == 0 {
return return
} }
for i := 0; i < l-1; i++ { for i := 0; i < l-1; i++ {
<-worker.JobQueue <-worker.JobQueue
} }
} }
return <-worker.JobQueue return <-worker.JobQueue
} }
// Close. // Close.
func (worker *Worker) Close() (err os.Error) { func (worker *Worker) Close() (err error) {
worker.running = false worker.running = false
for _, v := range worker.clients { for _, v := range worker.clients {
err = v.Close() err = v.Close()
} }
close(worker.incoming) close(worker.incoming)
return err return err
} }
// Write a job to job server. // Write a job to job server.
// Here, the job's mean is not the oraginal mean. // 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. // Just looks like a network package for job's result or tell job server, there was a fail.
func (worker *Worker) WriteJob(job *WorkerJob) (err os.Error) { func (worker *Worker) WriteJob(job *WorkerJob) (err error) {
e := make(chan os.Error) e := make(chan error)
for _, v := range worker.clients { for _, v := range worker.clients {
go func() { go func() {
e <- v.WriteJob(job) e <- v.WriteJob(job)
}() }()
} }
return <-e return <-e
} }
// Send a something out, get the samething back. // Send a something out, get the samething back.
func (worker *Worker) Echo(data []byte) (err os.Error) { func (worker *Worker) Echo(data []byte) (err error) {
job := NewWorkerJob(REQ, ECHO_REQ, data) job := NewWorkerJob(REQ, ECHO_REQ, data)
return worker.WriteJob(job) return worker.WriteJob(job)
} }
// Remove all of functions. // Remove all of functions.
// Both from the worker or job servers. // Both from the worker or job servers.
func (worker *Worker) Reset() (err os.Error) { func (worker *Worker) Reset() (err error) {
job := NewWorkerJob(REQ, RESET_ABILITIES, nil) job := NewWorkerJob(REQ, RESET_ABILITIES, nil)
err = worker.WriteJob(job) err = worker.WriteJob(job)
worker.functions = make(JobFunctionMap) worker.functions = make(JobFunctionMap)
return return
} }
// Set the worker's unique id. // Set the worker's unique id.
func (worker *Worker) SetId(id string) (err os.Error) { func (worker *Worker) SetId(id string) (err error) {
job := NewWorkerJob(REQ, SET_CLIENT_ID, []byte(id)) job := NewWorkerJob(REQ, SET_CLIENT_ID, []byte(id))
return worker.WriteJob(job) return worker.WriteJob(job)
} }
// Execute the job. And send back the result. // Execute the job. And send back the result.
func (worker *Worker) exec(job *WorkerJob) (err os.Error) { func (worker *Worker) exec(job *WorkerJob) (err error) {
var limit int var limit int
if job.DataType == JOB_ASSIGN { if job.DataType == JOB_ASSIGN {
limit = 3 limit = 3
} else { } else {
limit = 4 limit = 4
} }
jobdata := bytes.SplitN(job.Data, []byte{'\x00'}, limit) jobdata := bytes.SplitN(job.Data, []byte{'\x00'}, limit)
job.Handle = string(jobdata[0]) job.Handle = string(jobdata[0])
funcname := string(jobdata[1]) funcname := string(jobdata[1])
if job.DataType == JOB_ASSIGN { if job.DataType == JOB_ASSIGN {
job.Data = jobdata[2] job.Data = jobdata[2]
} else { } else {
job.UniqueId = string(jobdata[2]) job.UniqueId = string(jobdata[2])
job.Data = jobdata[3] job.Data = jobdata[3]
} }
f := worker.functions[funcname] f := worker.functions[funcname]
if f == nil { if f == nil {
return os.NewError("function is nil") return errors.New("function is nil")
} }
result, err := f(job) result, err := f(job)
var datatype uint32 var datatype uint32
if err == nil { if err == nil {
datatype = WORK_COMPLETE datatype = WORK_COMPLETE
} else { } else {
if result == nil { if result == nil {
datatype = WORK_FAIL datatype = WORK_FAIL
} else { } else {
datatype = WORK_EXCEPTION datatype = WORK_EXCEPTION
} }
} }
job.magicCode = REQ job.magicCode = REQ
job.DataType = datatype job.DataType = datatype
job.Data = result job.Data = result
worker.WriteJob(job) worker.WriteJob(job)
return return
} }

71
src/gearman/worker_test.go
View File

@ -1,49 +1,46 @@
package gearman package gearman
import ( import "testing"
"testing"
"os"
)
var worker *Worker var worker *Worker
func init() { func init() {
worker = NewWorker() worker = NewWorker()
} }
func TestWorkerAddServer(t *testing.T) { func TestWorkerAddServer(t *testing.T) {
t.Log("Add local server 127.0.0.1:4730.") 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("127.0.0.1:4730"); err != nil {
t.Error(err) t.Error(err)
} }
if l := len(worker.clients); l != 1 { if l := len(worker.clients); l != 1 {
t.Log(worker.clients) t.Log(worker.clients)
t.Error("The length of server list should be 1.") t.Error("The length of server list should be 1.")
} }
} }
func foobar(job *WorkerJob) ([]byte, os.Error) { func foobar(job *WorkerJob) ([]byte, error) {
return nil, nil return nil, nil
} }
func TestWorkerAddFunction(t *testing.T) { func TestWorkerAddFunction(t *testing.T) {
if err := worker.AddFunction("foobar", foobar, 0); err != nil { if err := worker.AddFunction("foobar", foobar, 0); err != nil {
t.Error(err) t.Error(err)
} }
if err := worker.AddFunction("timeout", foobar, 5); err != nil { if err := worker.AddFunction("timeout", foobar, 5); err != nil {
t.Error(err) t.Error(err)
} }
if l := len(worker.functions); l != 2 { if l := len(worker.functions); l != 2 {
t.Log(worker.functions) t.Log(worker.functions)
t.Errorf("The length of function map should be %d.", 2) t.Errorf("The length of function map should be %d.", 2)
} }
} }
func TestWorkerEcho(t *testing.T) { func TestWorkerEcho(t *testing.T) {
if err := worker.Echo([]byte("Hello World")); err != nil { if err := worker.Echo([]byte("Hello World")); err != nil {
t.Error(err) t.Error(err)
} }
} }
/* /*
func TestWorkerResult(t *testing.T) { func TestWorkerResult(t *testing.T) {
@ -56,19 +53,19 @@ func TestWorkerResult(t *testing.T) {
*/ */
func TestWorkerRemoveFunction(t *testing.T) { func TestWorkerRemoveFunction(t *testing.T) {
if err := worker.RemoveFunction("foobar"); err != nil { if err := worker.RemoveFunction("foobar"); err != nil {
t.Error(err) t.Error(err)
} }
} }
func TestWorkerReset(t *testing.T) { func TestWorkerReset(t *testing.T) {
if err := worker.Reset(); err != nil { if err := worker.Reset(); err != nil {
t.Error(err) t.Error(err)
} }
} }
func TestWorkerClose(t *testing.T) { func TestWorkerClose(t *testing.T) {
if err := worker.Close(); err != nil { if err := worker.Close(); err != nil {
t.Error(err) t.Error(err)
} }
} }

112
src/gearman/workerjob.go
View File

@ -5,83 +5,83 @@
package gearman package gearman
import ( import (
"os" "errors"
"strconv" "strconv"
) )
// Worker side job // Worker side job
type WorkerJob struct { type WorkerJob struct {
Data []byte Data []byte
Handle, UniqueId string Handle, UniqueId string
agent *jobAgent agent *jobAgent
magicCode, DataType uint32 magicCode, DataType uint32
} }
// Create a new job // Create a new job
func NewWorkerJob(magiccode, datatype uint32, data []byte) (job *WorkerJob) { func NewWorkerJob(magiccode, datatype uint32, data []byte) (job *WorkerJob) {
return &WorkerJob{magicCode: magiccode, return &WorkerJob{magicCode: magiccode,
DataType: datatype, DataType: datatype,
Data: data} Data: data}
} }
// Decode job from byte slice // Decode job from byte slice
func DecodeWorkerJob(data []byte) (job *WorkerJob, err os.Error) { func DecodeWorkerJob(data []byte) (job *WorkerJob, err error) {
if len(data) < 12 { if len(data) < 12 {
err = os.NewError("Data length is too small.") err = errors.New("Data length is too small.")
return return
} }
datatype := byteToUint32([4]byte{data[4], data[5], data[6], data[7]}) datatype := byteToUint32([4]byte{data[4], data[5], data[6], data[7]})
l := byteToUint32([4]byte{data[8], data[9], data[10], data[11]}) l := byteToUint32([4]byte{data[8], data[9], data[10], data[11]})
if len(data[12:]) != int(l) { if len(data[12:]) != int(l) {
err = os.NewError("Invalid data length.") err = errors.New("Invalid data length.")
return return
} }
data = data[12:] data = data[12:]
job = NewWorkerJob(RES, datatype, data) job = NewWorkerJob(RES, datatype, data)
return return
} }
// Encode a job to byte slice // Encode a job to byte slice
func (job *WorkerJob) Encode() (data []byte) { func (job *WorkerJob) Encode() (data []byte) {
magiccode := uint32ToByte(job.magicCode) magiccode := uint32ToByte(job.magicCode)
datatype := uint32ToByte(job.DataType) datatype := uint32ToByte(job.DataType)
data = make([]byte, 0, 1024*64) data = make([]byte, 0, 1024*64)
data = append(data, magiccode[:]...) data = append(data, magiccode[:]...)
data = append(data, datatype[:]...) data = append(data, datatype[:]...)
data = append(data, []byte{0, 0, 0, 0}...) data = append(data, []byte{0, 0, 0, 0}...)
l := len(job.Data) l := len(job.Data)
if job.Handle != "" { if job.Handle != "" {
data = append(data, []byte(job.Handle)...) data = append(data, []byte(job.Handle)...)
data = append(data, 0) data = append(data, 0)
l += len(job.Handle) + 1 l += len(job.Handle) + 1
} }
data = append(data, job.Data...) data = append(data, job.Data...)
datalength := uint32ToByte(uint32(l)) datalength := uint32ToByte(uint32(l))
copy(data[8:12], datalength[:]) copy(data[8:12], datalength[:])
return return
} }
// Send some datas to client. // Send some datas to client.
// Using this in a job's executing. // Using this in a job's executing.
func (job *WorkerJob) UpdateData(data []byte, iswaring bool) (err os.Error) { func (job *WorkerJob) UpdateData(data []byte, iswaring bool) (err error) {
result := append([]byte(job.Handle), 0) result := append([]byte(job.Handle), 0)
result = append(result, data...) result = append(result, data...)
var datatype uint32 var datatype uint32
if iswaring { if iswaring {
datatype = WORK_WARNING datatype = WORK_WARNING
} else { } else {
datatype = WORK_DATA datatype = WORK_DATA
} }
return job.agent.WriteJob(NewWorkerJob(REQ, datatype, result)) return job.agent.WriteJob(NewWorkerJob(REQ, datatype, result))
} }
// Update status. // Update status.
// Tall client how many percent job has been executed. // Tall client how many percent job has been executed.
func (job *WorkerJob) UpdateStatus(numerator, denominator int) (err os.Error) { func (job *WorkerJob) UpdateStatus(numerator, denominator int) (err error) {
n := []byte(strconv.Itoa(numerator)) n := []byte(strconv.Itoa(numerator))
d := []byte(strconv.Itoa(denominator)) d := []byte(strconv.Itoa(denominator))
result := append([]byte(job.Handle), 0) result := append([]byte(job.Handle), 0)
result = append(result, n...) result = append(result, n...)
result = append(result, d...) result = append(result, d...)
return job.agent.WriteJob(NewWorkerJob(REQ, WORK_STATUS, result)) return job.agent.WriteJob(NewWorkerJob(REQ, WORK_STATUS, result))
} }