code review 5305084: rpc: avoid infinite loop on input error

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 {
-                service, mtype, req, argv, replyv, err := server.readRequest(codec)
+                service, mtype, req, argv, replyv, keepReading, err := server.readRequest(codec)
                 if err != nil {
                         if err != os.EOF {
                                 log.Println("rpc:", err)
                         }
-                        if err == os.EOF || err == io.ErrUnexpectedEOF {
+                        if !keepReading {
                                 break
                         }
                         // 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
                 }
                 go service.call(server, sending, mtype, req, argv, replyv, codec)
         }
         codec.Close()
 }
 
 // ServeRequest is like ServeCodec but synchronously serves a single request.
 // It does not close the codec upon completion.
 func (server *Server) ServeRequest(codec ServerCodec) os.Error {
         sending := new(sync.Mutex)
-        service, mtype, req, argv, replyv, err := server.readRequest(codec)
+        service, mtype, req, argv, replyv, keepReading, err := server.readRequest(codec)

[rog, 2011/11/02 21:28:31]

does that set a record for most number of returned values?

         if err != nil {
-                if err == os.EOF || err == io.ErrUnexpectedEOF {
+                if !keepReading {
                         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)
@@ skipping 33 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) (service *service, mtype *methodType, req *Request, argv, replyv reflect.Value, err os.Error) {
-        service, mtype, req, err = server.readRequestHeader(codec)
+func (server *Server) readRequest(codec ServerCodec) (service *service, mtype *methodType, req *Request, argv, replyv reflect.Value, keepReading bool, err os.Error) {
+        service, mtype, req, keepReading, err = server.readRequestHeader(codec)
         if err != nil {
-                if err == os.EOF || err == io.ErrUnexpectedEOF {
+                if !keepReading {
                         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) {
+func (server *Server) readRequestHeader(codec ServerCodec) (service *service, mtype *methodType, req *Request, keepReading bool, 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
         }
+
+        // We read the header successfully.  If we see an error now,
+        // we can still recover and move on to the next request.
+        keepReading = true
 
         serviceMethod := strings.Split(req.ServiceMethod, ".")
         if len(serviceMethod) != 2 {
                 err = os.NewError("rpc: service/method request ill-formed: " + req.ServiceMethod)
                 return
         }
         // Look up the request.
         server.mu.Lock()
         service = server.serviceMap[serviceMethod[0]]
         server.mu.Unlock()
@@ skipping 98 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)
 }