1/* 2 * 3 * Copyright 2014 gRPC authors. 4 * 5 * Licensed under the Apache License, Version 2.0 (the "License"); 6 * you may not use this file except in compliance with the License. 7 * You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 * 17 */ 18 19package grpc 20 21import ( 22 "bytes" 23 "io" 24 "time" 25 26 "golang.org/x/net/context" 27 "golang.org/x/net/trace" 28 "google.golang.org/grpc/balancer" 29 "google.golang.org/grpc/codes" 30 "google.golang.org/grpc/peer" 31 "google.golang.org/grpc/stats" 32 "google.golang.org/grpc/status" 33 "google.golang.org/grpc/transport" 34) 35 36// recvResponse receives and parses an RPC response. 37// On error, it returns the error and indicates whether the call should be retried. 38// 39// TODO(zhaoq): Check whether the received message sequence is valid. 40// TODO ctx is used for stats collection and processing. It is the context passed from the application. 41func recvResponse(ctx context.Context, dopts dialOptions, t transport.ClientTransport, c *callInfo, stream *transport.Stream, reply interface{}) (err error) { 42 // Try to acquire header metadata from the server if there is any. 43 defer func() { 44 if err != nil { 45 if _, ok := err.(transport.ConnectionError); !ok { 46 t.CloseStream(stream, err) 47 } 48 } 49 }() 50 c.headerMD, err = stream.Header() 51 if err != nil { 52 return 53 } 54 p := &parser{r: stream} 55 var inPayload *stats.InPayload 56 if dopts.copts.StatsHandler != nil { 57 inPayload = &stats.InPayload{ 58 Client: true, 59 } 60 } 61 for { 62 if c.maxReceiveMessageSize == nil { 63 return Errorf(codes.Internal, "callInfo maxReceiveMessageSize field uninitialized(nil)") 64 } 65 if err = recv(p, dopts.codec, stream, dopts.dc, reply, *c.maxReceiveMessageSize, inPayload); err != nil { 66 if err == io.EOF { 67 break 68 } 69 return 70 } 71 } 72 if inPayload != nil && err == io.EOF && stream.Status().Code() == codes.OK { 73 // TODO in the current implementation, inTrailer may be handled before inPayload in some cases. 74 // Fix the order if necessary. 75 dopts.copts.StatsHandler.HandleRPC(ctx, inPayload) 76 } 77 c.trailerMD = stream.Trailer() 78 return nil 79} 80 81// sendRequest writes out various information of an RPC such as Context and Message. 82func sendRequest(ctx context.Context, dopts dialOptions, compressor Compressor, c *callInfo, callHdr *transport.CallHdr, stream *transport.Stream, t transport.ClientTransport, args interface{}, opts *transport.Options) (err error) { 83 defer func() { 84 if err != nil { 85 // If err is connection error, t will be closed, no need to close stream here. 86 if _, ok := err.(transport.ConnectionError); !ok { 87 t.CloseStream(stream, err) 88 } 89 } 90 }() 91 var ( 92 cbuf *bytes.Buffer 93 outPayload *stats.OutPayload 94 ) 95 if compressor != nil { 96 cbuf = new(bytes.Buffer) 97 } 98 if dopts.copts.StatsHandler != nil { 99 outPayload = &stats.OutPayload{ 100 Client: true, 101 } 102 } 103 hdr, data, err := encode(dopts.codec, args, compressor, cbuf, outPayload) 104 if err != nil { 105 return err 106 } 107 if c.maxSendMessageSize == nil { 108 return Errorf(codes.Internal, "callInfo maxSendMessageSize field uninitialized(nil)") 109 } 110 if len(data) > *c.maxSendMessageSize { 111 return Errorf(codes.ResourceExhausted, "grpc: trying to send message larger than max (%d vs. %d)", len(data), *c.maxSendMessageSize) 112 } 113 err = t.Write(stream, hdr, data, opts) 114 if err == nil && outPayload != nil { 115 outPayload.SentTime = time.Now() 116 dopts.copts.StatsHandler.HandleRPC(ctx, outPayload) 117 } 118 // t.NewStream(...) could lead to an early rejection of the RPC (e.g., the service/method 119 // does not exist.) so that t.Write could get io.EOF from wait(...). Leave the following 120 // recvResponse to get the final status. 121 if err != nil && err != io.EOF { 122 return err 123 } 124 // Sent successfully. 125 return nil 126} 127 128// Invoke sends the RPC request on the wire and returns after response is received. 129// Invoke is called by generated code. Also users can call Invoke directly when it 130// is really needed in their use cases. 131func Invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) error { 132 if cc.dopts.unaryInt != nil { 133 return cc.dopts.unaryInt(ctx, method, args, reply, cc, invoke, opts...) 134 } 135 return invoke(ctx, method, args, reply, cc, opts...) 136} 137 138func invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) (e error) { 139 c := defaultCallInfo() 140 mc := cc.GetMethodConfig(method) 141 if mc.WaitForReady != nil { 142 c.failFast = !*mc.WaitForReady 143 } 144 145 if mc.Timeout != nil && *mc.Timeout >= 0 { 146 var cancel context.CancelFunc 147 ctx, cancel = context.WithTimeout(ctx, *mc.Timeout) 148 defer cancel() 149 } 150 151 opts = append(cc.dopts.callOptions, opts...) 152 for _, o := range opts { 153 if err := o.before(c); err != nil { 154 return toRPCErr(err) 155 } 156 } 157 defer func() { 158 for _, o := range opts { 159 o.after(c) 160 } 161 }() 162 163 c.maxSendMessageSize = getMaxSize(mc.MaxReqSize, c.maxSendMessageSize, defaultClientMaxSendMessageSize) 164 c.maxReceiveMessageSize = getMaxSize(mc.MaxRespSize, c.maxReceiveMessageSize, defaultClientMaxReceiveMessageSize) 165 166 if EnableTracing { 167 c.traceInfo.tr = trace.New("grpc.Sent."+methodFamily(method), method) 168 defer c.traceInfo.tr.Finish() 169 c.traceInfo.firstLine.client = true 170 if deadline, ok := ctx.Deadline(); ok { 171 c.traceInfo.firstLine.deadline = deadline.Sub(time.Now()) 172 } 173 c.traceInfo.tr.LazyLog(&c.traceInfo.firstLine, false) 174 // TODO(dsymonds): Arrange for c.traceInfo.firstLine.remoteAddr to be set. 175 defer func() { 176 if e != nil { 177 c.traceInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{e}}, true) 178 c.traceInfo.tr.SetError() 179 } 180 }() 181 } 182 ctx = newContextWithRPCInfo(ctx, c.failFast) 183 sh := cc.dopts.copts.StatsHandler 184 if sh != nil { 185 ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method, FailFast: c.failFast}) 186 begin := &stats.Begin{ 187 Client: true, 188 BeginTime: time.Now(), 189 FailFast: c.failFast, 190 } 191 sh.HandleRPC(ctx, begin) 192 defer func() { 193 end := &stats.End{ 194 Client: true, 195 EndTime: time.Now(), 196 Error: e, 197 } 198 sh.HandleRPC(ctx, end) 199 }() 200 } 201 topts := &transport.Options{ 202 Last: true, 203 Delay: false, 204 } 205 for { 206 var ( 207 err error 208 t transport.ClientTransport 209 stream *transport.Stream 210 // Record the put handler from Balancer.Get(...). It is called once the 211 // RPC has completed or failed. 212 put func(balancer.DoneInfo) 213 ) 214 // TODO(zhaoq): Need a formal spec of fail-fast. 215 callHdr := &transport.CallHdr{ 216 Host: cc.authority, 217 Method: method, 218 } 219 if cc.dopts.cp != nil { 220 callHdr.SendCompress = cc.dopts.cp.Type() 221 } 222 if c.creds != nil { 223 callHdr.Creds = c.creds 224 } 225 226 gopts := BalancerGetOptions{ 227 BlockingWait: !c.failFast, 228 } 229 t, put, err = cc.getTransport(ctx, gopts) 230 if err != nil { 231 // TODO(zhaoq): Probably revisit the error handling. 232 if _, ok := status.FromError(err); ok { 233 return err 234 } 235 if err == errConnClosing || err == errConnUnavailable { 236 if c.failFast { 237 return Errorf(codes.Unavailable, "%v", err) 238 } 239 continue 240 } 241 // All the other errors are treated as Internal errors. 242 return Errorf(codes.Internal, "%v", err) 243 } 244 if c.traceInfo.tr != nil { 245 c.traceInfo.tr.LazyLog(&payload{sent: true, msg: args}, true) 246 } 247 stream, err = t.NewStream(ctx, callHdr) 248 if err != nil { 249 if put != nil { 250 if _, ok := err.(transport.ConnectionError); ok { 251 // If error is connection error, transport was sending data on wire, 252 // and we are not sure if anything has been sent on wire. 253 // If error is not connection error, we are sure nothing has been sent. 254 updateRPCInfoInContext(ctx, rpcInfo{bytesSent: true, bytesReceived: false}) 255 } 256 put(balancer.DoneInfo{Err: err}) 257 } 258 if _, ok := err.(transport.ConnectionError); (ok || err == transport.ErrStreamDrain) && !c.failFast { 259 continue 260 } 261 return toRPCErr(err) 262 } 263 if peer, ok := peer.FromContext(stream.Context()); ok { 264 c.peer = peer 265 } 266 err = sendRequest(ctx, cc.dopts, cc.dopts.cp, c, callHdr, stream, t, args, topts) 267 if err != nil { 268 if put != nil { 269 updateRPCInfoInContext(ctx, rpcInfo{ 270 bytesSent: stream.BytesSent(), 271 bytesReceived: stream.BytesReceived(), 272 }) 273 put(balancer.DoneInfo{Err: err}) 274 } 275 // Retry a non-failfast RPC when 276 // i) there is a connection error; or 277 // ii) the server started to drain before this RPC was initiated. 278 if _, ok := err.(transport.ConnectionError); (ok || err == transport.ErrStreamDrain) && !c.failFast { 279 continue 280 } 281 return toRPCErr(err) 282 } 283 err = recvResponse(ctx, cc.dopts, t, c, stream, reply) 284 if err != nil { 285 if put != nil { 286 updateRPCInfoInContext(ctx, rpcInfo{ 287 bytesSent: stream.BytesSent(), 288 bytesReceived: stream.BytesReceived(), 289 }) 290 put(balancer.DoneInfo{Err: err}) 291 } 292 if _, ok := err.(transport.ConnectionError); (ok || err == transport.ErrStreamDrain) && !c.failFast { 293 continue 294 } 295 return toRPCErr(err) 296 } 297 if c.traceInfo.tr != nil { 298 c.traceInfo.tr.LazyLog(&payload{sent: false, msg: reply}, true) 299 } 300 t.CloseStream(stream, nil) 301 if put != nil { 302 updateRPCInfoInContext(ctx, rpcInfo{ 303 bytesSent: stream.BytesSent(), 304 bytesReceived: stream.BytesReceived(), 305 }) 306 put(balancer.DoneInfo{Err: err}) 307 } 308 return stream.Status().Err() 309 } 310} 311