golang源码分析（35）grpc

func main() {
  // Set up a connection to the server.
  //建立一个链接
  conn, err := grpc.Dial(address, grpc.WithInsecure())
  if err != nil {
    log.Fatalf("did not connect: %v", err)
  }
  defer conn.Close()
  c := pb.NewGreeterClient(conn)

  // Contact the server and print out its response.
  name := defaultName
  if len(os.Args) > 1 {
    name = os.Args[1]
  }
  //调用函数
  r, err := c.SayHello(context.Background(), &pb.HelloRequest{Name: name})
  if err != nil {
    log.Fatalf("could not greet: %v", err)
  }
  log.Printf("Greeting: %s", r.Message)
}

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
  cc := &ClientConn{
    target: target,
    conns:  make(map[Address]*addrConn),
  }
  cc.ctx, cc.cancel = context.WithCancel(context.Background())
  defer func() {
    select {
    case <-ctx.Done():
      conn, err = nil, ctx.Err()
    default:
    }

    if err != nil {
      cc.Close()
    }
  }()

  //设置grpc的各种选项
  for _, opt := range opts {
    opt(&cc.dopts)
  }

  // Set defaults.
  if cc.dopts.codec == nil {
    //默认用protobuf编解码
    cc.dopts.codec = protoCodec{}
  }
  if cc.dopts.bs == nil {
    cc.dopts.bs = DefaultBackoffConfig
  }
  creds := cc.dopts.copts.TransportCredentials
  //验证信息
  if creds != nil && creds.Info().ServerName != "" {
    cc.authority = creds.Info().ServerName
  } else {
    colonPos := strings.LastIndex(target, ":")
    if colonPos == -1 {
      colonPos = len(target)
    }
    cc.authority = target[:colonPos]
  }
  var ok bool
  waitC := make(chan error, 1)
  //启动一个goroutine启动名字服务器(类似dns)
  go func() {
    var addrs []Address
    if cc.dopts.balancer == nil {
      // Connect to target directly if balancer is nil.
      // 如果没设置负载均衡器，则直接连接
      addrs = append(addrs, Address{Addr: target})
    } else {
      var credsClone credentials.TransportCredentials
      if creds != nil {
        credsClone = creds.Clone()
      }
      config := BalancerConfig{
        DialCreds: credsClone,
      }
      //启动负载均衡服务
      if err := cc.dopts.balancer.Start(target, config); err != nil {
        waitC <- err
        return
      }
      ch := cc.dopts.balancer.Notify()
      if ch == nil {
        // There is no name resolver installed.
        addrs = append(addrs, Address{Addr: target})
      } else {
        addrs, ok = <-ch
        if !ok || len(addrs) == 0 {
          waitC <- errNoAddr
          return
        }
      }
    }
    for _, a := range addrs {
      //给每个地址一个conn，连接池
      if err := cc.resetAddrConn(a, false, nil); err != nil {
        waitC <- err
        return
      }
    }
    close(waitC)
  }()
  var timeoutCh <-chan time.Time
  if cc.dopts.timeout > 0 {
    timeoutCh = time.After(cc.dopts.timeout)
  }
  select {
  case <-ctx.Done():
    return nil, ctx.Err()
  case err := <-waitC:
    if err != nil {
      return nil, err
    }
  case <-timeoutCh:
    return nil, ErrClientConnTimeout
  }
  // If balancer is nil or balancer.Notify() is nil, ok will be false here.
  // The lbWatcher goroutine will not be created.
  if ok {
    go cc.lbWatcher()
  }
  return cc, nil
}

func (c *greeterClient) SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error) {
  out := new(HelloReply)
  //调用实际的发送请求函数
  err := grpc.Invoke(ctx, "/helloworld.Greeter/SayHello", in, out, c.cc, opts...)
  if err != nil {
    return nil, err
  }
  return out, nil
}

