golang源码分析：grpc 链接池（6）超时的设置和传递

rpc error: code = DeadlineExceeded desc = context deadline exceeded

var DeadlineExceeded error = deadlineExceededError{}
func (deadlineExceededError) Error() string   { return "context deadline exceeded" }

func WithDeadline(parent Context, d time.Time) (Context, CancelFunc) {
  if cur, ok := parent.Deadline(); ok && cur.Before(d) {
    // The current deadline is already sooner than the new one.
    return WithCancel(parent)
  }


  dur := time.Until(d)
  if dur <= 0 {
    c.cancel(true, DeadlineExceeded) // deadline has already passed
    return c, func() { c.cancel(false, Canceled) }
  }


  if c.err == nil {
    c.timer = time.AfterFunc(dur, func() {
      c.cancel(true, DeadlineExceeded)
    })
  }


  return c, func() { c.cancel(true, Canceled) }
}

func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) {
  return WithDeadline(parent, time.Now().Add(timeout))
}

type ClientConn struct {
dopts           dialOptions
}

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
    if cc.dopts.timeout > 0 {
      var cancel context.CancelFunc
      ctx, cancel = context.WithTimeout(ctx, cc.dopts.timeout)
      defer cancel()
    }
}

func invoke(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error {
  cs, err := newClientStream(ctx, unaryStreamDesc, cc, method, opts...)

func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
    var newStream = func(ctx context.Context, done func()) (iresolver.ClientStream, error) {
    return newClientStreamWithParams(ctx, desc, cc, method, mc, onCommit, done, opts...)
    }
    rpcConfig, err := cc.safeConfigSelector.SelectConfig(rpcInfo)

func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, mc serviceconfig.MethodConfig, onCommit, doneFunc func(), opts ...CallOption) (_ iresolver.ClientStream, err error) {
    if mc.Timeout != nil && *mc.Timeout >= 0 {
    ctx, cancel = context.WithTimeout(ctx, *mc.Timeout)
    } else {
      ctx, cancel = context.WithCancel(ctx)
    }

clientConn, err := grpc.Dial(serverAddress, grpc.WithTimeout(5 * time.Second), grpc.WithInsecure())
if err != nil {
        log.Println("Dial failed!")
        return err
}

c := pb.NewGreeterClient(conn)
c.SayHello(context.Background(), &pb.HelloRequest{Name: "world"},
            WithForcedTimeout(time.Duration(10)*time.Second))

func WithTimeout(d time.Duration) DialOption {
  return newFuncDialOption(func(o *dialOptions) {
    o.timeout = d
  })
}

type dialOptions struct {
    timeout                     time.Duration
}

type DialOption interface {
  apply(*dialOptions)
}

type ClientConn struct {
    dopts           dialOptions   
}

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
    if cc.dopts.timeout > 0 {
    var cancel context.CancelFunc
    ctx, cancel = context.WithTimeout(ctx, cc.dopts.timeout)
    defer cancel()
  }
}

func TimeoutInterceptor(t time.Duration) grpc.UnaryClientInterceptor {
    return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn,
        invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
        timeout := t
        if v, ok := getForcedTimeout(opts); ok {
            timeout = v
        }
        if timeout <= 0 {
            return invoker(ctx, method, req, reply, cc, opts...)
        }
        ctx, cancel := context.WithTimeout(ctx, timeout)
        defer cancel()
        return invoker(ctx, method, req, reply, cc, opts...)
    }
}

func getForcedTimeout(callOptions []grpc.CallOption) (time.Duration, bool) {
    for _, opt := range callOptions {
        if co, ok := opt.(TimeoutCallOption); ok {
            return co.forcedTimeout, true
        }
    }
    return 0, false
}

type TimeoutCallOption struct {
    grpc.EmptyCallOption
    forcedTimeout time.Duration
}
func WithForcedTimeout(forceTimeout time.Duration) TimeoutCallOption {
    return TimeoutCallOption{forcedTimeout: forceTimeout}
}

grpc-timeout: 4995884u

func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream, error) {
    headerFields, err := t.createHeaderFields(ctx, callHdr)

func (t *http2Client) createHeaderFields(ctx context.Context, callHdr *CallHdr) ([]hpack.HeaderField, error) {
    if dl, ok := ctx.Deadline(); ok {
    // Send out timeout regardless its value. The server can detect timeout context by itself.
    // TODO(mmukhi): Perhaps this field should be updated when actually writing out to the wire.
    timeout := time.Until(dl)
    headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-timeout", Value: grpcutil.EncodeDuration(timeout)})
  }

func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request, stats []stats.Handler) (ServerTransport, error) {
  if v := r.Header.Get("grpc-timeout"); v != "" {
  to, err := decodeTimeout(v)
  if err != nil {
  return nil, status.Errorf(codes.Internal, "malformed time-out: %v", err)
  }
  st.timeoutSet = true
  st.timeout = to
  }
  
  if timeoutSet {
  s.ctx, s.cancel = context.WithTimeout(t.ctx, timeout)
  } else {
  s.ctx, s.cancel = context.WithCancel(t.ctx)
  }

func (ht *serverHandlerTransport) HandleStreams(startStream func(*Stream), traceCtx func(context.Context, string) context.Context) {
if ht.timeoutSet {
ctx, cancel = context.WithTimeout(ctx, ht.timeout)
} else {
ctx, cancel = context.WithCancel(ctx)
}

func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (fatal bool) {
case "grpc-timeout":
timeoutSet = true
var err error
if timeout, err = decodeTimeout(hf.Value); err != nil {
headerError = true
}