go gRPC 客户端内存暴涨原因分析

快来打我* 2022-10-24 11:52 354阅读 0赞

###  ###

创建一个 gRPC 客户端连接，会创建的几个协程：

1）transport.loopyWriter.run 往服务端发送数据协程，流控时会阻塞，结果是数据堆积，内存上涨

2）transport.http2Client.reader 接收服务端数据协程，并会调用 t.controlBuf.throttle() 执行流控

### 现象描述： ###

客户端到服务端单个连接，压测时内存快速增长，直到 OOM 挂掉。在 OOM 之前停止压测，内存会逐渐下降。客户端到服务端改为两个连接时，压测时未出现内存快速增长。

### 问题原因： ###

每一个 gRPC 连接均有一个独立的队列，挂在该连接的所有 streams 共享，请求相当于生产，往服务端发送请求相当于消费，当生产速度大于消费速度时，就会出现内存持续上长。该队列没有长度限制，所以会持续上长。快速上涨的原因是协程 transport.loopyWriter).run 没有被调度运行，队列消费停止，导致队列只增不减。停止压测后，协程 transport.loopyWriter).run 会恢复执行。

当不再消费时，可观察到大量如下协程：

grpc/internal/transport.(*Stream).waitOnHeader (0x90c8d5)
    runtime/netpoll.go:220 internal/poll.runtime_pollWait (0x46bdd5)

使用 netstat 命令可观察到发送队列大量堆积。

### 解决方案： ###

控制生产速度，即控制单个 gRPC 客户端连接发送的请求数量。此外，还可以启用客户端的 keepalive 关闭连接。

### 后话 ###

go gRPC 如果提供取 controlBuffer 的队列 list 的大小接口，可使得更为简单和友好。

### 相关源码： ###

*  **http2Client**

// 源码所在文件：google.golang.org/grpc/http2_client.go
    // http2Client 实现了接口 ClientTransport
    // http2Client implements the ClientTransport interface with HTTP2.
    type http2Client struct {
      conn net.Conn // underlying communication channel
      loopy *loopyWriter // 生产和消费关联的队列在这里面，所在文件：controlbuf.go
    
      // controlBuf delivers all the control related tasks (e.g., window
    	// updates, reset streams, and various settings) to the controller.
    	controlBuf *controlBuffer // 所在文件：controlbuf.go
      
      maxConcurrentStreams  uint32
      streamQuota           int64
    	streamsQuotaAvailable chan struct{}
      waitingStreams        uint32
      
      initialWindowSize int32
    }
    
    type controlBuffer struct {
      list *itemList // 队列
    }
    
    type loopyWriter struct {
      // 关联上 controlBuffer，
      // 消费 controlBuffer 中的队列 list，
      // 生产由 http2Client 通过 controlBuffer 进行。
      cbuf *controlBuffer
    }

*  **一个 gRPC 客户端连接被创建时，即会创建一个 run 协程，run 协程为队列的消费者**

