2011-05-23 12:15:48 +08:00
|
|
|
// Copyright 2011 Xing Xing <mikespook@gmail.com> All rights reserved.
|
|
|
|
// Use of this source code is governed by a MIT
|
|
|
|
// license that can be found in the LICENSE file.
|
|
|
|
|
2011-03-15 20:21:42 +08:00
|
|
|
package gearman
|
|
|
|
|
2011-05-24 13:20:22 +08:00
|
|
|
import (
|
2011-05-16 17:26:32 +08:00
|
|
|
"os"
|
2011-05-17 18:12:40 +08:00
|
|
|
"sync"
|
2011-05-24 13:20:22 +08:00
|
|
|
// "log"
|
2011-03-15 20:21:42 +08:00
|
|
|
)
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// The definition of the callback function.
|
2011-05-19 20:10:53 +08:00
|
|
|
type JobFunction func(job *WorkerJob) ([]byte, os.Error)
|
2011-05-23 12:15:48 +08:00
|
|
|
// Map for added function.
|
2011-05-18 19:40:03 +08:00
|
|
|
type JobFunctionMap map[string]JobFunction
|
2011-05-17 18:12:40 +08:00
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
/*
|
|
|
|
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
|
|
|
|
}
|
2011-05-24 13:20:22 +08:00
|
|
|
*/
|
2011-05-16 17:26:32 +08:00
|
|
|
type Worker struct {
|
2011-05-24 13:20:22 +08:00
|
|
|
clients []*jobClient
|
2011-05-18 19:40:03 +08:00
|
|
|
functions JobFunctionMap
|
2011-05-16 17:26:32 +08:00
|
|
|
|
2011-05-24 13:20:22 +08:00
|
|
|
running bool
|
2011-05-19 20:10:53 +08:00
|
|
|
incoming chan *WorkerJob
|
2011-05-24 13:20:22 +08:00
|
|
|
mutex sync.Mutex
|
2011-05-19 20:10:53 +08:00
|
|
|
JobQueue chan *WorkerJob
|
|
|
|
ErrQueue chan os.Error
|
2011-03-15 20:21:42 +08:00
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Get a new worker
|
2011-05-16 17:26:32 +08:00
|
|
|
func NewWorker() (worker *Worker) {
|
2011-05-18 19:40:03 +08:00
|
|
|
worker = &Worker{
|
|
|
|
// job server list
|
2011-05-24 13:20:22 +08:00
|
|
|
clients: make([]*jobClient, 0, WORKER_SERVER_CAP),
|
2011-05-18 19:40:03 +08:00
|
|
|
// function list
|
|
|
|
functions: make(JobFunctionMap),
|
2011-05-24 13:20:22 +08:00
|
|
|
incoming: make(chan *WorkerJob, QUEUE_CAP),
|
|
|
|
JobQueue: make(chan *WorkerJob, QUEUE_CAP),
|
|
|
|
ErrQueue: make(chan os.Error, QUEUE_CAP),
|
|
|
|
running: true,
|
2011-05-18 19:40:03 +08:00
|
|
|
}
|
2011-05-19 20:10:53 +08:00
|
|
|
return
|
2011-03-15 20:21:42 +08:00
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Add a server. The addr should be 'host:port' format.
|
|
|
|
// The connection is established at this time.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) AddServer(addr string) (err os.Error) {
|
2011-05-17 18:12:40 +08:00
|
|
|
worker.mutex.Lock()
|
|
|
|
defer worker.mutex.Unlock()
|
|
|
|
|
2011-05-18 19:40:03 +08:00
|
|
|
if len(worker.clients) == cap(worker.clients) {
|
|
|
|
return os.NewError("There were too many clients.")
|
2011-03-15 20:21:42 +08:00
|
|
|
}
|
2011-05-17 18:12:40 +08:00
|
|
|
|
|
|
|
// Create a new job server's client as a agent of server
|
2011-05-19 20:10:53 +08:00
|
|
|
server, err := newJobClient(addr, worker)
|
2011-05-16 17:26:32 +08:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2011-05-17 18:12:40 +08:00
|
|
|
|
2011-05-18 19:40:03 +08:00
|
|
|
n := len(worker.clients)
|
2011-05-24 13:20:22 +08:00
|
|
|
worker.clients = worker.clients[0 : n+1]
|
2011-05-18 19:40:03 +08:00
|
|
|
worker.clients[n] = server
|
2011-05-17 18:12:40 +08:00
|
|
|
return
|
2011-03-15 20:21:42 +08:00
|
|
|
}
|
2011-05-17 18:12:40 +08:00
|
|
|
|
2011-03-15 20:21:42 +08:00
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Add a function.
|
|
|
|
// Plz added job servers first, then functions.
|
|
|
|
// The API will tell every connected job server that 'I can do this'
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) AddFunction(funcname string,
|
|
|
|
f JobFunction, timeout uint32) (err os.Error) {
|
2011-05-18 19:40:03 +08:00
|
|
|
if len(worker.clients) < 1 {
|
|
|
|
return os.NewError("Did not connect to Job Server.")
|
|
|
|
}
|
|
|
|
worker.mutex.Lock()
|
|
|
|
defer worker.mutex.Unlock()
|
2011-05-17 18:12:40 +08:00
|
|
|
worker.functions[funcname] = f
|
2011-05-18 19:40:03 +08:00
|
|
|
|
|
|
|
var datatype uint32
|
|
|
|
var data []byte
|
|
|
|
if timeout == 0 {
|
|
|
|
datatype = CAN_DO
|
|
|
|
data = []byte(funcname)
|
|
|
|
} else {
|
|
|
|
datatype = CAN_DO_TIMEOUT
|
|
|
|
data = []byte(funcname + "\x00")
|
|
|
|
t := uint32ToByte(timeout)
|
2011-05-24 13:20:22 +08:00
|
|
|
data = append(data, t[:]...)
|
2011-05-18 19:40:03 +08:00
|
|
|
}
|
2011-05-19 20:10:53 +08:00
|
|
|
job := NewWorkerJob(REQ, datatype, data)
|
2011-05-18 19:40:03 +08:00
|
|
|
worker.WriteJob(job)
|
2011-05-17 18:12:40 +08:00
|
|
|
return
|
2011-03-15 20:21:42 +08:00
|
|
|
}
|
2011-05-18 19:40:03 +08:00
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Remove a function.
|
|
|
|
// Tell job servers 'I can not do this now' at the same time.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) RemoveFunction(funcname string) (err os.Error) {
|
2011-05-18 19:40:03 +08:00
|
|
|
worker.mutex.Lock()
|
|
|
|
defer worker.mutex.Unlock()
|
|
|
|
|
|
|
|
if worker.functions[funcname] == nil {
|
|
|
|
return os.NewError("No function named: " + funcname)
|
2011-05-17 20:32:36 +08:00
|
|
|
}
|
2011-05-18 19:40:03 +08:00
|
|
|
worker.functions[funcname] = nil, false
|
2011-05-19 20:10:53 +08:00
|
|
|
job := NewWorkerJob(REQ, CANT_DO, []byte(funcname))
|
2011-05-18 19:40:03 +08:00
|
|
|
worker.WriteJob(job)
|
|
|
|
return
|
|
|
|
}
|
2011-05-17 20:32:36 +08:00
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Main loop
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) Work() {
|
2011-05-18 19:40:03 +08:00
|
|
|
for _, v := range worker.clients {
|
2011-05-17 18:12:40 +08:00
|
|
|
go v.Work()
|
|
|
|
}
|
|
|
|
for worker.running {
|
|
|
|
select {
|
2011-05-24 13:20:22 +08:00
|
|
|
case job := <-worker.incoming:
|
|
|
|
if job == nil {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
switch job.DataType {
|
|
|
|
case NO_JOB:
|
|
|
|
// do nothing
|
|
|
|
case ERROR:
|
|
|
|
_, err := getError(job.Data)
|
|
|
|
worker.ErrQueue <- err
|
|
|
|
case JOB_ASSIGN, JOB_ASSIGN_UNIQ:
|
|
|
|
go func() {
|
|
|
|
if err := worker.exec(job); err != nil {
|
2011-05-19 20:10:53 +08:00
|
|
|
worker.ErrQueue <- err
|
2011-05-24 13:20:22 +08:00
|
|
|
}
|
|
|
|
}()
|
|
|
|
default:
|
|
|
|
worker.JobQueue <- job
|
|
|
|
}
|
2011-05-17 18:12:40 +08:00
|
|
|
}
|
2011-03-15 20:21:42 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// 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.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) LastJob() (job *WorkerJob) {
|
2011-05-19 20:10:53 +08:00
|
|
|
if l := len(worker.JobQueue); l != 1 {
|
2011-05-17 18:12:40 +08:00
|
|
|
if l == 0 {
|
|
|
|
return
|
|
|
|
}
|
2011-05-24 13:20:22 +08:00
|
|
|
for i := 0; i < l-1; i++ {
|
2011-05-19 20:10:53 +08:00
|
|
|
<-worker.JobQueue
|
2011-05-17 18:12:40 +08:00
|
|
|
}
|
|
|
|
}
|
2011-05-19 20:10:53 +08:00
|
|
|
return <-worker.JobQueue
|
2011-05-17 18:12:40 +08:00
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Close.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) Close() (err os.Error) {
|
2011-05-17 18:12:40 +08:00
|
|
|
worker.running = false
|
2011-05-18 19:40:03 +08:00
|
|
|
for _, v := range worker.clients {
|
2011-05-17 18:12:40 +08:00
|
|
|
err = v.Close()
|
|
|
|
}
|
|
|
|
close(worker.incoming)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// 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.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) WriteJob(job *WorkerJob) (err os.Error) {
|
2011-05-17 18:12:40 +08:00
|
|
|
e := make(chan os.Error)
|
2011-05-18 19:40:03 +08:00
|
|
|
for _, v := range worker.clients {
|
2011-05-17 18:12:40 +08:00
|
|
|
go func() {
|
2011-05-17 20:32:36 +08:00
|
|
|
e <- v.WriteJob(job)
|
2011-05-17 18:12:40 +08:00
|
|
|
}()
|
|
|
|
}
|
2011-05-24 13:20:22 +08:00
|
|
|
return <-e
|
2011-05-17 18:12:40 +08:00
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Send a something out, get the samething back.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) Echo(data []byte) (err os.Error) {
|
2011-05-19 20:10:53 +08:00
|
|
|
job := NewWorkerJob(REQ, ECHO_REQ, data)
|
2011-05-18 19:40:03 +08:00
|
|
|
return worker.WriteJob(job)
|
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Remove all of functions.
|
|
|
|
// Both from the worker or job servers.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) Reset() (err os.Error) {
|
2011-05-19 20:10:53 +08:00
|
|
|
job := NewWorkerJob(REQ, RESET_ABILITIES, nil)
|
2011-05-22 22:43:14 +08:00
|
|
|
err = worker.WriteJob(job)
|
|
|
|
worker.functions = make(JobFunctionMap)
|
|
|
|
return
|
2011-05-18 19:40:03 +08:00
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Set the worker's unique id.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) SetId(id string) (err os.Error) {
|
2011-05-19 20:10:53 +08:00
|
|
|
job := NewWorkerJob(REQ, SET_CLIENT_ID, []byte(id))
|
2011-05-18 19:40:03 +08:00
|
|
|
return worker.WriteJob(job)
|
2011-05-17 20:32:36 +08:00
|
|
|
}
|
|
|
|
|
2011-05-23 12:15:48 +08:00
|
|
|
// Execute the job. And send back the result.
|
2011-05-24 13:20:22 +08:00
|
|
|
func (worker *Worker) exec(job *WorkerJob) (err os.Error) {
|
2011-05-18 19:40:03 +08:00
|
|
|
jobdata := splitByteArray(job.Data, '\x00')
|
|
|
|
job.Handle = string(jobdata[0])
|
|
|
|
funcname := string(jobdata[1])
|
2011-05-21 12:32:44 +08:00
|
|
|
if job.DataType == JOB_ASSIGN {
|
2011-05-18 19:40:03 +08:00
|
|
|
job.Data = jobdata[2]
|
|
|
|
} else {
|
|
|
|
job.UniqueId = string(jobdata[2])
|
|
|
|
job.Data = jobdata[3]
|
|
|
|
}
|
2011-05-19 14:39:31 +08:00
|
|
|
f := worker.functions[funcname]
|
|
|
|
if f == nil {
|
|
|
|
return os.NewError("function is nil")
|
|
|
|
}
|
|
|
|
result, err := f(job)
|
2011-05-18 19:40:03 +08:00
|
|
|
var datatype uint32
|
|
|
|
if err == nil {
|
|
|
|
datatype = WORK_COMPLETE
|
2011-05-24 13:20:22 +08:00
|
|
|
} else {
|
2011-05-18 19:40:03 +08:00
|
|
|
if result == nil {
|
|
|
|
datatype = WORK_FAIL
|
|
|
|
} else {
|
|
|
|
datatype = WORK_EXCEPTION
|
|
|
|
}
|
|
|
|
}
|
2011-05-19 20:10:53 +08:00
|
|
|
|
2011-05-19 14:39:31 +08:00
|
|
|
job.magicCode = REQ
|
2011-05-21 12:32:44 +08:00
|
|
|
job.DataType = datatype
|
2011-05-19 14:39:31 +08:00
|
|
|
job.Data = result
|
2011-05-18 19:40:03 +08:00
|
|
|
|
2011-05-19 20:10:53 +08:00
|
|
|
worker.WriteJob(job)
|
2011-05-17 18:12:40 +08:00
|
|
|
return
|
2011-03-15 20:21:42 +08:00
|
|
|
}
|