vendor/google.golang.org/grpc/call.go

   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
  19 package grpc
  20
  21 import (
  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.
  41 func 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.
  82 func 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
 129 // received.  This is typically called by generated code.
 130 func (cc *ClientConn) Invoke(ctx context.Context, method string, args, reply interface{}, opts ...CallOption) error {
 131         if cc.dopts.unaryInt != nil {
 132                 return cc.dopts.unaryInt(ctx, method, args, reply, cc, invoke, opts...)
 133         }
 134         return invoke(ctx, method, args, reply, cc, opts...)
 135 }
 136
 137 // Invoke sends the RPC request on the wire and returns after response is
 138 // received.  This is typically called by generated code.
 139 //
 140 // DEPRECATED: Use ClientConn.Invoke instead.
 141 func Invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) error {
 142         return cc.Invoke(ctx, method, args, reply, opts...)
 143 }
 144
 145 func invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) (e error) {
 146         c := defaultCallInfo()
 147         mc := cc.GetMethodConfig(method)
 148         if mc.WaitForReady != nil {
 149                 c.failFast = !*mc.WaitForReady
 150         }
 151
 152         if mc.Timeout != nil && *mc.Timeout >= 0 {
 153                 var cancel context.CancelFunc
 154                 ctx, cancel = context.WithTimeout(ctx, *mc.Timeout)
 155                 defer cancel()
 156         }
 157
 158         opts = append(cc.dopts.callOptions, opts...)
 159         for _, o := range opts {
 160                 if err := o.before(c); err != nil {
 161                         return toRPCErr(err)
 162                 }
 163         }
 164         defer func() {
 165                 for _, o := range opts {
 166                         o.after(c)
 167                 }
 168         }()
 169
 170         c.maxSendMessageSize = getMaxSize(mc.MaxReqSize, c.maxSendMessageSize, defaultClientMaxSendMessageSize)
 171         c.maxReceiveMessageSize = getMaxSize(mc.MaxRespSize, c.maxReceiveMessageSize, defaultClientMaxReceiveMessageSize)
 172
 173         if EnableTracing {
 174                 c.traceInfo.tr = trace.New("grpc.Sent."+methodFamily(method), method)
 175                 defer c.traceInfo.tr.Finish()
 176                 c.traceInfo.firstLine.client = true
 177                 if deadline, ok := ctx.Deadline(); ok {
 178                         c.traceInfo.firstLine.deadline = deadline.Sub(time.Now())
 179                 }
 180                 c.traceInfo.tr.LazyLog(&c.traceInfo.firstLine, false)
 181                 // TODO(dsymonds): Arrange for c.traceInfo.firstLine.remoteAddr to be set.
 182                 defer func() {
 183                         if e != nil {
 184                                 c.traceInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{e}}, true)
 185                                 c.traceInfo.tr.SetError()
 186                         }
 187                 }()
 188         }
 189         ctx = newContextWithRPCInfo(ctx, c.failFast)
 190         sh := cc.dopts.copts.StatsHandler
 191         if sh != nil {
 192                 ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method, FailFast: c.failFast})
 193                 begin := &stats.Begin{
 194                         Client:    true,
 195                         BeginTime: time.Now(),
 196                         FailFast:  c.failFast,
 197                 }
 198                 sh.HandleRPC(ctx, begin)
 199                 defer func() {
 200                         end := &stats.End{
 201                                 Client:  true,
 202                                 EndTime: time.Now(),
 203                                 Error:   e,
 204                         }
 205                         sh.HandleRPC(ctx, end)
 206                 }()
 207         }
 208         topts := &transport.Options{
 209                 Last:  true,
 210                 Delay: false,
 211         }
 212         for {
 213                 var (
 214                         err    error
 215                         t      transport.ClientTransport
 216                         stream *transport.Stream
 217                         // Record the done handler from Balancer.Get(...). It is called once the
 218                         // RPC has completed or failed.
 219                         done func(balancer.DoneInfo)
 220                 )
 221                 // TODO(zhaoq): Need a formal spec of fail-fast.
 222                 callHdr := &transport.CallHdr{
 223                         Host:   cc.authority,
 224                         Method: method,
 225                 }
 226                 if cc.dopts.cp != nil {
 227                         callHdr.SendCompress = cc.dopts.cp.Type()
 228                 }
 229                 if c.creds != nil {
 230                         callHdr.Creds = c.creds
 231                 }
 232
 233                 t, done, err = cc.getTransport(ctx, c.failFast)
 234                 if err != nil {
 235                         // TODO(zhaoq): Probably revisit the error handling.
 236                         if _, ok := status.FromError(err); ok {
 237                                 return err
 238                         }
 239                         if err == errConnClosing || err == errConnUnavailable {
 240                                 if c.failFast {
 241                                         return Errorf(codes.Unavailable, "%v", err)
 242                                 }
 243                                 continue
 244                         }
 245                         // All the other errors are treated as Internal errors.
 246                         return Errorf(codes.Internal, "%v", err)
 247                 }
 248                 if c.traceInfo.tr != nil {
 249                         c.traceInfo.tr.LazyLog(&payload{sent: true, msg: args}, true)
 250                 }
 251                 stream, err = t.NewStream(ctx, callHdr)
 252                 if err != nil {
 253                         if done != nil {
 254                                 if _, ok := err.(transport.ConnectionError); ok {
 255                                         // If error is connection error, transport was sending data on wire,
 256                                         // and we are not sure if anything has been sent on wire.
 257                                         // If error is not connection error, we are sure nothing has been sent.
 258                                         updateRPCInfoInContext(ctx, rpcInfo{bytesSent: true, bytesReceived: false})
 259                                 }
 260                                 done(balancer.DoneInfo{Err: err})
 261                         }
 262                         if _, ok := err.(transport.ConnectionError); (ok || err == transport.ErrStreamDrain) && !c.failFast {
 263                                 continue
 264                         }
 265                         return toRPCErr(err)
 266                 }
 267                 if peer, ok := peer.FromContext(stream.Context()); ok {
 268                         c.peer = peer
 269                 }
 270                 err = sendRequest(ctx, cc.dopts, cc.dopts.cp, c, callHdr, stream, t, args, topts)
 271                 if err != nil {
 272                         if done != nil {
 273                                 updateRPCInfoInContext(ctx, rpcInfo{
 274                                         bytesSent:     stream.BytesSent(),
 275                                         bytesReceived: stream.BytesReceived(),
 276                                 })
 277                                 done(balancer.DoneInfo{Err: err})
 278                         }
 279                         // Retry a non-failfast RPC when
 280                         // i) there is a connection error; or
 281                         // ii) the server started to drain before this RPC was initiated.
 282                         if _, ok := err.(transport.ConnectionError); (ok || err == transport.ErrStreamDrain) && !c.failFast {
 283                                 continue
 284                         }
 285                         return toRPCErr(err)
 286                 }
 287                 err = recvResponse(ctx, cc.dopts, t, c, stream, reply)
 288                 if err != nil {
 289                         if done != nil {
 290                                 updateRPCInfoInContext(ctx, rpcInfo{
 291                                         bytesSent:     stream.BytesSent(),
 292                                         bytesReceived: stream.BytesReceived(),
 293                                 })
 294                                 done(balancer.DoneInfo{Err: err})
 295                         }
 296                         if _, ok := err.(transport.ConnectionError); (ok || err == transport.ErrStreamDrain) && !c.failFast {
 297                                 continue
 298                         }
 299                         return toRPCErr(err)
 300                 }
 301                 if c.traceInfo.tr != nil {
 302                         c.traceInfo.tr.LazyLog(&payload{sent: false, msg: reply}, true)
 303                 }
 304                 t.CloseStream(stream, nil)
 305                 if done != nil {
 306                         updateRPCInfoInContext(ctx, rpcInfo{
 307                                 bytesSent:     stream.BytesSent(),
 308                                 bytesReceived: stream.BytesReceived(),
 309                         })
 310                         done(balancer.DoneInfo{Err: err})
 311                 }
 312                 return stream.Status().Err()
 313         }
 314 }