// 源码所在文件：internal/transport/http2_client.go
    // 所在包名：transport
    // 打断点方法：
    // (dlv) b transport.newHTTP2Client
    // 被调用：协程 grpc.addrConn.resetTransport
    func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (_ *http2Client, err error) {
      // 建立连接，注意不同于 grpc.Dial，
      // grpc.Dial 实际不包含连接，对于 block 调用也只是等待连接状态为 Ready 。
      // transport.dial 的实现调用了 net.Dialer.DialContext，
      // 而 net.Dialer.DialContext 是更底层 Go 自带包的组成部分，不是 gRPC 的组成部分。
      // net.Dialer.DialContext 的实现支持：TCP、UDP、Unix等：。
      conn, err := dial(connectCtx, opts.Dialer, addr.Addr)
      t.controlBuf = newControlBuffer(t.ctxDone) // 含发送队列的初始化
    
      if t.keepaliveEnabled {
    		t.kpDormancyCond = sync.NewCond(&t.mu)
    		go t.keepalive() // 保活协程
    	}
      
      // Start the reader goroutine for incoming message. Each transport has
    	// a dedicated goroutine which reads HTTP2 frame from network. Then it
    	// dispatches the frame to the corresponding stream entity.
      go t.reader()
      
      // Send connection preface to server.
    	n, err := t.conn.Write(clientPreface)
      
      go func() {
        t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst)
        err := t.loopy.run()
      }
    }
    
    0  0x00000000008f305b in google.golang.org/grpc/internal/transport.newHTTP2Client
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/http2_client.go:166
    1  0x00000000009285a8 in google.golang.org/grpc/internal/transport.NewClientTransport
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/transport.go:577
    2  0x00000000009285a8 in google.golang.org/grpc.(*addrConn).createTransport
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/clientconn.go:1297
    3  0x0000000000927e48 in google.golang.org/grpc.(*addrConn).tryAllAddrs
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/clientconn.go:1227
    
       // 下列的 grpc.addrConn.resetTransport 是一个协程
    4  0x000000000092737f in google.golang.org/grpc.(*addrConn).resetTransport
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/clientconn.go:1142
    5  0x0000000000471821 in runtime.goexit
       at /usr/local/go/src/runtime/asm_amd64.s:1374
       
    // 源码所在文件：grpc/clientconn.go
    // 所在包名：grpc
    // 被调用：grpc.addrConn.getReadyTransport
    func (ac *addrConn) connect() error {
      // Start a goroutine connecting to the server asynchronously.
    	go ac.resetTransport()
    }
    
    // 传统类型的 RPC 调用从 grpc.ClientConn.Invoke 开始：
    //    XXX.pb.go // 编译 .proto 生成的文件
    // -> main.helloServiceClient.Hello
    // -> grpc.ClientConn.Invoke // 在 call.go 中，如果是 stream RPC，则从调用 grpc.ClientConn.NewStream 开始
    // -> grpc.invoke // 在 call.go 中
    // -> grpc.newClientStream // 在 stream.go 中
    // -> grpc.clientStream.newAttemptLocked // 在 stream.go 中
    // -> grpc.ClientConn.getTransport // 在 clientconn.go 中
    // -> grpc.pickerWrapper.pick // 在 picker_wrapper.go 中
    // -> grpc.addrConn.getReadyTransport
    // -> grpc.addrConn.connect // 创建协程 resetTransport
    // -> grpc.addrConn.resetTransport // ***是一个协程***
    // -> grpc.addrConn.tryAllAddrs
    // -> grpc.addrConn.createTransport // 在clientconn.go 中
    // -> transport.NewClientTransport // 在 transport.go 中
    // -> transport.newHTTP2Client
    // -> transport.dial
    // -> net.Dialer.DialContext // net 为 Go 自带包，不是 gRPC 包
    // -> net.sysDialer.dialSerial
    // -> net.sysDialer.dialSingle
    // -> net.sysDialer.dialTCP/dialUDP/dialUnix/dialIP
    // -> net.sysDialer.doDialTCP // 以 dialTCP 为例
    // -> net.internetSocket // 从这开始，和 C 语言的使用类似了，只不过包装了不同平台的
    // -> net.socket
    // -> net.sysSocket
    //
    // stream 类型的 RPC 从 NewStream 开始：
    // grpc.newClientStream 除被 grpc.invoke 调用外，还会被 stream.go 中的 grpc.ClientConn.NewStream 直接调用
    //    XXX.pb.go // 编译 .proto 生成的文件
    // -> grpc.ClientConn.NewStream // 在 stream.go 中
    // -> grpc.newClientStream // 在 stream.go 中
    // -> 从这开始同上述流程
    
    // 源码所在文件：grpc/clientconn.go
    // 所在包名：grpc
    // 被调用：调用源头为 grpc.ClientConn.NewStream，其实是 grpc.newClientStream 。
    // getReadyTransport returns the transport if ac's state is READY.
    // Otherwise it returns nil, false.
    // If ac's state is IDLE, it will trigger ac to connect.
    func (ac *addrConn) getReadyTransport() (transport.ClientTransport, bool) {
      // Trigger idle ac to connect.
    	if idle {
    		ac.connect()
    	}
    }

*  **消费协程 run 相关源代码摘要**