//最后主要是invoke函数
func invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) (e error) {
  c := defaultCallInfo
  for _, o := range opts {
    //调用之前的hook
    if err := o.before(&c); err != nil {
      return toRPCErr(err)
    }
  }
  defer func() {
    for _, o := range opts {
      //执行完后的hook
      o.after(&c)
    }
  }()
  //trace相关代码
  if EnableTracing {
    c.traceInfo.tr = trace.New("grpc.Sent."+methodFamily(method), method)
    defer c.traceInfo.tr.Finish()
    c.traceInfo.firstLine.client = true
    if deadline, ok := ctx.Deadline(); ok {
      c.traceInfo.firstLine.deadline = deadline.Sub(time.Now())
    }
    c.traceInfo.tr.LazyLog(&c.traceInfo.firstLine, false)
    // TODO(dsymonds): Arrange for c.traceInfo.firstLine.remoteAddr to be set.
    defer func() {
      if e != nil {
        c.traceInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{e}}, true)
        c.traceInfo.tr.SetError()
      }
    }()
  }
  //统计相关代码
  if stats.On() {
    ctx = stats.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method})
    begin := &stats.Begin{
      Client:    true,
      BeginTime: time.Now(),
      FailFast:  c.failFast,
    }
    stats.HandleRPC(ctx, begin)
  }
  defer func() {
    //结束后的统计相关代码
    if stats.On() {
      end := &stats.End{
        Client:  true,
        EndTime: time.Now(),
        Error:   e,
      }
      stats.HandleRPC(ctx, end)
    }
  }()
  topts := &transport.Options{
    Last:  true,
    Delay: false,
  }
  for {
    var (
      err    error
      t      transport.ClientTransport
      stream *transport.Stream
      // Record the put handler from Balancer.Get(...). It is called once the
      // RPC has completed or failed.
      put func()
    )
    // TODO(zhaoq): Need a formal spec of fail-fast.
    //传输层的配置
    callHdr := &transport.CallHdr{
      Host:   cc.authority,
      Method: method,
    }
    if cc.dopts.cp != nil {
      callHdr.SendCompress = cc.dopts.cp.Type()
    }
    gopts := BalancerGetOptions{
      BlockingWait: !c.failFast,
    }
    //得到传输成连接，在http2中一个传输单位是一个流。
    t, put, err = cc.getTransport(ctx, gopts)
    if err != nil {
      // TODO(zhaoq): Probably revisit the error handling.
      if _, ok := err.(*rpcError); ok {
        return err
      }
      if err == errConnClosing || err == errConnUnavailable {
        if c.failFast {
          return Errorf(codes.Unavailable, "%v", err)
        }
        continue
      }
      // All the other errors are treated as Internal errors.
      return Errorf(codes.Internal, "%v", err)
    }
    if c.traceInfo.tr != nil {
      c.traceInfo.tr.LazyLog(&payload{sent: true, msg: args}, true)
    }
    // 发送请求
    stream, err = sendRequest(ctx, cc.dopts.codec, cc.dopts.cp, callHdr, t, args, topts)
    if err != nil {
      if put != nil {
        put()
        put = nil
      }
      // Retry a non-failfast RPC when
      // i) there is a connection error; or
      // ii) the server started to drain before this RPC was initiated.
      // 在这两种情况下重试，1 链接错误 2 在rpc初始化之前服务端已经开始服务
      if _, ok := err.(transport.ConnectionError); ok || err == transport.ErrStreamDrain {
        if c.failFast {
          return toRPCErr(err)
        }
        continue
      }
      return toRPCErr(err)
    }
    //收消息
    err = recvResponse(ctx, cc.dopts, t, &c, stream, reply)
    if err != nil {
      if put != nil {
        put()
        put = nil
      }
      if _, ok := err.(transport.ConnectionError); ok || err == transport.ErrStreamDrain {
        if c.failFast {
          return toRPCErr(err)
        }
        continue
      }
      return toRPCErr(err)
    }
    if c.traceInfo.tr != nil {
      c.traceInfo.tr.LazyLog(&payload{sent: false, msg: reply}, true)
    }
    //关闭一个http2流
    t.CloseStream(stream, nil)
    if put != nil {
      put()
      put = nil
    }
    //Errorf会判断返回十分ok
    return Errorf(stream.StatusCode(), "%s", stream.StatusDesc())
  }
}

