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

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了go gRPC 客户端内存暴涨原因分析，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含21728字，纯文字阅读大概需要32分钟。

内容图文

创建一个 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 客户端连接发送的请求数量。

后话

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{
  })
}

内容总结

以上是互联网集市为您收集整理的go gRPC 客户端内存暴涨原因分析全部内容，希望文章能够帮你解决go gRPC 客户端内存暴涨原因分析所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---