// 源码所在文件：internal/transport/controlbuf.go
    // Loopy receives frames from the control buffer.
    // Each frame is handled individually; most of the work done by loopy goes
    // into handling data frames. Loopy maintains a queue of active streams, and each
    // stream maintains a queue of data frames; as loopy receives data frames
    // it gets added to the queue of the relevant stream.
    // Loopy goes over this list of active streams by processing one node every iteration,
    // thereby closely resemebling to a round-robin scheduling over all streams. While
    // processing a stream, loopy writes out data bytes from this stream capped by the min
    // of http2MaxFrameLen, connection-level flow control and stream-level flow control.
    type loopyWriter struct {
      // cbuf 维护了队列（list *itemList），
      // 如果不加控制，就会导致内存大涨。
      cbuf      *controlBuffer
      sendQuota uint32
      
      // estdStreams is map of all established streams that are not cleaned-up yet.
    	// On client-side, this is all streams whose headers were sent out.
    	// On server-side, this is all streams whose headers were received.
      estdStreams map[uint32]*outStream // Established streams.
    
      // activeStreams is a linked-list of all streams that have data to send and some
    	// stream-level flow control quota.
    	// Each of these streams internally have a list of data items(and perhaps trailers
    	// on the server-side) to be sent out.
      activeStreams *outStreamList
    }
    
    // 源码所在文件：internal/transport/controlbuf.go
    func (l *loopyWriter) run() (err error) {
      // 通过 get 间接调用 dequeue 和 dequeueAll
      for {
        it, err := l.cbuf.get(true)
        if err != nil {
    			return err
    		}
    		if err = l.handle(it); err != nil {
    			return err
    		}
    		if _, err = l.processData(); err != nil {
    			return err
    		}
      }
    }
    
    func (c *controlBuffer) get(block bool) (interface{}, error) {
      for {
        c.mu.Lock() // 队列操作需要加锁保护
        ......
        // 消费队列（出队）
        h := c.list.dequeue().(cbItem)
        ......
        if !block {
    			c.mu.Unlock()
    			return nil, nil
    		}
        // 阻塞
        c.consumerWaiting = true
    		c.mu.Unlock()
    		select {
    		case <-c.ch: // 对应 executeAndPut 中唤醒的：c.ch <- struct{}
    		case <-c.done:
    			c.finish() // 清空队列
    			return nil, ErrConnClosing // indicates that the transport is closing
    		}
      }
    }
    
    func (c *controlBuffer) finish() {
      ......
      // 清空队列
      for head := c.list.dequeueAll(); head != nil; head = head.next {
      ......
    }

*  **特别说明**

每一次 gRPC 调用，客户端均会创建一个新的 Stream，  
该特性使得同一 gRPC 连接可以同时处理多个调用。请求的发送并不是同步的，而是基于队列的异步发送。  
每一个 gRPC 客户端连接均有一个自己的队列，gRPC 并没有直接限定队列大小，所以如果不加任何限制则会内存暴涨，直到 OOM 发生。

*  **生产者：发起调用的客户端**\*

message HelloReq { // 请求
        string text = 1;
    }
    message HelloRes { // 响应
        string text = 1;
    }
    service HelloService {
        rpc Hello(HelloReq) returns (HelloRes) {}
    }
    
    grpcClient := grpc.Dial(endpoint, opts)
    helloClient := NewHelloServiceClient(grpcClient)
    // Hello 调用为生产源头
    res, err := helloClient.Hello(ctx, &req)
    
    // Hello 的实现，为 protoc 编译生成的代码
    func (c *helloServiceClient) Hello(ctx context.Context, in *HelloReq, opts ...grpc.CallOption) (*HelloRes, error) {
    	out := new(HelloRes)
    	err := c.cc.Invoke(ctx, "/main.HelloService/Hello", in, out, opts...)
    	if err != nil {
    		return nil, err
    	}
    	return out, nil
    }
    
    // 源码所在文件：google.golang.org/grpc/call.go
    func (cc *ClientConn) Invoke(ctx context.Context, method string, args, reply interface{}, opts ...CallOption) error {
    	// allow interceptor to see all applicable call options, which means those
    	// configured as defaults from dial option as well as per-call options
    	opts = combine(cc.dopts.callOptions, opts)
    
    	if cc.dopts.unaryInt != nil {
    		return cc.dopts.unaryInt(ctx, method, args, reply, cc, invoke, opts...)
    	}
      // 转调用私有的 invoke 函数
    	return invoke(ctx, method, args, reply, cc, opts...)
    }
    
    // 源码所在文件：google.golang.org/grpc/call.go
    func invoke(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error {
      // newClientStream 间接往队列中生产消息
      // pprof 显示 newClientStream 调用的 withRetry 占用内存大头
    	cs, err := newClientStream(ctx, unaryStreamDesc, cc, method, opts...)
    	if err != nil {
    		return err
    	}
    	if err := cs.SendMsg(req); err != nil {
    		return err
    	}
    	return cs.RecvMsg(reply)
    }
    
    // 源码所在文件：google.golang.org/grpc/stream.go
    // 设置断点：(dlv) b clientStream.SendMsg
    func (cs *clientStream) SendMsg(m interface{}) (err error) {
    }
    
    // 源码所在文件：google.golang.org/grpc/stream.go
    // pprof 显示 newClientStream 消耗太多内存，而这又发生在其调用的 withRetry 中
    func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
      // 问题出在 newStream 分配的内存越来越多，
      // 但并非严格的泄漏，只是不断积累，但压力下来后会缓慢释放。
      // newStream 的实现是调用 NewStream：
      // cs.callHdr.PreviousAttempts = cs.numRetries
      // s, err := a.t.NewStream(cs.ctx, cs.callHdr)
      // cs.attempt.s = s
      // 这里 a 的类型为 csAttempt：
      // implements a single transport stream attempt within a clientStream
      op := func(a *csAttempt) error { return a.newStream() }
      // 内存问题所在：withRetry，进一步内存发生在非直接调用的：NewStream
    	if err := cs.withRetry(op, func() { cs.bufferForRetryLocked(0, op) }); err != nil {
    		cs.finish(err)
    		return nil, err
    	}
    }
    
    // 源码所在文件：google.golang.org/grpc/stream.go
    func (cs *clientStream) withRetry(op func(a *csAttempt) error, onSuccess func()) error {
      for { // 循环，内存上涨，这儿并没有循环
        // 调用 newStream，
        // 这里间接往队列中生产消息。
        err := op(a)
      }
    }
    
    // 源码所在文件：google.golang.org/grpc/stream.go
    func (a *csAttempt) newStream() error {
    	cs := a.cs
    	cs.callHdr.PreviousAttempts = cs.numRetries
      // 下列的 t 类型为：transport.ClientTransport，
      // 但注意 transport.ClientTransport 是一个 interface，并不是 struct。
      // 而 http2Client 是一个针对 ClientTransport 接口的实现。
    	s, err := a.t.NewStream(cs.ctx, cs.callHdr)
    }
    
    // 源码所在文件：google.golang.org/grpc/internal/transport/http2_client.go
    // NewStream creates a stream and registers it into the transport as "active"
    // streams.
    func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) {
      // 内存上涨，是因为队列中存了大量的 headerFields 和 s
      headerFields, err := t.createHeaderFields(ctx, callHdr)
      s := t.newStream(ctx, callHdr)
      // hdr 聚合了 headerFields 和 s
      hdr := &headerFrame{
        hf:        headerFields,
        initStream: func(id uint32) error {
          t.activeStreams[id] = s
        },
        wq:         s.wq,
      }
      for {
        // 调用 executeAndPut 入队（生产）
        // 内存上涨，是因为队列中存了大量的 hdr 。
        success, err := t.controlBuf.executeAndPut(func(it interface{}) bool {
        }, hdr)
      }
    }
    
    // 源码所在文件：google.golang.org/grpc/internal/transport/controlbuf.go
    func (c *controlBuffer) executeAndPut(f func(it interface{}) bool, it cbItem) (bool, error) {
      // 入队操作（生产）
      // 当入队快于出队（消费）时，就会出现内存上涨。
      c.list.enqueue(it)
      if it.isTransportResponseFrame() { // 调用接口 cbItem 定义的方法
        // counts the number of queued items that represent the response of an action initiated by the peer
        // 变量 transportResponseFrames 记录了队列大小
        c.transportResponseFrames++
      }
    }
    
    // 源码所在文件：google.golang.org/grpc/internal/transport/controlbuf.go
    func (c *controlBuffer) put(it cbItem) error {
      // 入队操作（生产）
    	_, err := c.executeAndPut(nil, it)
    	return err
    }
    
    type cbItem interface {
    	isTransportResponseFrame() bool
    }
    func (*registerStream) isTransportResponseFrame() bool { return false }
    func (h *headerFrame) isTransportResponseFrame() bool {
    	return h.cleanup != nil && h.cleanup.rst // Results in a RST_STREAM
    }
    func (c *cleanupStream) isTransportResponseFrame() bool { return c.rst } // Results in a RST_STREAM
    func (*dataFrame) isTransportResponseFrame() bool { return false }
    func (*incomingWindowUpdate) isTransportResponseFrame() bool { return false }
    func (*outgoingWindowUpdate) isTransportResponseFrame() bool {
    	return false // window updates are throttled by thresholds
    }
    func (*incomingSettings) isTransportResponseFrame() bool { return true } // Results in a settings ACK
    func (*outgoingSettings) isTransportResponseFrame() bool { return false }
    func (*incomingGoAway) isTransportResponseFrame() bool { return false }
    func (*goAway) isTransportResponseFrame() bool { return false }
    func (*ping) isTransportResponseFrame() bool { return true }
    func (*outFlowControlSizeRequest) isTransportResponseFrame() bool { return false }

