diff --git a/README b/README index fdea0a1..4795023 100644 --- a/README +++ b/README @@ -19,5 +19,4 @@ It was implemented a native protocol for both worker and client API. ---- xingxing http://mikespook.com - - +http://twitter.com/mikespook diff --git a/src/pkg/gearman/client.go b/src/pkg/gearman/client.go index a66cf2f..d34f5da 100644 --- a/src/pkg/gearman/client.go +++ b/src/pkg/gearman/client.go @@ -1,3 +1,7 @@ +// Copyright 2011 Xing Xing All rights reserved. +// Use of this source code is governed by a MIT +// license that can be found in the LICENSE file. + package gearman import ( @@ -8,6 +12,15 @@ import ( "strconv" ) +/* +The client side api for gearman. + +usage: + client = NewClient() + client.AddServer("127.0.0.1:4730") + handle := client.Do("foobar", []byte("data here"), JOB_LOW | JOB_BG) + +*/ type Client struct { mutex sync.Mutex conn net.Conn @@ -16,6 +29,7 @@ type Client struct { UId uint32 } +// Create a new client. func NewClient() (client * Client){ client = &Client{JobQueue:make(chan *ClientJob, QUEUE_CAP), incoming:make(chan []byte, QUEUE_CAP), @@ -23,6 +37,9 @@ func NewClient() (client * Client){ return } +// Add a server. +// In this version, one client connect to one job server. +// Sample is better. Plz do the load balancing by your self. func (client *Client) AddServer(addr string) (err os.Error) { conn, err := net.Dial(TCP, addr) if err != nil { @@ -32,10 +49,13 @@ func (client *Client) AddServer(addr string) (err os.Error) { return } +// Internal read func (client *Client) read() (data []byte, err os.Error) { if len(client.incoming) > 0 { + // incoming queue is not empty data = <-client.incoming } else { + // empty queue, read data from socket for { buf := make([]byte, BUFFER_SIZE) var n int @@ -51,6 +71,7 @@ func (client *Client) read() (data []byte, err os.Error) { } } } + // split package start, end := 0, 4 for i := 0; i < len(data); i ++{ if string(data[start:end]) == RES_STR { @@ -72,6 +93,8 @@ func (client *Client) read() (data []byte, err os.Error) { return } +// Read a job from job server. +// This function will return the job, and add it to the job queue. func (client *Client) ReadJob() (job *ClientJob, err os.Error) { var rel []byte if rel, err = client.read(); err != nil { @@ -91,6 +114,12 @@ func (client *Client) ReadJob() (job *ClientJob, err os.Error) { return } +// Do the function. +// funcname is a string with function name. +// data is encoding to byte array. +// flag set the job type, include running level: JOB_LOW, JOB_NORMAL, JOB_HIGH, +// and if it is background job: JOB_BG. +// 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) { var datatype uint32 if flag & JOB_LOW == JOB_LOW { @@ -147,6 +176,7 @@ func (client *Client) Do(funcname string, data []byte, flag byte) (handle string return } +// Internal read last job func (client *Client) readLastJob(datatype uint32) (job *ClientJob, err os.Error){ for { if job, err = client.ReadJob(); err != nil { @@ -162,6 +192,8 @@ func (client *Client) readLastJob(datatype uint32) (job *ClientJob, err os.Error return } +// Get job status from job server. +// !!!Not fully tested.!!! func (client *Client) Status(handle string) (known, running bool, numerator, denominator uint, err os.Error) { if err = client.WriteJob(NewClientJob(REQ, GET_STATUS, []byte(handle))); err != nil { @@ -191,6 +223,7 @@ func (client *Client) Status(handle string) (known, running bool, numerator, den return } +// Send a something out, get the samething back. func (client *Client) Echo(data []byte) (echo []byte, err os.Error) { if err = client.WriteJob(NewClientJob(REQ, ECHO_REQ, data)); err != nil { return @@ -203,7 +236,10 @@ func (client *Client) Echo(data []byte) (echo []byte, err os.Error) { return } -func (client *Client) LastResult() (job *ClientJob) { +// Get the last job. +// the job means a network package. +// Normally, it is the job executed result. +func (client *Client) LastJob() (job *ClientJob) { if l := len(client.JobQueue); l != 1 { if l == 0 { return @@ -215,10 +251,12 @@ func (client *Client) LastResult() (job *ClientJob) { return <-client.JobQueue } +// Send the job to job server. func (client *Client) WriteJob(job *ClientJob) (err os.Error) { return client.write(job.Encode()) } +// Internal write func (client *Client) write(buf []byte) (err os.Error) { var n int for i := 0; i < len(buf); i += n { @@ -230,6 +268,7 @@ func (client *Client) write(buf []byte) (err os.Error) { return } +// Close. func (client *Client) Close() (err os.Error) { err = client.conn.Close() close(client.JobQueue) diff --git a/src/pkg/gearman/client/job.go b/src/pkg/gearman/client/job.go index f35ce39..0f33293 100644 --- a/src/pkg/gearman/client/job.go +++ b/src/pkg/gearman/client/job.go @@ -1,3 +1,7 @@ +// Copyright 2011 Xing Xing All rights reserved. +// Use of this source code is governed by a MIT +// license that can be found in the LICENSE file. + package gearman import ( @@ -5,18 +9,21 @@ import ( // "log" ) +// Client side job type ClientJob struct { Data []byte Handle, UniqueId string magicCode, DataType uint32 } +// Create a new job func NewClientJob(magiccode, datatype uint32, data []byte) (job *ClientJob) { return &ClientJob{magicCode:magiccode, DataType:datatype, Data:data} } +// Decode a job from byte slice func DecodeClientJob(data []byte) (job * ClientJob, err os.Error) { if len(data) < 12 { err = os.NewError("Data length is too small.") @@ -33,6 +40,7 @@ func DecodeClientJob(data []byte) (job * ClientJob, err os.Error) { return } +// Encode a job to byte slice func (job *ClientJob) Encode() (data []byte) { magiccode := uint32ToByte(job.magicCode) datatype := uint32ToByte(job.DataType) @@ -46,6 +54,7 @@ func (job *ClientJob) Encode() (data []byte) { return } +// Extract the job's result. func (job * ClientJob) Result() (data []byte, err os.Error){ switch job.DataType { case WORK_FAIL: @@ -69,6 +78,7 @@ func (job * ClientJob) Result() (data []byte, err os.Error){ return } +// Extract the job's update func (job *ClientJob) Update() (data []byte, err os.Error) { if job.DataType != WORK_DATA && job.DataType != WORK_WARNING { err = os.NewError("The job is not a update.") diff --git a/src/pkg/gearman/gearman.go b/src/pkg/gearman/gearman.go index 8e8605e..d1da4ad 100644 --- a/src/pkg/gearman/gearman.go +++ b/src/pkg/gearman/gearman.go @@ -1,3 +1,12 @@ +// Copyright 2011 Xing Xing All rights reserved. +// Use of this source code is governed by a MIT +// license that can be found in the LICENSE file. + +/* +This module is Gearman API for golang. +The protocol was implemented by native way. +*/ + package gearman import ( @@ -5,20 +14,27 @@ import ( ) const ( + // tcp4 is tested. You can modify this to 'tcp' for both ipv4 and ipv6, + // or 'tcp6' only for ipv6. TCP = "tcp4" + // the number limited for job servers. WORKER_SERVER_CAP = 32 + // the number limited for functions. WORKER_FUNCTION_CAP = 512 + // queue size QUEUE_CAP = 512 + // read buffer size BUFFER_SIZE = 1024 - // \x00REQ + // \x00REQ REQ = 5391697 REQ_STR = "\x00REQ" // \x00RES RES = 5391699 RES_STR = "\x00RES" + // package data type CAN_DO = 1 CANT_DO = 2 RESET_ABILITIES = 3 @@ -51,16 +67,24 @@ const ( SUBMIT_JOB_LOW = 33 SUBMIT_JOB_LOW_BG = 34 + // Job type + // JOB_NORMAL | JOB_BG means a normal level job run in background + // normal level JOB_NORMAL = 0 + // background job JOB_BG = 1 + // low level JOB_LOW = 2 + // high level JOB_HIGH = 4 ) +// No use type Job interface { Encode() []byte } +// Splite the byte array by a byte func splitByteArray(slice []byte, spot byte) (data [][]byte){ data = make([][]byte, 0, 10) start, end := 0, 0 @@ -78,6 +102,7 @@ func splitByteArray(slice []byte, spot byte) (data [][]byte){ return } +// Extract the error message func getError(data []byte) (eno os.Errno, err os.Error) { rel := splitByteArray(data, '\x00') if len(rel) != 2 { @@ -90,6 +115,7 @@ func getError(data []byte) (eno os.Errno, err os.Error) { return } +// Decode [4]byte to uint32 func byteToUint32(buf [4]byte) uint32 { return uint32(buf[0]) << 24 + uint32(buf[1]) << 16 + @@ -97,6 +123,7 @@ func byteToUint32(buf [4]byte) uint32 { uint32(buf[3]) } +// Encode uint32 to [4]byte func uint32ToByte(i uint32) (data [4]byte) { data[0] = byte((i >> 24) & 0xff) data[1] = byte((i >> 16) & 0xff) diff --git a/src/pkg/gearman/worker.go b/src/pkg/gearman/worker.go index 4dd0eaa..2b90000 100644 --- a/src/pkg/gearman/worker.go +++ b/src/pkg/gearman/worker.go @@ -1,3 +1,7 @@ +// Copyright 2011 Xing Xing All rights reserved. +// Use of this source code is governed by a MIT +// license that can be found in the LICENSE file. + package gearman import( @@ -6,9 +10,29 @@ import( // "log" ) +// The definition of the callback function. type JobFunction func(job *WorkerJob) ([]byte, os.Error) +// Map for added function. type JobFunctionMap map[string]JobFunction +/* +Worker side api for gearman. + +usage: + worker = NewWorker() + worker.AddFunction("foobar", foobar) + worker.AddServer("127.0.0.1:4730") + worker.Work() // Enter the worker's main loop + +The definition of the callback function 'foobar' should suit for the type 'JobFunction'. +It looks like this: + +func foobar(job *WorkerJob) (data []byte, err os.Error) { + //sth. here + //plaplapla... + return +} +*/ type Worker struct { clients []*jobClient functions JobFunctionMap @@ -20,6 +44,7 @@ type Worker struct { ErrQueue chan os.Error } +// Get a new worker func NewWorker() (worker *Worker) { worker = &Worker{ // job server list @@ -34,8 +59,8 @@ func NewWorker() (worker *Worker) { return } -// add server -// worker.AddServer("127.0.0.1:4730") +// Add a server. The addr should be 'host:port' format. +// The connection is established at this time. func (worker * Worker) AddServer(addr string) (err os.Error) { worker.mutex.Lock() defer worker.mutex.Unlock() @@ -57,12 +82,11 @@ func (worker * Worker) AddServer(addr string) (err os.Error) { } -// add function +// Add a function. +// Plz added job servers first, then functions. +// The API will tell every connected job server that 'I can do this' func (worker * Worker) AddFunction(funcname string, f JobFunction, timeout uint32) (err os.Error) { - if f == nil { - return os.NewError("Job function should not be nil.") - } if len(worker.clients) < 1 { return os.NewError("Did not connect to Job Server.") } @@ -86,7 +110,8 @@ func (worker * Worker) AddFunction(funcname string, return } -// remove function +// Remove a function. +// Tell job servers 'I can not do this now' at the same time. func (worker * Worker) RemoveFunction(funcname string) (err os.Error) { worker.mutex.Lock() defer worker.mutex.Unlock() @@ -100,7 +125,7 @@ func (worker * Worker) RemoveFunction(funcname string) (err os.Error) { return } -// work +// Main loop func (worker * Worker) Work() { for _, v := range worker.clients { go v.Work() @@ -118,10 +143,11 @@ func (worker * Worker) Work() { _, err := getError(job.Data) worker.ErrQueue <- err case JOB_ASSIGN, JOB_ASSIGN_UNIQ: - if err := worker.exec(job); err != nil { - worker.ErrQueue <- err - continue - } + go func() { + if err := worker.exec(job); err != nil { + worker.ErrQueue <- err + } + }() default: worker.JobQueue <- job } @@ -129,7 +155,11 @@ func (worker * Worker) Work() { } } -func (worker * Worker) LastResult() (job *WorkerJob) { +// Get the last job in queue. +// If there are more than one job in the queue, +// the last one will be returned, +// the others will be lost. +func (worker * Worker) LastJob() (job *WorkerJob) { if l := len(worker.JobQueue); l != 1 { if l == 0 { return @@ -141,8 +171,7 @@ func (worker * Worker) LastResult() (job *WorkerJob) { return <-worker.JobQueue } -// Close -// should used as defer +// Close. func (worker * Worker) Close() (err os.Error){ worker.running = false for _, v := range worker.clients { @@ -152,6 +181,9 @@ func (worker * Worker) Close() (err os.Error){ return err } +// 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) WriteJob(job *WorkerJob) (err os.Error) { e := make(chan os.Error) for _, v := range worker.clients { @@ -162,13 +194,14 @@ func (worker * Worker) WriteJob(job *WorkerJob) (err os.Error) { return <- e } -// Echo +// Send a something out, get the samething back. func (worker * Worker) Echo(data []byte) (err os.Error) { job := NewWorkerJob(REQ, ECHO_REQ, data) return worker.WriteJob(job) } -// Reset +// Remove all of functions. +// Both from the worker or job servers. func (worker * Worker) Reset() (err os.Error){ job := NewWorkerJob(REQ, RESET_ABILITIES, nil) err = worker.WriteJob(job) @@ -176,13 +209,13 @@ func (worker * Worker) Reset() (err os.Error){ return } -// SetId +// Set the worker's unique id. func (worker * Worker) SetId(id string) (err os.Error) { job := NewWorkerJob(REQ, SET_CLIENT_ID, []byte(id)) return worker.WriteJob(job) } -// Exec +// Execute the job. And send back the result. func (worker * Worker) exec(job *WorkerJob) (err os.Error) { jobdata := splitByteArray(job.Data, '\x00') job.Handle = string(jobdata[0]) diff --git a/src/pkg/gearman/worker/job.go b/src/pkg/gearman/worker/job.go index ec6b456..8215463 100644 --- a/src/pkg/gearman/worker/job.go +++ b/src/pkg/gearman/worker/job.go @@ -1,10 +1,16 @@ +// Copyright 2011 Xing Xing All rights reserved. +// Use of this source code is governed by a MIT +// license that can be found in the LICENSE file. + package gearman import ( "os" + "strconv" // "log" ) +// Worker side job type WorkerJob struct { Data []byte Handle, UniqueId string @@ -13,12 +19,14 @@ type WorkerJob struct { Job } +// Create a new job func NewWorkerJob(magiccode, datatype uint32, data []byte) (job *WorkerJob) { return &WorkerJob{magicCode:magiccode, DataType: datatype, Data:data} } +// Decode job from byte slice func DecodeWorkerJob(data []byte) (job *WorkerJob, err os.Error) { if len(data) < 12 { err = os.NewError("Data length is too small.") @@ -35,6 +43,7 @@ func DecodeWorkerJob(data []byte) (job *WorkerJob, err os.Error) { return } +// Encode a job to byte slice func (job *WorkerJob) Encode() (data []byte) { magiccode := uint32ToByte(job.magicCode) datatype := uint32ToByte(job.DataType) @@ -54,7 +63,8 @@ func (job *WorkerJob) Encode() (data []byte) { return } -// update data +// Send some datas to client. +// Using this in a job's executing. func (job * WorkerJob) UpdateData(data []byte, iswaring bool) (err os.Error) { result := append([]byte(job.Handle), 0) result = append(result, data ...) @@ -67,13 +77,14 @@ func (job * WorkerJob) UpdateData(data []byte, iswaring bool) (err os.Error) { return job.client.WriteJob(NewWorkerJob(REQ, datatype, result)) } -// update status -func (job * WorkerJob) UpdateStatus(numerator, denominator uint32) (err os.Error) { - n := uint32ToByte(numerator) - d := uint32ToByte(denominator) +// Update status. +// Tall client how many percent job has been executed. +func (job * WorkerJob) UpdateStatus(numerator, denominator int) (err os.Error) { + n := []byte(strconv.Itoa(numerator)) + d := []byte(strconv.Itoa(denominator)) result := append([]byte(job.Handle), 0) - result = append(result, n[:] ...) - result = append(result, d[:] ...) + result = append(result, n ...) + result = append(result, d ...) return job.client.WriteJob(NewWorkerJob(REQ, WORK_STATUS, result)) } diff --git a/src/pkg/gearman/worker/jobclient.go b/src/pkg/gearman/worker/jobclient.go index ae49374..31ca16d 100644 --- a/src/pkg/gearman/worker/jobclient.go +++ b/src/pkg/gearman/worker/jobclient.go @@ -1,3 +1,7 @@ +// Copyright 2011 Xing Xing All rights reserved. +// Use of this source code is governed by a MIT +// license that can be found in the LICENSE file. + package gearman import ( @@ -6,12 +10,14 @@ import ( // "log" ) +// The client of job server. type jobClient struct { conn net.Conn worker *Worker running bool } +// Create the client of job server. func newJobClient(addr string, worker *Worker) (jobclient *jobClient, err os.Error) { conn, err := net.Dial(TCP, addr) if err != nil { @@ -21,6 +27,7 @@ func newJobClient(addr string, worker *Worker) (jobclient *jobClient, err os.Err return jobclient, err } +// Main loop. func (client *jobClient) Work() { noop := true OUT: for client.running { @@ -64,10 +71,12 @@ func (client *jobClient) Work() { return } +// Send a job to the job server. func (client *jobClient) WriteJob(job *WorkerJob) (err os.Error) { return client.Write(job.Encode()) } +// Write the encoded job. func (client *jobClient) Write(buf []byte) (err os.Error) { var n int for i := 0; i < len(buf); i += n { @@ -79,6 +88,7 @@ func (client *jobClient) Write(buf []byte) (err os.Error) { return } +// Close. func (client *jobClient) Close() (err os.Error) { client.running = false err = client.conn.Close()