code review 4889043: rpc: implement ServeRequest to synchronously serve a si...

src/pkg/rpc/server.go

 // 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 provides access to the exported methods of an object across a
         network or other I/O connection.  A server registers an object, making it visible
         as a service with the name of the type of the object.  After registration, exported
         methods of the object will be accessible remotely.  A server may register multiple
         objects (services) of different types but it is an error to register multiple
@@ skipping 376 matching lines @@
         buf := bufio.NewWriter(conn)
         srv := &gobServerCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(buf), buf}
         server.ServeCodec(srv)
 }
 
 // ServeCodec is like ServeConn but uses the specified codec to
 // decode requests and encode responses.
 func (server *Server) ServeCodec(codec ServerCodec) {
         sending := new(sync.Mutex)
         for {
-                req, service, mtype, err := server.readRequest(codec)
+                service, mtype, req, argv, replyv, err := server.readRequest(codec)
                 if err != nil {
                         if err != os.EOF {
                                 log.Println("rpc:", err)
                         }
                         if err == os.EOF || err == io.ErrUnexpectedEOF {
                                 break
                         }
-                        // discard body
-                        codec.ReadRequestBody(nil)
-
                         // send a response if we actually managed to read a header.
                         if req != nil {
                                 server.sendResponse(sending, req, invalidRequest, codec, err.String())
                                 server.freeRequest(req)
                         }
                         continue
                 }
-
-                // Decode the argument value.
-                var argv reflect.Value
-                argIsValue := false // if true, need to indirect before calling.
-                if mtype.ArgType.Kind() == reflect.Ptr {
-                        argv = reflect.New(mtype.ArgType.Elem())
-                } else {
-                        argv = reflect.New(mtype.ArgType)
-                        argIsValue = true
-                }
-                // argv guaranteed to be a pointer now.
-                replyv := reflect.New(mtype.ReplyType.Elem())
-                err = codec.ReadRequestBody(argv.Interface())
-                if err != nil {
-                        if err == os.EOF || err == io.ErrUnexpectedEOF {
-                                if err == io.ErrUnexpectedEOF {
-                                        log.Println("rpc:", err)
-                                }
-                                break
-                        }
-                        server.sendResponse(sending, req, replyv.Interface(), codec, err.String())
-                        continue
-                }
-                if argIsValue {
-                        argv = argv.Elem()
-                }
                 go service.call(server, sending, mtype, req, argv, replyv, codec)
         }
         codec.Close()
 }
 
+// ServeSingle is like ServeCodec but synchronously serves a single request

[rsc, 2011/08/15 16:11:09]

suggest ServeRequest instead of ServeSingle.

// ServeRequest serves one request from codec and then returns.
// It does not close the codec.

the part about goroutines is not strictly
relevant: what's relevant is when it returns.

[sougou, 2011/08/15 20:17:18]

Done.

+// without launching a goroutine. It does not close the codec upon completion.
+func (server *Server) ServeSingle(codec ServerCodec) os.Error {
+        sending := new(sync.Mutex)
+        service, mtype, req, argv, replyv, err := server.readRequest(codec)
+        if err != nil {
+                if err == os.EOF || err == io.ErrUnexpectedEOF {
+                        return err
+                }
+                // send a response if we actually managed to read a header.
+                if req != nil {
+                        server.sendResponse(sending, req, invalidRequest, codec, err.String())
+                        server.freeRequest(req)
+                }
+                return err
+        }
+        service.call(server, sending, mtype, req, argv, replyv, codec)
+        return nil
+}
+
 func (server *Server) getRequest() *Request {
         server.reqLock.Lock()
         req := server.freeReq
         if req == nil {
                 req = new(Request)
         } else {
                 server.freeReq = req.next
                 *req = Request{}
         }
         server.reqLock.Unlock()
@@ skipping 20 matching lines @@
         return resp
 }
 
 func (server *Server) freeResponse(resp *Response) {
         server.respLock.Lock()
         resp.next = server.freeResp
         server.freeResp = resp
         server.respLock.Unlock()
 }
 
-func (server *Server) readRequest(codec ServerCodec) (req *Request, service *service, mtype *methodType, err os.Error) {
+func (server *Server) readRequest(codec ServerCodec) (service *service, mtype *methodType, req *Request, argv, replyv reflect.Value, err os.Error) {
+        service, mtype, req, err = server.readRequestHeader(codec)
+        if err != nil {
+                if err == os.EOF || err == io.ErrUnexpectedEOF {
+                        return
+                }
+                // discard body
+                codec.ReadRequestBody(nil)
+                return
+        }
+
+        // Decode the argument value.
+        argIsValue := false // if true, need to indirect before calling.
+        if mtype.ArgType.Kind() == reflect.Ptr {
+                argv = reflect.New(mtype.ArgType.Elem())
+        } else {
+                argv = reflect.New(mtype.ArgType)
+                argIsValue = true
+        }
+        // argv guaranteed to be a pointer now.
+        if err = codec.ReadRequestBody(argv.Interface()); err != nil {
+                return
+        }
+        if argIsValue {
+                argv = argv.Elem()
+        }
+
+        replyv = reflect.New(mtype.ReplyType.Elem())
+        return
+}
+
+func (server *Server) readRequestHeader(codec ServerCodec) (service *service, mtype *methodType, req *Request, err os.Error) {
         // Grab the request header.
         req = server.getRequest()
         err = codec.ReadRequestHeader(req)
         if err != nil {
                 req = nil
                 if err == os.EOF || err == io.ErrUnexpectedEOF {
                         return
                 }
                 err = os.NewError("rpc: server cannot decode request: " + err.String())
                 return
@@ skipping 64 matching lines @@
 func ServeConn(conn io.ReadWriteCloser) {
         DefaultServer.ServeConn(conn)
 }
 
 // ServeCodec is like ServeConn but uses the specified codec to
 // decode requests and encode responses.
 func ServeCodec(codec ServerCodec) {
         DefaultServer.ServeCodec(codec)
 }
 
+// ServeSingle is like ServeCodec but synchronously serves a single request

[rsc, 2011/08/15 16:11:09]

same

[sougou, 2011/08/15 20:17:18]

Done.

+// without launching a goroutine. It does not close the codec upon completion.
+func ServeSingle(codec ServerCodec) os.Error {
+        return DefaultServer.ServeSingle(codec)
+}
+
 // Accept accepts connections on the listener and serves requests
 // to DefaultServer for each incoming connection.··
 // Accept blocks; the caller typically invokes it in a go statement.
 func Accept(lis net.Listener) { DefaultServer.Accept(lis) }
 
 // Can connect to RPC service using HTTP CONNECT to rpcPath.
 var connected = "200 Connected to Go RPC"
 
 // ServeHTTP implements an http.Handler that answers RPC requests.
 func (server *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
@@ skipping 19 matching lines @@
         http.Handle(rpcPath, server)
         http.Handle(debugPath, debugHTTP{server})
 }
 
 // HandleHTTP registers an HTTP handler for RPC messages to DefaultServer
 // on DefaultRPCPath and a debugging handler on DefaultDebugPath.
 // It is still necessary to invoke http.Serve(), typically in a go statement.
 func HandleHTTP() {
         DefaultServer.HandleHTTP(DefaultRPCPath, DefaultDebugPath)
 }