*  \*\*协程 transport.loopyWriter.run \*\*

0  0x000000000043bfa5 in runtime.gopark
       at /usr/local/go/src/runtime/proc.go:307
    1  0x000000000044c10f in runtime.selectgo
       at /usr/local/go/src/runtime/select.go:338
       一个连接只有一个 run 协程
    2  0x00000000008eca2f in google.golang.org/grpc/internal/transport.(*controlBuffer).get
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/controlbuf.go:417
    3  0x00000000008ed76e in google.golang.org/grpc/internal/transport.(*loopyWriter).run
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/controlbuf.go:544
    4  0x000000000090f13b in google.golang.org/grpc/internal/transport.newHTTP2Client.func3
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/http2_client.go:356
    5  0x0000000000471821 in runtime.goexit
       at /usr/local/go/src/runtime/asm_amd64.s:1374

*  **协程 transport.controlBuffer.throttle**
 *  **协程 transport.http2Client.reader**

0  0x000000000043bfa5 in runtime.gopark
       at /usr/local/go/src/runtime/proc.go:307
    1  0x000000000044c10f in runtime.selectgo
       at /usr/local/go/src/runtime/select.go:338
    2  0x00000000008ec335 in google.golang.org/grpc/internal/transport.(*controlBuffer).throttle
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/controlbuf.go:319
       
       文件 http2_client.go 函数 transport.newHTTP2Client 会创建协程 reader
    3  0x00000000008fdadd in google.golang.org/grpc/internal/transport.(*http2Client).reader
       at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/http2_client.go:1293
    4  0x0000000000471821 in runtime.goexit
       at /usr/local/go/src/runtime/asm_amd64.s:1374