func sendRequest(ctx context.Context, codec Codec, compressor Compressor, callHdr *transport.CallHdr, t transport.ClientTransport, args interface{}, opts *transport.Options) (_ *transport.Stream, err error) {
  // 创建一个http2流
  stream, err := t.NewStream(ctx, callHdr)
  if err != nil {
    return nil, err
  }
  defer func() {
    if err != nil {
      // If err is connection error, t will be closed, no need to close stream here.
      if _, ok := err.(transport.ConnectionError); !ok {
        t.CloseStream(stream, err)
      }
    }
  }()
  var (
    cbuf       *bytes.Buffer
    outPayload *stats.OutPayload
  )
  //压缩不为空
  if compressor != nil {
    cbuf = new(bytes.Buffer)
  }
  //统计
  if stats.On() {
    outPayload = &stats.OutPayload{
      Client: true,
    }
  }
  //编码并压缩数据
  outBuf, err := encode(codec, args, compressor, cbuf, outPayload)
  if err != nil {
    return nil, Errorf(codes.Internal, "grpc: %v", err)
  }
  //写入流
  err = t.Write(stream, outBuf, opts)
  if err == nil && outPayload != nil {
    outPayload.SentTime = time.Now()
    stats.HandleRPC(ctx, outPayload)
  }
  // t.NewStream(...) could lead to an early rejection of the RPC (e.g., the service/method
  // does not exist.) so that t.Write could get io.EOF from wait(...). Leave the following
  // recvResponse to get the final status.
  if err != nil && err != io.EOF {
    return nil, err
  }
  // Sent successfully.
  return stream, nil
}

func recvResponse(ctx context.Context, dopts dialOptions, t transport.ClientTransport, c *callInfo, stream *transport.Stream, reply interface{}) (err error) {
  // Try to acquire header metadata from the server if there is any.
  defer func() {
    if err != nil {
      if _, ok := err.(transport.ConnectionError); !ok {
        t.CloseStream(stream, err)
      }
    }
  }()
  c.headerMD, err = stream.Header()
  if err != nil {
    return
  }
  p := &parser{r: stream}
  var inPayload *stats.InPayload
  if stats.On() {
    inPayload = &stats.InPayload{
      Client: true,
    }
  }
  for {
    //一直读到流关闭
    if err = recv(p, dopts.codec, stream, dopts.dc, reply, math.MaxInt32, inPayload); err != nil {
      if err == io.EOF {
        break
      }
      return
    }
  }
  if inPayload != nil && err == io.EOF && stream.StatusCode() == codes.OK {
    // TODO in the current implementation, inTrailer may be handled before inPayload in some cases.
    // Fix the order if necessary.
    stats.HandleRPC(ctx, inPayload)
  }
  c.trailerMD = stream.Trailer()
  return nil
}

func recv(p *parser, c Codec, s *transport.Stream, dc Decompressor, m interface{}, maxMsgSize int, inPayload *stats.InPayload) error {
  //接受数据
  pf, d, err := p.recvMsg(maxMsgSize)
  if err != nil {
    return err
  }
  if inPayload != nil {
    inPayload.WireLength = len(d)
  }
  if err := checkRecvPayload(pf, s.RecvCompress(), dc); err != nil {
    return err
  }
  if pf == compressionMade {
    //解压
    d, err = dc.Do(bytes.NewReader(d))
    if err != nil {
      return Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
    }
  }
  if len(d) > maxMsgSize {
    // TODO: Revisit the error code. Currently keep it consistent with java
    // implementation.
    return Errorf(codes.Internal, "grpc: received a message of %d bytes exceeding %d limit", len(d), maxMsgSize)
  }
  //数据解码
  if err := c.Unmarshal(d, m); err != nil {
    return Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err)
  }
  if inPayload != nil {
    inPayload.RecvTime = time.Now()
    inPayload.Payload = m
    // TODO truncate large payload.
    inPayload.Data = d
    inPayload.Length = len(d)
  }
  return nil
}