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 transport 20 21import ( 22 "context" 23 "fmt" 24 "io" 25 "math" 26 "net" 27 "strconv" 28 "strings" 29 "sync" 30 "sync/atomic" 31 "time" 32 33 "golang.org/x/net/http2" 34 "golang.org/x/net/http2/hpack" 35 36 "google.golang.org/grpc/codes" 37 "google.golang.org/grpc/credentials" 38 "google.golang.org/grpc/internal/channelz" 39 "google.golang.org/grpc/internal/syscall" 40 "google.golang.org/grpc/keepalive" 41 "google.golang.org/grpc/metadata" 42 "google.golang.org/grpc/peer" 43 "google.golang.org/grpc/stats" 44 "google.golang.org/grpc/status" 45) 46 47// http2Client implements the ClientTransport interface with HTTP2. 48type http2Client struct { 49 ctx context.Context 50 cancel context.CancelFunc 51 ctxDone <-chan struct{} // Cache the ctx.Done() chan. 52 userAgent string 53 md interface{} 54 conn net.Conn // underlying communication channel 55 loopy *loopyWriter 56 remoteAddr net.Addr 57 localAddr net.Addr 58 authInfo credentials.AuthInfo // auth info about the connection 59 60 readerDone chan struct{} // sync point to enable testing. 61 writerDone chan struct{} // sync point to enable testing. 62 // goAway is closed to notify the upper layer (i.e., addrConn.transportMonitor) 63 // that the server sent GoAway on this transport. 64 goAway chan struct{} 65 // awakenKeepalive is used to wake up keepalive when after it has gone dormant. 66 awakenKeepalive chan struct{} 67 68 framer *framer 69 // controlBuf delivers all the control related tasks (e.g., window 70 // updates, reset streams, and various settings) to the controller. 71 controlBuf *controlBuffer 72 fc *trInFlow 73 // The scheme used: https if TLS is on, http otherwise. 74 scheme string 75 76 isSecure bool 77 78 perRPCCreds []credentials.PerRPCCredentials 79 80 // Boolean to keep track of reading activity on transport. 81 // 1 is true and 0 is false. 82 activity uint32 // Accessed atomically. 83 kp keepalive.ClientParameters 84 keepaliveEnabled bool 85 86 statsHandler stats.Handler 87 88 initialWindowSize int32 89 90 // configured by peer through SETTINGS_MAX_HEADER_LIST_SIZE 91 maxSendHeaderListSize *uint32 92 93 bdpEst *bdpEstimator 94 // onPrefaceReceipt is a callback that client transport calls upon 95 // receiving server preface to signal that a succefull HTTP2 96 // connection was established. 97 onPrefaceReceipt func() 98 99 maxConcurrentStreams uint32 100 streamQuota int64 101 streamsQuotaAvailable chan struct{} 102 waitingStreams uint32 103 nextID uint32 104 105 mu sync.Mutex // guard the following variables 106 state transportState 107 activeStreams map[uint32]*Stream 108 // prevGoAway ID records the Last-Stream-ID in the previous GOAway frame. 109 prevGoAwayID uint32 110 // goAwayReason records the http2.ErrCode and debug data received with the 111 // GoAway frame. 112 goAwayReason GoAwayReason 113 114 // Fields below are for channelz metric collection. 115 channelzID int64 // channelz unique identification number 116 czData *channelzData 117 118 onGoAway func(GoAwayReason) 119 onClose func() 120} 121 122func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error), addr string) (net.Conn, error) { 123 if fn != nil { 124 return fn(ctx, addr) 125 } 126 return (&net.Dialer{}).DialContext(ctx, "tcp", addr) 127} 128 129func isTemporary(err error) bool { 130 switch err := err.(type) { 131 case interface { 132 Temporary() bool 133 }: 134 return err.Temporary() 135 case interface { 136 Timeout() bool 137 }: 138 // Timeouts may be resolved upon retry, and are thus treated as 139 // temporary. 140 return err.Timeout() 141 } 142 return true 143} 144 145// newHTTP2Client constructs a connected ClientTransport to addr based on HTTP2 146// and starts to receive messages on it. Non-nil error returns if construction 147// fails. 148func newHTTP2Client(connectCtx, ctx context.Context, addr TargetInfo, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (_ *http2Client, err error) { 149 scheme := "http" 150 ctx, cancel := context.WithCancel(ctx) 151 defer func() { 152 if err != nil { 153 cancel() 154 } 155 }() 156 157 conn, err := dial(connectCtx, opts.Dialer, addr.Addr) 158 if err != nil { 159 if opts.FailOnNonTempDialError { 160 return nil, connectionErrorf(isTemporary(err), err, "transport: error while dialing: %v", err) 161 } 162 return nil, connectionErrorf(true, err, "transport: Error while dialing %v", err) 163 } 164 // Any further errors will close the underlying connection 165 defer func(conn net.Conn) { 166 if err != nil { 167 conn.Close() 168 } 169 }(conn) 170 kp := opts.KeepaliveParams 171 // Validate keepalive parameters. 172 if kp.Time == 0 { 173 kp.Time = defaultClientKeepaliveTime 174 } 175 if kp.Timeout == 0 { 176 kp.Timeout = defaultClientKeepaliveTimeout 177 } 178 keepaliveEnabled := false 179 if kp.Time != infinity { 180 if err = syscall.SetTCPUserTimeout(conn, kp.Timeout); err != nil { 181 return nil, connectionErrorf(false, err, "transport: failed to set TCP_USER_TIMEOUT: %v", err) 182 } 183 keepaliveEnabled = true 184 } 185 var ( 186 isSecure bool 187 authInfo credentials.AuthInfo 188 ) 189 transportCreds := opts.TransportCredentials 190 perRPCCreds := opts.PerRPCCredentials 191 192 if b := opts.CredsBundle; b != nil { 193 if t := b.TransportCredentials(); t != nil { 194 transportCreds = t 195 } 196 if t := b.PerRPCCredentials(); t != nil { 197 perRPCCreds = append(perRPCCreds, t) 198 } 199 } 200 if transportCreds != nil { 201 scheme = "https" 202 conn, authInfo, err = transportCreds.ClientHandshake(connectCtx, addr.Authority, conn) 203 if err != nil { 204 return nil, connectionErrorf(isTemporary(err), err, "transport: authentication handshake failed: %v", err) 205 } 206 isSecure = true 207 } 208 dynamicWindow := true 209 icwz := int32(initialWindowSize) 210 if opts.InitialConnWindowSize >= defaultWindowSize { 211 icwz = opts.InitialConnWindowSize 212 dynamicWindow = false 213 } 214 writeBufSize := opts.WriteBufferSize 215 readBufSize := opts.ReadBufferSize 216 maxHeaderListSize := defaultClientMaxHeaderListSize 217 if opts.MaxHeaderListSize != nil { 218 maxHeaderListSize = *opts.MaxHeaderListSize 219 } 220 t := &http2Client{ 221 ctx: ctx, 222 ctxDone: ctx.Done(), // Cache Done chan. 223 cancel: cancel, 224 userAgent: opts.UserAgent, 225 md: addr.Metadata, 226 conn: conn, 227 remoteAddr: conn.RemoteAddr(), 228 localAddr: conn.LocalAddr(), 229 authInfo: authInfo, 230 readerDone: make(chan struct{}), 231 writerDone: make(chan struct{}), 232 goAway: make(chan struct{}), 233 awakenKeepalive: make(chan struct{}, 1), 234 framer: newFramer(conn, writeBufSize, readBufSize, maxHeaderListSize), 235 fc: &trInFlow{limit: uint32(icwz)}, 236 scheme: scheme, 237 activeStreams: make(map[uint32]*Stream), 238 isSecure: isSecure, 239 perRPCCreds: perRPCCreds, 240 kp: kp, 241 statsHandler: opts.StatsHandler, 242 initialWindowSize: initialWindowSize, 243 onPrefaceReceipt: onPrefaceReceipt, 244 nextID: 1, 245 maxConcurrentStreams: defaultMaxStreamsClient, 246 streamQuota: defaultMaxStreamsClient, 247 streamsQuotaAvailable: make(chan struct{}, 1), 248 czData: new(channelzData), 249 onGoAway: onGoAway, 250 onClose: onClose, 251 keepaliveEnabled: keepaliveEnabled, 252 } 253 t.controlBuf = newControlBuffer(t.ctxDone) 254 if opts.InitialWindowSize >= defaultWindowSize { 255 t.initialWindowSize = opts.InitialWindowSize 256 dynamicWindow = false 257 } 258 if dynamicWindow { 259 t.bdpEst = &bdpEstimator{ 260 bdp: initialWindowSize, 261 updateFlowControl: t.updateFlowControl, 262 } 263 } 264 // Make sure awakenKeepalive can't be written upon. 265 // keepalive routine will make it writable, if need be. 266 t.awakenKeepalive <- struct{}{} 267 if t.statsHandler != nil { 268 t.ctx = t.statsHandler.TagConn(t.ctx, &stats.ConnTagInfo{ 269 RemoteAddr: t.remoteAddr, 270 LocalAddr: t.localAddr, 271 }) 272 connBegin := &stats.ConnBegin{ 273 Client: true, 274 } 275 t.statsHandler.HandleConn(t.ctx, connBegin) 276 } 277 if channelz.IsOn() { 278 t.channelzID = channelz.RegisterNormalSocket(t, opts.ChannelzParentID, fmt.Sprintf("%s -> %s", t.localAddr, t.remoteAddr)) 279 } 280 if t.keepaliveEnabled { 281 go t.keepalive() 282 } 283 // Start the reader goroutine for incoming message. Each transport has 284 // a dedicated goroutine which reads HTTP2 frame from network. Then it 285 // dispatches the frame to the corresponding stream entity. 286 go t.reader() 287 288 // Send connection preface to server. 289 n, err := t.conn.Write(clientPreface) 290 if err != nil { 291 t.Close() 292 return nil, connectionErrorf(true, err, "transport: failed to write client preface: %v", err) 293 } 294 if n != len(clientPreface) { 295 t.Close() 296 return nil, connectionErrorf(true, err, "transport: preface mismatch, wrote %d bytes; want %d", n, len(clientPreface)) 297 } 298 var ss []http2.Setting 299 300 if t.initialWindowSize != defaultWindowSize { 301 ss = append(ss, http2.Setting{ 302 ID: http2.SettingInitialWindowSize, 303 Val: uint32(t.initialWindowSize), 304 }) 305 } 306 if opts.MaxHeaderListSize != nil { 307 ss = append(ss, http2.Setting{ 308 ID: http2.SettingMaxHeaderListSize, 309 Val: *opts.MaxHeaderListSize, 310 }) 311 } 312 err = t.framer.fr.WriteSettings(ss...) 313 if err != nil { 314 t.Close() 315 return nil, connectionErrorf(true, err, "transport: failed to write initial settings frame: %v", err) 316 } 317 // Adjust the connection flow control window if needed. 318 if delta := uint32(icwz - defaultWindowSize); delta > 0 { 319 if err := t.framer.fr.WriteWindowUpdate(0, delta); err != nil { 320 t.Close() 321 return nil, connectionErrorf(true, err, "transport: failed to write window update: %v", err) 322 } 323 } 324 325 if err := t.framer.writer.Flush(); err != nil { 326 return nil, err 327 } 328 go func() { 329 t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst) 330 err := t.loopy.run() 331 if err != nil { 332 errorf("transport: loopyWriter.run returning. Err: %v", err) 333 } 334 // If it's a connection error, let reader goroutine handle it 335 // since there might be data in the buffers. 336 if _, ok := err.(net.Error); !ok { 337 t.conn.Close() 338 } 339 close(t.writerDone) 340 }() 341 return t, nil 342} 343 344func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *Stream { 345 // TODO(zhaoq): Handle uint32 overflow of Stream.id. 346 s := &Stream{ 347 done: make(chan struct{}), 348 method: callHdr.Method, 349 sendCompress: callHdr.SendCompress, 350 buf: newRecvBuffer(), 351 headerChan: make(chan struct{}), 352 contentSubtype: callHdr.ContentSubtype, 353 } 354 s.wq = newWriteQuota(defaultWriteQuota, s.done) 355 s.requestRead = func(n int) { 356 t.adjustWindow(s, uint32(n)) 357 } 358 // The client side stream context should have exactly the same life cycle with the user provided context. 359 // That means, s.ctx should be read-only. And s.ctx is done iff ctx is done. 360 // So we use the original context here instead of creating a copy. 361 s.ctx = ctx 362 s.trReader = &transportReader{ 363 reader: &recvBufferReader{ 364 ctx: s.ctx, 365 ctxDone: s.ctx.Done(), 366 recv: s.buf, 367 closeStream: func(err error) { 368 t.CloseStream(s, err) 369 }, 370 }, 371 windowHandler: func(n int) { 372 t.updateWindow(s, uint32(n)) 373 }, 374 } 375 return s 376} 377 378func (t *http2Client) getPeer() *peer.Peer { 379 pr := &peer.Peer{ 380 Addr: t.remoteAddr, 381 } 382 // Attach Auth info if there is any. 383 if t.authInfo != nil { 384 pr.AuthInfo = t.authInfo 385 } 386 return pr 387} 388 389func (t *http2Client) createHeaderFields(ctx context.Context, callHdr *CallHdr) ([]hpack.HeaderField, error) { 390 aud := t.createAudience(callHdr) 391 authData, err := t.getTrAuthData(ctx, aud) 392 if err != nil { 393 return nil, err 394 } 395 callAuthData, err := t.getCallAuthData(ctx, aud, callHdr) 396 if err != nil { 397 return nil, err 398 } 399 // TODO(mmukhi): Benchmark if the performance gets better if count the metadata and other header fields 400 // first and create a slice of that exact size. 401 // Make the slice of certain predictable size to reduce allocations made by append. 402 hfLen := 7 // :method, :scheme, :path, :authority, content-type, user-agent, te 403 hfLen += len(authData) + len(callAuthData) 404 headerFields := make([]hpack.HeaderField, 0, hfLen) 405 headerFields = append(headerFields, hpack.HeaderField{Name: ":method", Value: "POST"}) 406 headerFields = append(headerFields, hpack.HeaderField{Name: ":scheme", Value: t.scheme}) 407 headerFields = append(headerFields, hpack.HeaderField{Name: ":path", Value: callHdr.Method}) 408 headerFields = append(headerFields, hpack.HeaderField{Name: ":authority", Value: callHdr.Host}) 409 headerFields = append(headerFields, hpack.HeaderField{Name: "content-type", Value: contentType(callHdr.ContentSubtype)}) 410 headerFields = append(headerFields, hpack.HeaderField{Name: "user-agent", Value: t.userAgent}) 411 headerFields = append(headerFields, hpack.HeaderField{Name: "te", Value: "trailers"}) 412 if callHdr.PreviousAttempts > 0 { 413 headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-previous-rpc-attempts", Value: strconv.Itoa(callHdr.PreviousAttempts)}) 414 } 415 416 if callHdr.SendCompress != "" { 417 headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-encoding", Value: callHdr.SendCompress}) 418 } 419 if dl, ok := ctx.Deadline(); ok { 420 // Send out timeout regardless its value. The server can detect timeout context by itself. 421 // TODO(mmukhi): Perhaps this field should be updated when actually writing out to the wire. 422 timeout := time.Until(dl) 423 headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-timeout", Value: encodeTimeout(timeout)}) 424 } 425 for k, v := range authData { 426 headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)}) 427 } 428 for k, v := range callAuthData { 429 headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)}) 430 } 431 if b := stats.OutgoingTags(ctx); b != nil { 432 headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-tags-bin", Value: encodeBinHeader(b)}) 433 } 434 if b := stats.OutgoingTrace(ctx); b != nil { 435 headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-trace-bin", Value: encodeBinHeader(b)}) 436 } 437 438 if md, added, ok := metadata.FromOutgoingContextRaw(ctx); ok { 439 var k string 440 for _, vv := range added { 441 for i, v := range vv { 442 if i%2 == 0 { 443 k = v 444 continue 445 } 446 // HTTP doesn't allow you to set pseudoheaders after non pseudoheaders were set. 447 if isReservedHeader(k) { 448 continue 449 } 450 headerFields = append(headerFields, hpack.HeaderField{Name: strings.ToLower(k), Value: encodeMetadataHeader(k, v)}) 451 } 452 } 453 for k, vv := range md { 454 // HTTP doesn't allow you to set pseudoheaders after non pseudoheaders were set. 455 if isReservedHeader(k) { 456 continue 457 } 458 for _, v := range vv { 459 headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)}) 460 } 461 } 462 } 463 if md, ok := t.md.(*metadata.MD); ok { 464 for k, vv := range *md { 465 if isReservedHeader(k) { 466 continue 467 } 468 for _, v := range vv { 469 headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)}) 470 } 471 } 472 } 473 return headerFields, nil 474} 475 476func (t *http2Client) createAudience(callHdr *CallHdr) string { 477 // Create an audience string only if needed. 478 if len(t.perRPCCreds) == 0 && callHdr.Creds == nil { 479 return "" 480 } 481 // Construct URI required to get auth request metadata. 482 // Omit port if it is the default one. 483 host := strings.TrimSuffix(callHdr.Host, ":443") 484 pos := strings.LastIndex(callHdr.Method, "/") 485 if pos == -1 { 486 pos = len(callHdr.Method) 487 } 488 return "https://" + host + callHdr.Method[:pos] 489} 490 491func (t *http2Client) getTrAuthData(ctx context.Context, audience string) (map[string]string, error) { 492 authData := map[string]string{} 493 for _, c := range t.perRPCCreds { 494 data, err := c.GetRequestMetadata(ctx, audience) 495 if err != nil { 496 if _, ok := status.FromError(err); ok { 497 return nil, err 498 } 499 500 return nil, status.Errorf(codes.Unauthenticated, "transport: %v", err) 501 } 502 for k, v := range data { 503 // Capital header names are illegal in HTTP/2. 504 k = strings.ToLower(k) 505 authData[k] = v 506 } 507 } 508 return authData, nil 509} 510 511func (t *http2Client) getCallAuthData(ctx context.Context, audience string, callHdr *CallHdr) (map[string]string, error) { 512 callAuthData := map[string]string{} 513 // Check if credentials.PerRPCCredentials were provided via call options. 514 // Note: if these credentials are provided both via dial options and call 515 // options, then both sets of credentials will be applied. 516 if callCreds := callHdr.Creds; callCreds != nil { 517 if !t.isSecure && callCreds.RequireTransportSecurity() { 518 return nil, status.Error(codes.Unauthenticated, "transport: cannot send secure credentials on an insecure connection") 519 } 520 data, err := callCreds.GetRequestMetadata(ctx, audience) 521 if err != nil { 522 return nil, status.Errorf(codes.Internal, "transport: %v", err) 523 } 524 for k, v := range data { 525 // Capital header names are illegal in HTTP/2 526 k = strings.ToLower(k) 527 callAuthData[k] = v 528 } 529 } 530 return callAuthData, nil 531} 532 533// NewStream creates a stream and registers it into the transport as "active" 534// streams. 535func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) { 536 ctx = peer.NewContext(ctx, t.getPeer()) 537 headerFields, err := t.createHeaderFields(ctx, callHdr) 538 if err != nil { 539 return nil, err 540 } 541 s := t.newStream(ctx, callHdr) 542 cleanup := func(err error) { 543 if s.swapState(streamDone) == streamDone { 544 // If it was already done, return. 545 return 546 } 547 // The stream was unprocessed by the server. 548 atomic.StoreUint32(&s.unprocessed, 1) 549 s.write(recvMsg{err: err}) 550 close(s.done) 551 // If headerChan isn't closed, then close it. 552 if atomic.SwapUint32(&s.headerDone, 1) == 0 { 553 close(s.headerChan) 554 } 555 556 } 557 hdr := &headerFrame{ 558 hf: headerFields, 559 endStream: false, 560 initStream: func(id uint32) (bool, error) { 561 t.mu.Lock() 562 if state := t.state; state != reachable { 563 t.mu.Unlock() 564 // Do a quick cleanup. 565 err := error(errStreamDrain) 566 if state == closing { 567 err = ErrConnClosing 568 } 569 cleanup(err) 570 return false, err 571 } 572 t.activeStreams[id] = s 573 if channelz.IsOn() { 574 atomic.AddInt64(&t.czData.streamsStarted, 1) 575 atomic.StoreInt64(&t.czData.lastStreamCreatedTime, time.Now().UnixNano()) 576 } 577 var sendPing bool 578 // If the number of active streams change from 0 to 1, then check if keepalive 579 // has gone dormant. If so, wake it up. 580 if len(t.activeStreams) == 1 && t.keepaliveEnabled { 581 select { 582 case t.awakenKeepalive <- struct{}{}: 583 sendPing = true 584 // Fill the awakenKeepalive channel again as this channel must be 585 // kept non-writable except at the point that the keepalive() 586 // goroutine is waiting either to be awaken or shutdown. 587 t.awakenKeepalive <- struct{}{} 588 default: 589 } 590 } 591 t.mu.Unlock() 592 return sendPing, nil 593 }, 594 onOrphaned: cleanup, 595 wq: s.wq, 596 } 597 firstTry := true 598 var ch chan struct{} 599 checkForStreamQuota := func(it interface{}) bool { 600 if t.streamQuota <= 0 { // Can go negative if server decreases it. 601 if firstTry { 602 t.waitingStreams++ 603 } 604 ch = t.streamsQuotaAvailable 605 return false 606 } 607 if !firstTry { 608 t.waitingStreams-- 609 } 610 t.streamQuota-- 611 h := it.(*headerFrame) 612 h.streamID = t.nextID 613 t.nextID += 2 614 s.id = h.streamID 615 s.fc = &inFlow{limit: uint32(t.initialWindowSize)} 616 if t.streamQuota > 0 && t.waitingStreams > 0 { 617 select { 618 case t.streamsQuotaAvailable <- struct{}{}: 619 default: 620 } 621 } 622 return true 623 } 624 var hdrListSizeErr error 625 checkForHeaderListSize := func(it interface{}) bool { 626 if t.maxSendHeaderListSize == nil { 627 return true 628 } 629 hdrFrame := it.(*headerFrame) 630 var sz int64 631 for _, f := range hdrFrame.hf { 632 if sz += int64(f.Size()); sz > int64(*t.maxSendHeaderListSize) { 633 hdrListSizeErr = status.Errorf(codes.Internal, "header list size to send violates the maximum size (%d bytes) set by server", *t.maxSendHeaderListSize) 634 return false 635 } 636 } 637 return true 638 } 639 for { 640 success, err := t.controlBuf.executeAndPut(func(it interface{}) bool { 641 if !checkForStreamQuota(it) { 642 return false 643 } 644 if !checkForHeaderListSize(it) { 645 return false 646 } 647 return true 648 }, hdr) 649 if err != nil { 650 return nil, err 651 } 652 if success { 653 break 654 } 655 if hdrListSizeErr != nil { 656 return nil, hdrListSizeErr 657 } 658 firstTry = false 659 select { 660 case <-ch: 661 case <-s.ctx.Done(): 662 return nil, ContextErr(s.ctx.Err()) 663 case <-t.goAway: 664 return nil, errStreamDrain 665 case <-t.ctx.Done(): 666 return nil, ErrConnClosing 667 } 668 } 669 if t.statsHandler != nil { 670 outHeader := &stats.OutHeader{ 671 Client: true, 672 FullMethod: callHdr.Method, 673 RemoteAddr: t.remoteAddr, 674 LocalAddr: t.localAddr, 675 Compression: callHdr.SendCompress, 676 } 677 t.statsHandler.HandleRPC(s.ctx, outHeader) 678 } 679 return s, nil 680} 681 682// CloseStream clears the footprint of a stream when the stream is not needed any more. 683// This must not be executed in reader's goroutine. 684func (t *http2Client) CloseStream(s *Stream, err error) { 685 var ( 686 rst bool 687 rstCode http2.ErrCode 688 ) 689 if err != nil { 690 rst = true 691 rstCode = http2.ErrCodeCancel 692 } 693 t.closeStream(s, err, rst, rstCode, status.Convert(err), nil, false) 694} 695 696func (t *http2Client) closeStream(s *Stream, err error, rst bool, rstCode http2.ErrCode, st *status.Status, mdata map[string][]string, eosReceived bool) { 697 // Set stream status to done. 698 if s.swapState(streamDone) == streamDone { 699 // If it was already done, return. If multiple closeStream calls 700 // happen simultaneously, wait for the first to finish. 701 <-s.done 702 return 703 } 704 // status and trailers can be updated here without any synchronization because the stream goroutine will 705 // only read it after it sees an io.EOF error from read or write and we'll write those errors 706 // only after updating this. 707 s.status = st 708 if len(mdata) > 0 { 709 s.trailer = mdata 710 } 711 if err != nil { 712 // This will unblock reads eventually. 713 s.write(recvMsg{err: err}) 714 } 715 // If headerChan isn't closed, then close it. 716 if atomic.SwapUint32(&s.headerDone, 1) == 0 { 717 s.noHeaders = true 718 close(s.headerChan) 719 } 720 cleanup := &cleanupStream{ 721 streamID: s.id, 722 onWrite: func() { 723 t.mu.Lock() 724 if t.activeStreams != nil { 725 delete(t.activeStreams, s.id) 726 } 727 t.mu.Unlock() 728 if channelz.IsOn() { 729 if eosReceived { 730 atomic.AddInt64(&t.czData.streamsSucceeded, 1) 731 } else { 732 atomic.AddInt64(&t.czData.streamsFailed, 1) 733 } 734 } 735 }, 736 rst: rst, 737 rstCode: rstCode, 738 } 739 addBackStreamQuota := func(interface{}) bool { 740 t.streamQuota++ 741 if t.streamQuota > 0 && t.waitingStreams > 0 { 742 select { 743 case t.streamsQuotaAvailable <- struct{}{}: 744 default: 745 } 746 } 747 return true 748 } 749 t.controlBuf.executeAndPut(addBackStreamQuota, cleanup) 750 // This will unblock write. 751 close(s.done) 752} 753 754// Close kicks off the shutdown process of the transport. This should be called 755// only once on a transport. Once it is called, the transport should not be 756// accessed any more. 757// 758// This method blocks until the addrConn that initiated this transport is 759// re-connected. This happens because t.onClose() begins reconnect logic at the 760// addrConn level and blocks until the addrConn is successfully connected. 761func (t *http2Client) Close() error { 762 t.mu.Lock() 763 // Make sure we only Close once. 764 if t.state == closing { 765 t.mu.Unlock() 766 return nil 767 } 768 t.state = closing 769 streams := t.activeStreams 770 t.activeStreams = nil 771 t.mu.Unlock() 772 t.controlBuf.finish() 773 t.cancel() 774 err := t.conn.Close() 775 if channelz.IsOn() { 776 channelz.RemoveEntry(t.channelzID) 777 } 778 // Notify all active streams. 779 for _, s := range streams { 780 t.closeStream(s, ErrConnClosing, false, http2.ErrCodeNo, status.New(codes.Unavailable, ErrConnClosing.Desc), nil, false) 781 } 782 if t.statsHandler != nil { 783 connEnd := &stats.ConnEnd{ 784 Client: true, 785 } 786 t.statsHandler.HandleConn(t.ctx, connEnd) 787 } 788 t.onClose() 789 return err 790} 791 792// GracefulClose sets the state to draining, which prevents new streams from 793// being created and causes the transport to be closed when the last active 794// stream is closed. If there are no active streams, the transport is closed 795// immediately. This does nothing if the transport is already draining or 796// closing. 797func (t *http2Client) GracefulClose() error { 798 t.mu.Lock() 799 // Make sure we move to draining only from active. 800 if t.state == draining || t.state == closing { 801 t.mu.Unlock() 802 return nil 803 } 804 t.state = draining 805 active := len(t.activeStreams) 806 t.mu.Unlock() 807 if active == 0 { 808 return t.Close() 809 } 810 t.controlBuf.put(&incomingGoAway{}) 811 return nil 812} 813 814// Write formats the data into HTTP2 data frame(s) and sends it out. The caller 815// should proceed only if Write returns nil. 816func (t *http2Client) Write(s *Stream, hdr []byte, data []byte, opts *Options) error { 817 if opts.Last { 818 // If it's the last message, update stream state. 819 if !s.compareAndSwapState(streamActive, streamWriteDone) { 820 return errStreamDone 821 } 822 } else if s.getState() != streamActive { 823 return errStreamDone 824 } 825 df := &dataFrame{ 826 streamID: s.id, 827 endStream: opts.Last, 828 } 829 if hdr != nil || data != nil { // If it's not an empty data frame. 830 // Add some data to grpc message header so that we can equally 831 // distribute bytes across frames. 832 emptyLen := http2MaxFrameLen - len(hdr) 833 if emptyLen > len(data) { 834 emptyLen = len(data) 835 } 836 hdr = append(hdr, data[:emptyLen]...) 837 data = data[emptyLen:] 838 df.h, df.d = hdr, data 839 // TODO(mmukhi): The above logic in this if can be moved to loopyWriter's data handler. 840 if err := s.wq.get(int32(len(hdr) + len(data))); err != nil { 841 return err 842 } 843 } 844 return t.controlBuf.put(df) 845} 846 847func (t *http2Client) getStream(f http2.Frame) (*Stream, bool) { 848 t.mu.Lock() 849 defer t.mu.Unlock() 850 s, ok := t.activeStreams[f.Header().StreamID] 851 return s, ok 852} 853 854// adjustWindow sends out extra window update over the initial window size 855// of stream if the application is requesting data larger in size than 856// the window. 857func (t *http2Client) adjustWindow(s *Stream, n uint32) { 858 if w := s.fc.maybeAdjust(n); w > 0 { 859 t.controlBuf.put(&outgoingWindowUpdate{streamID: s.id, increment: w}) 860 } 861} 862 863// updateWindow adjusts the inbound quota for the stream. 864// Window updates will be sent out when the cumulative quota 865// exceeds the corresponding threshold. 866func (t *http2Client) updateWindow(s *Stream, n uint32) { 867 if w := s.fc.onRead(n); w > 0 { 868 t.controlBuf.put(&outgoingWindowUpdate{streamID: s.id, increment: w}) 869 } 870} 871 872// updateFlowControl updates the incoming flow control windows 873// for the transport and the stream based on the current bdp 874// estimation. 875func (t *http2Client) updateFlowControl(n uint32) { 876 t.mu.Lock() 877 for _, s := range t.activeStreams { 878 s.fc.newLimit(n) 879 } 880 t.mu.Unlock() 881 updateIWS := func(interface{}) bool { 882 t.initialWindowSize = int32(n) 883 return true 884 } 885 t.controlBuf.executeAndPut(updateIWS, &outgoingWindowUpdate{streamID: 0, increment: t.fc.newLimit(n)}) 886 t.controlBuf.put(&outgoingSettings{ 887 ss: []http2.Setting{ 888 { 889 ID: http2.SettingInitialWindowSize, 890 Val: n, 891 }, 892 }, 893 }) 894} 895 896func (t *http2Client) handleData(f *http2.DataFrame) { 897 size := f.Header().Length 898 var sendBDPPing bool 899 if t.bdpEst != nil { 900 sendBDPPing = t.bdpEst.add(size) 901 } 902 // Decouple connection's flow control from application's read. 903 // An update on connection's flow control should not depend on 904 // whether user application has read the data or not. Such a 905 // restriction is already imposed on the stream's flow control, 906 // and therefore the sender will be blocked anyways. 907 // Decoupling the connection flow control will prevent other 908 // active(fast) streams from starving in presence of slow or 909 // inactive streams. 910 // 911 if w := t.fc.onData(size); w > 0 { 912 t.controlBuf.put(&outgoingWindowUpdate{ 913 streamID: 0, 914 increment: w, 915 }) 916 } 917 if sendBDPPing { 918 // Avoid excessive ping detection (e.g. in an L7 proxy) 919 // by sending a window update prior to the BDP ping. 920 921 if w := t.fc.reset(); w > 0 { 922 t.controlBuf.put(&outgoingWindowUpdate{ 923 streamID: 0, 924 increment: w, 925 }) 926 } 927 928 t.controlBuf.put(bdpPing) 929 } 930 // Select the right stream to dispatch. 931 s, ok := t.getStream(f) 932 if !ok { 933 return 934 } 935 if size > 0 { 936 if err := s.fc.onData(size); err != nil { 937 t.closeStream(s, io.EOF, true, http2.ErrCodeFlowControl, status.New(codes.Internal, err.Error()), nil, false) 938 return 939 } 940 if f.Header().Flags.Has(http2.FlagDataPadded) { 941 if w := s.fc.onRead(size - uint32(len(f.Data()))); w > 0 { 942 t.controlBuf.put(&outgoingWindowUpdate{s.id, w}) 943 } 944 } 945 // TODO(bradfitz, zhaoq): A copy is required here because there is no 946 // guarantee f.Data() is consumed before the arrival of next frame. 947 // Can this copy be eliminated? 948 if len(f.Data()) > 0 { 949 data := make([]byte, len(f.Data())) 950 copy(data, f.Data()) 951 s.write(recvMsg{data: data}) 952 } 953 } 954 // The server has closed the stream without sending trailers. Record that 955 // the read direction is closed, and set the status appropriately. 956 if f.FrameHeader.Flags.Has(http2.FlagDataEndStream) { 957 t.closeStream(s, io.EOF, false, http2.ErrCodeNo, status.New(codes.Internal, "server closed the stream without sending trailers"), nil, true) 958 } 959} 960 961func (t *http2Client) handleRSTStream(f *http2.RSTStreamFrame) { 962 s, ok := t.getStream(f) 963 if !ok { 964 return 965 } 966 if f.ErrCode == http2.ErrCodeRefusedStream { 967 // The stream was unprocessed by the server. 968 atomic.StoreUint32(&s.unprocessed, 1) 969 } 970 statusCode, ok := http2ErrConvTab[f.ErrCode] 971 if !ok { 972 warningf("transport: http2Client.handleRSTStream found no mapped gRPC status for the received http2 error %v", f.ErrCode) 973 statusCode = codes.Unknown 974 } 975 if statusCode == codes.Canceled { 976 // Our deadline was already exceeded, and that was likely the cause of 977 // this cancelation. Alter the status code accordingly. 978 if d, ok := s.ctx.Deadline(); ok && d.After(time.Now()) { 979 statusCode = codes.DeadlineExceeded 980 } 981 } 982 t.closeStream(s, io.EOF, false, http2.ErrCodeNo, status.Newf(statusCode, "stream terminated by RST_STREAM with error code: %v", f.ErrCode), nil, false) 983} 984 985func (t *http2Client) handleSettings(f *http2.SettingsFrame, isFirst bool) { 986 if f.IsAck() { 987 return 988 } 989 var maxStreams *uint32 990 var ss []http2.Setting 991 var updateFuncs []func() 992 f.ForeachSetting(func(s http2.Setting) error { 993 switch s.ID { 994 case http2.SettingMaxConcurrentStreams: 995 maxStreams = new(uint32) 996 *maxStreams = s.Val 997 case http2.SettingMaxHeaderListSize: 998 updateFuncs = append(updateFuncs, func() { 999 t.maxSendHeaderListSize = new(uint32) 1000 *t.maxSendHeaderListSize = s.Val 1001 }) 1002 default: 1003 ss = append(ss, s) 1004 } 1005 return nil 1006 }) 1007 if isFirst && maxStreams == nil { 1008 maxStreams = new(uint32) 1009 *maxStreams = math.MaxUint32 1010 } 1011 sf := &incomingSettings{ 1012 ss: ss, 1013 } 1014 if maxStreams != nil { 1015 updateStreamQuota := func() { 1016 delta := int64(*maxStreams) - int64(t.maxConcurrentStreams) 1017 t.maxConcurrentStreams = *maxStreams 1018 t.streamQuota += delta 1019 if delta > 0 && t.waitingStreams > 0 { 1020 close(t.streamsQuotaAvailable) // wake all of them up. 1021 t.streamsQuotaAvailable = make(chan struct{}, 1) 1022 } 1023 } 1024 updateFuncs = append(updateFuncs, updateStreamQuota) 1025 } 1026 t.controlBuf.executeAndPut(func(interface{}) bool { 1027 for _, f := range updateFuncs { 1028 f() 1029 } 1030 return true 1031 }, sf) 1032} 1033 1034func (t *http2Client) handlePing(f *http2.PingFrame) { 1035 if f.IsAck() { 1036 // Maybe it's a BDP ping. 1037 if t.bdpEst != nil { 1038 t.bdpEst.calculate(f.Data) 1039 } 1040 return 1041 } 1042 pingAck := &ping{ack: true} 1043 copy(pingAck.data[:], f.Data[:]) 1044 t.controlBuf.put(pingAck) 1045} 1046 1047func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) { 1048 t.mu.Lock() 1049 if t.state == closing { 1050 t.mu.Unlock() 1051 return 1052 } 1053 if f.ErrCode == http2.ErrCodeEnhanceYourCalm { 1054 infof("Client received GoAway with http2.ErrCodeEnhanceYourCalm.") 1055 } 1056 id := f.LastStreamID 1057 if id > 0 && id%2 != 1 { 1058 t.mu.Unlock() 1059 t.Close() 1060 return 1061 } 1062 // A client can receive multiple GoAways from the server (see 1063 // https://github.com/grpc/grpc-go/issues/1387). The idea is that the first 1064 // GoAway will be sent with an ID of MaxInt32 and the second GoAway will be 1065 // sent after an RTT delay with the ID of the last stream the server will 1066 // process. 1067 // 1068 // Therefore, when we get the first GoAway we don't necessarily close any 1069 // streams. While in case of second GoAway we close all streams created after 1070 // the GoAwayId. This way streams that were in-flight while the GoAway from 1071 // server was being sent don't get killed. 1072 select { 1073 case <-t.goAway: // t.goAway has been closed (i.e.,multiple GoAways). 1074 // If there are multiple GoAways the first one should always have an ID greater than the following ones. 1075 if id > t.prevGoAwayID { 1076 t.mu.Unlock() 1077 t.Close() 1078 return 1079 } 1080 default: 1081 t.setGoAwayReason(f) 1082 close(t.goAway) 1083 t.state = draining 1084 t.controlBuf.put(&incomingGoAway{}) 1085 1086 // This has to be a new goroutine because we're still using the current goroutine to read in the transport. 1087 t.onGoAway(t.goAwayReason) 1088 } 1089 // All streams with IDs greater than the GoAwayId 1090 // and smaller than the previous GoAway ID should be killed. 1091 upperLimit := t.prevGoAwayID 1092 if upperLimit == 0 { // This is the first GoAway Frame. 1093 upperLimit = math.MaxUint32 // Kill all streams after the GoAway ID. 1094 } 1095 for streamID, stream := range t.activeStreams { 1096 if streamID > id && streamID <= upperLimit { 1097 // The stream was unprocessed by the server. 1098 atomic.StoreUint32(&stream.unprocessed, 1) 1099 t.closeStream(stream, errStreamDrain, false, http2.ErrCodeNo, statusGoAway, nil, false) 1100 } 1101 } 1102 t.prevGoAwayID = id 1103 active := len(t.activeStreams) 1104 t.mu.Unlock() 1105 if active == 0 { 1106 t.Close() 1107 } 1108} 1109 1110// setGoAwayReason sets the value of t.goAwayReason based 1111// on the GoAway frame received. 1112// It expects a lock on transport's mutext to be held by 1113// the caller. 1114func (t *http2Client) setGoAwayReason(f *http2.GoAwayFrame) { 1115 t.goAwayReason = GoAwayNoReason 1116 switch f.ErrCode { 1117 case http2.ErrCodeEnhanceYourCalm: 1118 if string(f.DebugData()) == "too_many_pings" { 1119 t.goAwayReason = GoAwayTooManyPings 1120 } 1121 } 1122} 1123 1124func (t *http2Client) GetGoAwayReason() GoAwayReason { 1125 t.mu.Lock() 1126 defer t.mu.Unlock() 1127 return t.goAwayReason 1128} 1129 1130func (t *http2Client) handleWindowUpdate(f *http2.WindowUpdateFrame) { 1131 t.controlBuf.put(&incomingWindowUpdate{ 1132 streamID: f.Header().StreamID, 1133 increment: f.Increment, 1134 }) 1135} 1136 1137// operateHeaders takes action on the decoded headers. 1138func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) { 1139 s, ok := t.getStream(frame) 1140 if !ok { 1141 return 1142 } 1143 endStream := frame.StreamEnded() 1144 atomic.StoreUint32(&s.bytesReceived, 1) 1145 initialHeader := atomic.SwapUint32(&s.headerDone, 1) == 0 1146 1147 if !initialHeader && !endStream { 1148 // As specified by RFC 7540, a HEADERS frame (and associated CONTINUATION frames) can only appear 1149 // at the start or end of a stream. Therefore, second HEADERS frame must have EOS bit set. 1150 st := status.New(codes.Internal, "a HEADERS frame cannot appear in the middle of a stream") 1151 t.closeStream(s, st.Err(), true, http2.ErrCodeProtocol, st, nil, false) 1152 return 1153 } 1154 1155 state := &decodeState{} 1156 // Initialize isGRPC value to be !initialHeader, since if a gRPC ResponseHeader has been received 1157 // which indicates peer speaking gRPC, we are in gRPC mode. 1158 state.data.isGRPC = !initialHeader 1159 if err := state.decodeHeader(frame); err != nil { 1160 t.closeStream(s, err, true, http2.ErrCodeProtocol, status.Convert(err), nil, endStream) 1161 return 1162 } 1163 1164 var isHeader bool 1165 defer func() { 1166 if t.statsHandler != nil { 1167 if isHeader { 1168 inHeader := &stats.InHeader{ 1169 Client: true, 1170 WireLength: int(frame.Header().Length), 1171 } 1172 t.statsHandler.HandleRPC(s.ctx, inHeader) 1173 } else { 1174 inTrailer := &stats.InTrailer{ 1175 Client: true, 1176 WireLength: int(frame.Header().Length), 1177 } 1178 t.statsHandler.HandleRPC(s.ctx, inTrailer) 1179 } 1180 } 1181 }() 1182 1183 // If headers haven't been received yet. 1184 if initialHeader { 1185 if !endStream { 1186 // Headers frame is ResponseHeader. 1187 isHeader = true 1188 // These values can be set without any synchronization because 1189 // stream goroutine will read it only after seeing a closed 1190 // headerChan which we'll close after setting this. 1191 s.recvCompress = state.data.encoding 1192 if len(state.data.mdata) > 0 { 1193 s.header = state.data.mdata 1194 } 1195 close(s.headerChan) 1196 return 1197 } 1198 // Headers frame is Trailers-only. 1199 s.noHeaders = true 1200 close(s.headerChan) 1201 } 1202 1203 // if client received END_STREAM from server while stream was still active, send RST_STREAM 1204 rst := s.getState() == streamActive 1205 t.closeStream(s, io.EOF, rst, http2.ErrCodeNo, state.status(), state.data.mdata, true) 1206} 1207 1208// reader runs as a separate goroutine in charge of reading data from network 1209// connection. 1210// 1211// TODO(zhaoq): currently one reader per transport. Investigate whether this is 1212// optimal. 1213// TODO(zhaoq): Check the validity of the incoming frame sequence. 1214func (t *http2Client) reader() { 1215 defer close(t.readerDone) 1216 // Check the validity of server preface. 1217 frame, err := t.framer.fr.ReadFrame() 1218 if err != nil { 1219 t.Close() // this kicks off resetTransport, so must be last before return 1220 return 1221 } 1222 t.conn.SetReadDeadline(time.Time{}) // reset deadline once we get the settings frame (we didn't time out, yay!) 1223 if t.keepaliveEnabled { 1224 atomic.CompareAndSwapUint32(&t.activity, 0, 1) 1225 } 1226 sf, ok := frame.(*http2.SettingsFrame) 1227 if !ok { 1228 t.Close() // this kicks off resetTransport, so must be last before return 1229 return 1230 } 1231 t.onPrefaceReceipt() 1232 t.handleSettings(sf, true) 1233 1234 // loop to keep reading incoming messages on this transport. 1235 for { 1236 frame, err := t.framer.fr.ReadFrame() 1237 if t.keepaliveEnabled { 1238 atomic.CompareAndSwapUint32(&t.activity, 0, 1) 1239 } 1240 if err != nil { 1241 // Abort an active stream if the http2.Framer returns a 1242 // http2.StreamError. This can happen only if the server's response 1243 // is malformed http2. 1244 if se, ok := err.(http2.StreamError); ok { 1245 t.mu.Lock() 1246 s := t.activeStreams[se.StreamID] 1247 t.mu.Unlock() 1248 if s != nil { 1249 // use error detail to provide better err message 1250 code := http2ErrConvTab[se.Code] 1251 msg := t.framer.fr.ErrorDetail().Error() 1252 t.closeStream(s, status.Error(code, msg), true, http2.ErrCodeProtocol, status.New(code, msg), nil, false) 1253 } 1254 continue 1255 } else { 1256 // Transport error. 1257 t.Close() 1258 return 1259 } 1260 } 1261 switch frame := frame.(type) { 1262 case *http2.MetaHeadersFrame: 1263 t.operateHeaders(frame) 1264 case *http2.DataFrame: 1265 t.handleData(frame) 1266 case *http2.RSTStreamFrame: 1267 t.handleRSTStream(frame) 1268 case *http2.SettingsFrame: 1269 t.handleSettings(frame, false) 1270 case *http2.PingFrame: 1271 t.handlePing(frame) 1272 case *http2.GoAwayFrame: 1273 t.handleGoAway(frame) 1274 case *http2.WindowUpdateFrame: 1275 t.handleWindowUpdate(frame) 1276 default: 1277 errorf("transport: http2Client.reader got unhandled frame type %v.", frame) 1278 } 1279 } 1280} 1281 1282// keepalive running in a separate goroutune makes sure the connection is alive by sending pings. 1283func (t *http2Client) keepalive() { 1284 p := &ping{data: [8]byte{}} 1285 timer := time.NewTimer(t.kp.Time) 1286 for { 1287 select { 1288 case <-timer.C: 1289 if atomic.CompareAndSwapUint32(&t.activity, 1, 0) { 1290 timer.Reset(t.kp.Time) 1291 continue 1292 } 1293 // Check if keepalive should go dormant. 1294 t.mu.Lock() 1295 if len(t.activeStreams) < 1 && !t.kp.PermitWithoutStream { 1296 // Make awakenKeepalive writable. 1297 <-t.awakenKeepalive 1298 t.mu.Unlock() 1299 select { 1300 case <-t.awakenKeepalive: 1301 // If the control gets here a ping has been sent 1302 // need to reset the timer with keepalive.Timeout. 1303 case <-t.ctx.Done(): 1304 return 1305 } 1306 } else { 1307 t.mu.Unlock() 1308 if channelz.IsOn() { 1309 atomic.AddInt64(&t.czData.kpCount, 1) 1310 } 1311 // Send ping. 1312 t.controlBuf.put(p) 1313 } 1314 1315 // By the time control gets here a ping has been sent one way or the other. 1316 timer.Reset(t.kp.Timeout) 1317 select { 1318 case <-timer.C: 1319 if atomic.CompareAndSwapUint32(&t.activity, 1, 0) { 1320 timer.Reset(t.kp.Time) 1321 continue 1322 } 1323 t.Close() 1324 return 1325 case <-t.ctx.Done(): 1326 if !timer.Stop() { 1327 <-timer.C 1328 } 1329 return 1330 } 1331 case <-t.ctx.Done(): 1332 if !timer.Stop() { 1333 <-timer.C 1334 } 1335 return 1336 } 1337 } 1338} 1339 1340func (t *http2Client) Error() <-chan struct{} { 1341 return t.ctx.Done() 1342} 1343 1344func (t *http2Client) GoAway() <-chan struct{} { 1345 return t.goAway 1346} 1347 1348func (t *http2Client) ChannelzMetric() *channelz.SocketInternalMetric { 1349 s := channelz.SocketInternalMetric{ 1350 StreamsStarted: atomic.LoadInt64(&t.czData.streamsStarted), 1351 StreamsSucceeded: atomic.LoadInt64(&t.czData.streamsSucceeded), 1352 StreamsFailed: atomic.LoadInt64(&t.czData.streamsFailed), 1353 MessagesSent: atomic.LoadInt64(&t.czData.msgSent), 1354 MessagesReceived: atomic.LoadInt64(&t.czData.msgRecv), 1355 KeepAlivesSent: atomic.LoadInt64(&t.czData.kpCount), 1356 LastLocalStreamCreatedTimestamp: time.Unix(0, atomic.LoadInt64(&t.czData.lastStreamCreatedTime)), 1357 LastMessageSentTimestamp: time.Unix(0, atomic.LoadInt64(&t.czData.lastMsgSentTime)), 1358 LastMessageReceivedTimestamp: time.Unix(0, atomic.LoadInt64(&t.czData.lastMsgRecvTime)), 1359 LocalFlowControlWindow: int64(t.fc.getSize()), 1360 SocketOptions: channelz.GetSocketOption(t.conn), 1361 LocalAddr: t.localAddr, 1362 RemoteAddr: t.remoteAddr, 1363 // RemoteName : 1364 } 1365 if au, ok := t.authInfo.(credentials.ChannelzSecurityInfo); ok { 1366 s.Security = au.GetSecurityValue() 1367 } 1368 s.RemoteFlowControlWindow = t.getOutFlowWindow() 1369 return &s 1370} 1371 1372func (t *http2Client) RemoteAddr() net.Addr { return t.remoteAddr } 1373 1374func (t *http2Client) IncrMsgSent() { 1375 atomic.AddInt64(&t.czData.msgSent, 1) 1376 atomic.StoreInt64(&t.czData.lastMsgSentTime, time.Now().UnixNano()) 1377} 1378 1379func (t *http2Client) IncrMsgRecv() { 1380 atomic.AddInt64(&t.czData.msgRecv, 1) 1381 atomic.StoreInt64(&t.czData.lastMsgRecvTime, time.Now().UnixNano()) 1382} 1383 1384func (t *http2Client) getOutFlowWindow() int64 { 1385 resp := make(chan uint32, 1) 1386 timer := time.NewTimer(time.Second) 1387 defer timer.Stop() 1388 t.controlBuf.put(&outFlowControlSizeRequest{resp}) 1389 select { 1390 case sz := <-resp: 1391 return int64(sz) 1392 case <-t.ctxDone: 1393 return -1 1394 case <-timer.C: 1395 return -2 1396 } 1397} 1398