*  **协程 transport.Stream.waitOnHeader**

// 大量 waitOnHeader 协程
    func (s *Stream) waitOnHeader() {
    	if s.headerChan == nil {
    		// On the server headerChan is always nil since a stream originates
    		// only after having received headers.
    		return
    	}
    	select {
    	case <-s.ctx.Done():
    		// Close the stream to prevent headers/trailers from changing after
    		// this function returns.
    		s.ct.CloseStream(s, ContextErr(s.ctx.Err()))
    		// headerChan could possibly not be closed yet if closeStream raced
    		// with operateHeaders; wait until it is closed explicitly here.
    		<-s.headerChan
    	case <-s.headerChan:
    	}
    }
    
    // 实为调用协程
    0  0x000000000043bfa5 in runtime.gopark
        at /usr/local/go/src/runtime/proc.go:307
     1  0x000000000044c10f in runtime.selectgo
        at /usr/local/go/src/runtime/select.go:338
    
     2  0x000000000090c8d5 in google.golang.org/grpc/internal/transport.(*Stream).waitOnHeader
        at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/transport.go:321
     3  0x0000000000942805 in google.golang.org/grpc/internal/transport.(*Stream).RecvCompress
        at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/internal/transport/transport.go:336
     4  0x0000000000942805 in google.golang.org/grpc.(*csAttempt).recvMsg
        at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/stream.go:894
     5  0x000000000094ad06 in google.golang.org/grpc.(*clientStream).RecvMsg.func1
        at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/stream.go:759
     6  0x000000000094057c in google.golang.org/grpc.(*clientStream).withRetry
        at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/stream.go:617
     7  0x0000000000941505 in google.golang.org/grpc.(*clientStream).RecvMsg
        at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/stream.go:758
     8  0x0000000000921d3b in google.golang.org/grpc.invoke
        at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/call.go:73
     9  0x0000000000921ad3 in google.golang.org/grpc.(*ClientConn).Invoke
        at /root/go/pkg/mod/google.golang.org/grpc@v1.33.2/call.go:37
    10  0x0000000000b185f4 in /root/hello/grpc/proto.(*HelloClient).Call
        at /root/hello/hello.pb.go:70

*  **协程 poll\_runtime\_pollWait**

