From efb918b7dbed995c470c68db52776052e640d6ed Mon Sep 17 00:00:00 2001 From: Rob Pike Date: Mon, 13 Jul 2009 12:58:14 -0700 Subject: [PATCH] the beginnings of an rpc service. server side only; no client help yet (but it's easy). no http yet. service is synchronous. all this will improve. R=rsc DELTA=403 (403 added, 0 deleted, 0 changed) OCL=31522 CL=31536 --- src/pkg/rpc/Makefile | 62 ++++++++++++ src/pkg/rpc/client.go | 14 +++ src/pkg/rpc/server.go | 200 +++++++++++++++++++++++++++++++++++++ src/pkg/rpc/server_test.go | 134 +++++++++++++++++++++++++ 4 files changed, 410 insertions(+) create mode 100644 src/pkg/rpc/Makefile create mode 100644 src/pkg/rpc/client.go create mode 100644 src/pkg/rpc/server.go create mode 100644 src/pkg/rpc/server_test.go diff --git a/src/pkg/rpc/Makefile b/src/pkg/rpc/Makefile new file mode 100644 index 0000000000..4d8c15629e --- /dev/null +++ b/src/pkg/rpc/Makefile @@ -0,0 +1,62 @@ +# Copyright 2009 The Go Authors. All rights reserved. +# Use of this source code is governed by a BSD-style +# license that can be found in the LICENSE file. + + +# DO NOT EDIT. Automatically generated by gobuild. +# gobuild -m >Makefile + +D= + +include $(GOROOT)/src/Make.$(GOARCH) +AR=gopack + +default: packages + +clean: + rm -rf *.[$(OS)] *.a [$(OS)].out _obj + +test: packages + gotest + +coverage: packages + gotest + 6cov -g $$(pwd) | grep -v '_test\.go:' + +%.$O: %.go + $(GC) -I_obj $*.go + +%.$O: %.c + $(CC) $*.c + +%.$O: %.s + $(AS) $*.s + +O1=\ + client.$O\ + server.$O\ + + +phases: a1 +_obj$D/rpc.a: phases + +a1: $(O1) + $(AR) grc _obj$D/rpc.a client.$O server.$O + rm -f $(O1) + + +newpkg: clean + mkdir -p _obj$D + $(AR) grc _obj$D/rpc.a + +$(O1): newpkg +$(O2): a1 + +nuke: clean + rm -f $(GOROOT)/pkg/$(GOOS)_$(GOARCH)$D/rpc.a + +packages: _obj$D/rpc.a + +install: packages + test -d $(GOROOT)/pkg && mkdir -p $(GOROOT)/pkg/$(GOOS)_$(GOARCH)$D + cp _obj$D/rpc.a $(GOROOT)/pkg/$(GOOS)_$(GOARCH)$D/rpc.a diff --git a/src/pkg/rpc/client.go b/src/pkg/rpc/client.go new file mode 100644 index 0000000000..7df3dc8d51 --- /dev/null +++ b/src/pkg/rpc/client.go @@ -0,0 +1,14 @@ +// Copyright 2009 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package rpc + +import ( + "gob"; + "io"; + "os"; + "reflect"; + "sync"; +) + diff --git a/src/pkg/rpc/server.go b/src/pkg/rpc/server.go new file mode 100644 index 0000000000..b03685298d --- /dev/null +++ b/src/pkg/rpc/server.go @@ -0,0 +1,200 @@ +// Copyright 2009 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package rpc + +import ( + "gob"; + "log"; + "io"; + "os"; + "reflect"; + "strings"; + "sync"; + "unicode"; + "utf8"; +) + +// Precompute the reflect type for os.Error. Can't use os.Error directly +// because Typeof takes an empty interface value. This is annoying. +var unusedError *os.Error; +var typeOfOsError = reflect.Typeof(unusedError).(*reflect.PtrType).Elem() + +type methodType struct { + method reflect.Method; + argType *reflect.PtrType; + replyType *reflect.PtrType; +} + +type service struct { + name string; // name of service + rcvr reflect.Value; // receiver of methods for the service + typ reflect.Type; // type of the receiver + method map[string] *methodType; // registered methods +} + +// Request is a header written before every RPC call. +type Request struct { + ServiceMethod string; + Seq uint64; +} + +// Response is a header written before every RPC return. +type Response struct { + ServiceMethod string; + Seq uint64; + Error string; +} + +// Server represents the set of services available to an RPC client. +// The zero type for Server is ready to have services added. +type Server struct { + serviceMap map[string] *service; +} + +// Is this a publicly vislble - upper case - name? +func isPublic(name string) bool { + rune, wid_ := utf8.DecodeRuneInString(name); + return unicode.IsUpper(rune) +} + +// Add publishes in the server the set of methods of the +// recevier value that satisfy the following conditions: +// - public method +// - two arguments, both pointers to structs +// - one return value of type os.Error +// It returns an error if the receiver is not suitable. +func (server *Server) Add(rcvr interface{}) os.Error { + if server.serviceMap == nil { + server.serviceMap = make(map[string] *service); + } + s := new(service); + s.typ = reflect.Typeof(rcvr); + s.rcvr = reflect.NewValue(rcvr); + path_, sname := reflect.Indirect(s.rcvr).Type().Name(); + if sname == "" { + log.Exit("rpc: no service name for type", s.typ.String()) + } + if !isPublic(sname) { + s := "rpc server.Add: type " + sname + " is not public"; + log.Stderr(s); + return os.ErrorString(s); + } + s.name = sname; + s.method = make(map[string] *methodType); + + // Install the methods + for m := 0; m < s.typ.NumMethod(); m++ { + method := s.typ.Method(m); + mtype := method.Type; + mname := method.Name; + if !isPublic(mname) { + continue + } + // Method needs three ins: receiver, *args, *reply. + // The args and reply must be structs until gobs are more general. + if mtype.NumIn() != 3 { + log.Stderr("method", mname, "has wrong number of ins:", mtype.NumIn()); + continue; + } + argType, ok := mtype.In(1).(*reflect.PtrType); + if !ok { + log.Stderr(mname, "arg type not a pointer:", argType.String()); + continue; + } + if _, ok := argType.Elem().(*reflect.StructType); !ok { + log.Stderr(mname, "arg type not a pointer to a struct:", argType.String()); + continue; + } + replyType, ok := mtype.In(2).(*reflect.PtrType); + if !ok { + log.Stderr(mname, "reply type not a pointer:", replyType.String()); + continue; + } + if _, ok := replyType.Elem().(*reflect.StructType); !ok { + log.Stderr(mname, "reply type not a pointer to a struct:", replyType.String()); + continue; + } + // Method needs one out: os.Error. + if mtype.NumOut() != 1 { + log.Stderr("method", mname, "has wrong number of outs:", mtype.NumOut()); + continue; + } + if returnType := mtype.Out(0); returnType != typeOfOsError { + log.Stderr("method", mname, "returns", returnType.String(), "not os.Error"); + continue; + } + s.method[mname] = &methodType{method, argType, replyType}; + } + + if len(s.method) == 0 { + s := "rpc server.Add: type " + sname + " has no public methods of suitable type"; + log.Stderr(s); + return os.ErrorString(s); + } + server.serviceMap[s.name] = s; + return nil; +} + +func _new(t *reflect.PtrType) *reflect.PtrValue { + v := reflect.MakeZero(t).(*reflect.PtrValue); + v.PointTo(reflect.MakeZero(t.Elem())); + return v; +} + +// Blocks until the decoder is ready for the next message. +// TODO(r): blocks longer than that. make this async. +func (s *service) call(req *Request, mt *methodType, dec *gob.Decoder, enc *gob.Encoder) { + method := mt.method; + // Decode the argument value. + argv := _new(mt.argType); + dec.Decode(argv.Interface()); + // Invoke the method, providing a new value for the reply. + replyv := _new(mt.replyType); + returnValues := method.Func.Call([]reflect.Value{s.rcvr, argv, replyv}); + // The return value for the method is an os.Error. + err := returnValues[0].Interface(); + resp := new(Response); + if err != nil { + resp.Error = err.(os.Error).String(); + } + // Encode the response header + resp.ServiceMethod = req.ServiceMethod; + resp.Seq = req.Seq; + enc.Encode(resp); + // Encode the reply value. + enc.Encode(replyv.Interface()); +} + +func (server *Server) serve(conn io.ReadWriteCloser) { + dec := gob.NewDecoder(conn); + enc := gob.NewEncoder(conn); + for { + // Grab the request header. + req := new(Request); + err := dec.Decode(req); + if err != nil { + panicln("can't handle decode error yet", err); + } + serviceMethod := strings.Split(req.ServiceMethod, ".", 0); + if len(serviceMethod) != 2 { + panicln("service/Method request ill-formed:", req.ServiceMethod); + } + // Look up the request. + service, ok := server.serviceMap[serviceMethod[0]]; + if !ok { + panicln("can't find service", serviceMethod[0]); + } + method, ok := service.method[serviceMethod[1]]; + if !ok { + panicln("can't find method", serviceMethod[1]); + } + service.call(req, method, dec, enc); + } +} + +// Serve runs the server on the connection. +func (server *Server) Serve(conn io.ReadWriteCloser) { + go server.serve(conn) +} diff --git a/src/pkg/rpc/server_test.go b/src/pkg/rpc/server_test.go new file mode 100644 index 0000000000..01d991f148 --- /dev/null +++ b/src/pkg/rpc/server_test.go @@ -0,0 +1,134 @@ +// Copyright 2009 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package rpc + +import ( + "fmt"; + "gob"; + "http"; + "io"; + "log"; + "net"; + "os"; + "rpc"; + "testing"; +) + +var serverAddr string + +const second = 1e9 + + +type Args struct { + A, B int +} + +type Reply struct { + C int +} + +type Arith int + +func (t *Arith) Add(args *Args, reply *Reply) os.Error { + reply.C = args.A + args.B; + return nil +} + +func (t *Arith) Mul(args *Args, reply *Reply) os.Error { + reply.C = args.A * args.B; + return nil +} + +func (t *Arith) Div(args *Args, reply *Reply) os.Error { + if args.B == 0 { + return os.ErrorString("divide by zero"); + } + reply.C = args.A / args.B; + return nil +} + +func (t *Arith) Error(args *Args, reply *Reply) os.Error { + panicln("ERROR"); +} + +func run(server *Server, l net.Listener) { + conn, addr, err := l.Accept(); + if err != nil { + println("accept:", err.String()); + os.Exit(1); + } + server.Serve(conn); +} + +func startServer() { + server := new(Server); + server.Add(new(Arith)); + l, e := net.Listen("tcp", ":0"); // any available address + if e != nil { + log.Stderrf("net.Listen tcp :0: %v", e); + os.Exit(1); + } + serverAddr = l.Addr(); + log.Stderr("Test RPC server listening on ", serverAddr); +// go http.Serve(l, nil); + go run(server, l); +} + +func TestRPC(t *testing.T) { + var i int; + + startServer(); + + conn, err := net.Dial("tcp", "", serverAddr); + if err != nil { + t.Fatal("dialing:", err) + } + + enc := gob.NewEncoder(conn); + dec := gob.NewDecoder(conn); + req := new(rpc.Request); + req.ServiceMethod = "Arith.Add"; + req.Seq = 1; + enc.Encode(req); + args := &Args{7,8}; + enc.Encode(args); + response := new(rpc.Response); + dec.Decode(response); + reply := new(Reply); + dec.Decode(reply); + fmt.Printf("%d\n", reply.C); + if reply.C != args.A + args.B { + t.Errorf("Add: expected %d got %d", reply.C != args.A + args.B); + } + + req.ServiceMethod = "Arith.Mul"; + req.Seq++; + enc.Encode(req); + args = &Args{7,8}; + enc.Encode(args); + response = new(rpc.Response); + dec.Decode(response); + reply = new(Reply); + dec.Decode(reply); + fmt.Printf("%d\n", reply.C); + if reply.C != args.A * args.B { + t.Errorf("Mul: expected %d got %d", reply.C != args.A * args.B); + } + + req.ServiceMethod = "Arith.Div"; + req.Seq++; + enc.Encode(req); + args = &Args{7,0}; + enc.Encode(args); + response = new(rpc.Response); + dec.Decode(response); + reply = new(Reply); + dec.Decode(reply); + // expect an error: zero divide + if response.Error == "" { + t.Errorf("Div: expected error"); + } +} + -- 2.48.1