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

@@ -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

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";
+)
+

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)
+}

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");
+	}
+}
+