Browse Source

Job server can echo sth. to worker now.

tags/0.0.1
mikespook 13 years ago
parent
commit
dcd42b5c91
4 changed files with 119 additions and 18 deletions
  1. +2
    -0
      src/pkg/gearman/Makefile
  2. +3
    -0
      src/pkg/gearman/gearman.go
  3. +33
    -1
      src/pkg/gearman/gearman_test.go
  4. +81
    -17
      src/pkg/gearman/worker.go

+ 2
- 0
src/pkg/gearman/Makefile View File

@@ -7,6 +7,8 @@ include $(GOROOT)/src/Make.inc
TARG=gearman
GOFILES=\
gearman.go\
job.go\
jobclient.go\
worker.go\
# client.go\



+ 3
- 0
src/pkg/gearman/gearman.go View File

@@ -3,4 +3,7 @@ package gearman
const (
TCP = "tcp4"
WORKER_SERVER_CAP = 32
WORKER_FUNCTION_CAP = 512
)



+ 33
- 1
src/pkg/gearman/gearman_test.go View File

@@ -4,8 +4,13 @@ import (
"testing"
)

var worker *Worker

func init() {
worker = NewWorker()
}

func TestAddServer(t *testing.T) {
worker := NewWorker()
t.Log("Add local server 127.0.0.1:4730.")
if err := worker.AddServer("127.0.0.1:4730"); err != nil {
t.Error(err)
@@ -16,3 +21,30 @@ func TestAddServer(t *testing.T) {
t.Error("The length of server list should be 1.")
}
}

func TestAddFunction(t *testing.T) {
f := func(job *Job) []byte {
return nil
}

if err := worker.AddFunction("foobar", f); err != nil {
t.Error(err)
}
if l := len(worker.functions); l != 1 {
t.Log(worker.functions)
t.Error("The length of function map should be 1.")
}
}

func TestEcho(t * testing.T) {
go worker.Work()
if err := worker.Echo([]byte("Hello World")); err != nil {
t.Error(err)
}
}

func TestClose(t *testing.T) {
if err := worker.Close(); err != nil {
t.Error(err)
}
}

+ 81
- 17
src/pkg/gearman/worker.go View File

@@ -1,54 +1,118 @@
package gearman

import(
"net"
"os"
"sync"
"log"
)

type JobFunction func(job *Job) []byte

type Worker struct {
servers []*JobClient
functions map[string]JobFunction

servers []net.Conn
running bool
incoming chan *Job
mutex sync.Mutex
queue chan *Job
}

func NewWorker() (worker *Worker) {
worker = &Worker{servers:make([]net.Conn, 0, WORKER_SERVER_CAP)}
worker = &Worker{servers:make([]*JobClient, 0, WORKER_SERVER_CAP),
functions: make(map[string]JobFunction),
incoming: make(chan *Job, 512),
queue: make(chan *Job, 512),
running: true,}
return worker
}


// add server
// worker.AddServer("127.0.0.1:4730")
func (worker * Worker) AddServer(addr string) (err os.Error) {
worker.mutex.Lock()
defer worker.mutex.Unlock()

if len(worker.servers) == cap(worker.servers) {
return os.NewError("There were too many servers.")
}
conn, err := net.Dial(TCP, addr)

// Create a new job server's client as a agent of server
server, err := NewJobClient(addr, worker.incoming)
if err != nil {
return err
}

n := len(worker.servers)
worker.servers = worker.servers[0: n + 1]
worker.servers[n] = conn
return nil
worker.servers[n] = server
return
}
/*

// add function
func (worker * Worker) AddFunction(funcname string,
f interface{}, context interface{}) (err Error) {
func (worker * Worker) AddFunction(funcname string,
f JobFunction) (err os.Error) {
worker.mutex.Lock()
defer worker.mutex.Unlock()

if f == nil {
return os.NewError("Job function should not be nil.")
}
worker.functions[funcname] = f
return
}

// work
func (worker * GearmanWorker) Work() {
for {
func (worker * Worker) Work() {
for _, v := range worker.servers {
go v.Work()
}
for worker.running {
select {
case job := <-worker.incoming:
if err := worker.Exec(job); err != nil {
log.Panicln(err)
}
worker.queue <- job
}
}
}

func (worker * Worker) Result() (job *Job) {
if l := len(worker.queue); l != 1 {
if l == 0 {
return
}
for i := 0; i < l - 1; i ++ {
<-worker.queue
}
}
return <-worker.queue
}

// Close
// should used as defer
func (worker * GearmanWorker) Close() (err Error){
func (worker * Worker) Close() (err os.Error){
worker.running = false
for _, v := range worker.servers {
err = v.Close()
}
close(worker.incoming)
return err
}

// Echo
func (worker * Worker) Echo(data []byte) (err os.Error) {
e := make(chan os.Error)
for _, v := range worker.servers {
go func() {
e <- v.Echo(data)
}()
}
return <- e
}

// Exec
func (worker * Worker) Exec(job *Job) (err os.Error) {
return
}
*/

Loading…
Cancel
Save