// poll_runtime_pollWait, which is internal/poll.runtime_pollWait,
    // waits for a descriptor to be ready for reading or writing,
    // according to mode, which is 'r' or 'w'.
    // This returns an error code; the codes are defined above.
    //go:linkname poll_runtime_pollWait internal/poll.runtime_pollWait
    func poll_runtime_pollWait(pd *pollDesc, mode int) int {
            errcode := netpollcheckerr(pd, int32(mode))
            if errcode != pollNoError {
                    return errcode
            }
            // As for now only Solaris, illumos, and AIX use level-triggered IO.
            if GOOS == "solaris" || GOOS == "illumos" || GOOS == "aix" {
                    netpollarm(pd, mode)
            }
            for !netpollblock(pd, int32(mode), false) {
                    errcode = netpollcheckerr(pd, int32(mode))
                    if errcode != pollNoError {
                            return errcode
                    }
                    // Can happen if timeout has fired and unblocked us,
                    // but before we had a chance to run, timeout has been reset.
                    // Pretend it has not happened and retry.
            }
            return pollNoError
    }
    
    函数 runtime.gopark 用于协程的切换
    0  0x000000000043bfa5 in runtime.gopark
       at /usr/local/go/src/runtime/proc.go:307
    1  0x000000000043447b in runtime.netpollblock
       at /usr/local/go/src/runtime/netpoll.go:436
    
    2  0x000000000046bdd5 in internal/poll.runtime_pollWait
       at /usr/local/go/src/runtime/netpoll.go:220
    3  0x00000000004d9685 in internal/poll.(*pollDesc).wait
       at /usr/local/go/src/internal/poll/fd_poll_runtime.go:87
    4  0x00000000004da6c5 in internal/poll.(*pollDesc).waitRead
       at /usr/local/go/src/internal/poll/fd_poll_runtime.go:92
    5  0x00000000004da6c5 in internal/poll.(*FD).Read
       at /usr/local/go/src/internal/poll/fd_unix.go:159
    6  0x00000000005327af in net.(*netFD).Read
       at /usr/local/go/src/net/fd_posix.go:55
    7  0x000000000054688e in net.(*conn).Read
       at /usr/local/go/src/net/net.go:182
    8  0x00000000006e14b8 in net/http.(*connReader).backgroundRead
       at /usr/local/go/src/net/http/server.go:690
    9  0x0000000000471821 in runtime.goexit
       at /usr/local/go/src/runtime/asm_amd64.s:1374

*  **队列**

type itemList struct {
    	head *itemNode
    	tail *itemNode
    }
    
    type itemNode struct {
    	it   interface{}
    	next *itemNode
    }
    
    // 入队（生产）
    //
    // 从这可看到，
    // 队列没有大小限制，生产（入队）不受限，
    // 所以一旦生产速度大于消费速度，就会出现堆积导致内存上涨。
    func (il *itemList) enqueue(i interface{}) {
    	n := &itemNode{it: i}
    	if il.tail == nil {
    		il.head, il.tail = n, n
    		return
    	}
    	il.tail.next = n
    	il.tail = n
    }
    
    // 出队（消费）
    func (il *itemList) dequeue() interface{} {
    	if il.head == nil {
    		return nil
    	}
    	i := il.head.it
    	il.head = il.head.next
    	if il.head == nil {
    		il.tail = nil
    	}
    	return i
    }
    
    // 清空（消费），直接丢弃了
    func (il *itemList) dequeueAll() *itemNode {
    	h := il.head
    	il.head, il.tail = nil, nil
    	return h
    }
    
    func (il *itemList) isEmpty() bool {
    	return il.head == nil
    }

// 源码所在文件：internal/transport/controlbuf.go
    //    transport.loopyWriter.run
    // -> transport.loopyWriter.handle
    // -> transport.loopyWriter.headerHandler
    // -> transport.loopyWriter.writeHeader // 阻塞在这了
    func (l *loopyWriter) handle(i interface{}) error {
    	switch i := i.(type) {
    	case *incomingWindowUpdate:
    		return l.incomingWindowUpdateHandler(i)
    	case *outgoingWindowUpdate:
    		return l.outgoingWindowUpdateHandler(i)
    	case *incomingSettings:
    		return l.incomingSettingsHandler(i)
    	case *outgoingSettings:
    		return l.outgoingSettingsHandler(i)
    	case *headerFrame:
    		return l.headerHandler(i) // 阻塞在这了
      ......
    }
    
    // 源码所在文件：internal/transport/controlbuf.go
    func (l *loopyWriter) writeHeader(streamID uint32, endStream bool, hf []hpack.HeaderField, onWrite func()) error {
      // 阻塞在这儿：
      // 结构体 frame 定义在 internal/transport/http_util.go 文件中，
      // 成员 fr 的类型为 http2.Framer，定义在 x/net/http2/frame.go 文件中
      err = l.framer.fr.WriteHeaders(http2.HeadersFrameParam{
      })
    }