graph TB
    subgraph "gRPC 传输层架构"
        subgraph "传输接口层"
            T1[ClientTransport 接口]
            T2[ServerTransport 接口]
        end
        
        subgraph "HTTP/2 实现层"
            H1[http2Client]
            H2[http2Server]
            H3[HTTP/2 Framer]
            H4[HPACK 编码器]
        end
        
        subgraph "流管理层"
            S1[Stream 管理器]
            S2[流状态机]
            S3[流配额管理]
            S4[流窗口控制]
        end
        
        subgraph "控制层"
            C1[controlBuffer]
            C2[loopyWriter]
            C3[指令队列]
            C4[批量写入器]
        end
        
        subgraph "数据处理层"
            D1[数据帧处理]
            D2[头部帧处理]
            D3[控制帧处理]
            D4[错误处理]
        end
        
        subgraph "网络层"
            N1[TCP 连接]
            N2[TLS 安全层]
            N3[Keepalive 管理]
        end
    end
    
    T1 --> H1
    T2 --> H2
    H1 --> H3
    H2 --> H3
    H3 --> H4
    H1 --> S1
    H2 --> S1
    S1 --> S2
    S1 --> S3
    S3 --> S4
    H1 --> C1
    H2 --> C1
    C1 --> C2
    C2 --> C3
    C3 --> C4
    H1 --> D1
    H2 --> D1
    D1 --> D2
    D2 --> D3
    D3 --> D4
    H1 --> N1
    H2 --> N1
    N1 --> N2
    N1 --> N3
    
    style T1 fill:#e3f2fd
    style H1 fill:#f3e5f5
    style S1 fill:#e8f5e8
    style C1 fill:#fff3e0
    style D1 fill:#fce4ec
    style N1 fill:#f1f8e9

sequenceDiagram
    participant App as 应用层
    participant CT as ClientTransport
    participant ST as ServerTransport
    participant CB as controlBuffer
    participant LW as loopyWriter
    participant F as Framer
    participant Net as 网络层
    
    Note over App,Net: 连接建立阶段
    App->>CT: NewHTTP2Client()
    CT->>Net: 建立 TCP 连接
    Net-->>CT: 连接就绪
    CT->>F: 发送 HTTP/2 preface
    CT->>F: 发送 SETTINGS 帧
    F->>Net: 写入网络
    
    Note over App,Net: 流创建阶段
    App->>CT: NewStream()
    CT->>CT: 分配 streamID
    CT->>CB: put(headerFrame)
    CB->>LW: 通知有数据
    LW->>F: WriteHeaders()
    F->>Net: 发送 HEADERS 帧
    
    Note over App,Net: 数据传输阶段
    App->>CT: Write(data)
    CT->>CB: put(dataFrame)
    CB->>LW: 处理数据帧
    LW->>LW: 检查流控配额
    LW->>F: WriteData()
    F->>Net: 发送 DATA 帧
    
    Note over App,Net: 响应处理阶段
    Net->>ST: 接收帧数据
    ST->>ST: 解析帧类型
    ST->>ST: 处理业务逻辑
    ST->>CB: put(responseFrame)
    CB->>LW: 发送响应
    LW->>F: WriteData()
    F->>Net: 发送响应帧

// 位置：internal/transport/transport.go
type ClientTransport interface {
    // Write 发送数据到指定流
    Write(s *Stream, hdr []byte, data mem.BufferSlice, opts *Options) error
    
    // NewStream 创建新的流
    NewStream(ctx context.Context, callHdr *CallHdr) (*Stream, error)
    
    // CloseStream 关闭指定流
    CloseStream(stream *Stream, err error) error
    
    // Error 返回错误通道
    Error() <-chan struct{}
    
    // GoAway 返回 GoAway 通道
    GoAway() <-chan struct{}
    
    // GetGoAwayReason 获取 GoAway 原因
    GetGoAwayReason() GoAwayReason
    
    // RemoteAddr 返回远程地址
    RemoteAddr() net.Addr
    
    // LocalAddr 返回本地地址
    LocalAddr() net.Addr
    
    // WaitForHandshake 等待握手完成
    WaitForHandshake() error
}

// 位置：internal/transport/http2_client.go
type http2Client struct {
    ctx        context.Context
    ctxDone    <-chan struct{}
    cancel     context.CancelFunc
    userAgent  string
    
    // 连接相关
    conn       net.Conn
    localAddr  net.Addr
    remoteAddr net.Addr
    
    // HTTP/2 相关
    framer     *framer
    hBuf       *bytes.Buffer
    hEnc       *hpack.Encoder
    
    // 控制相关
    controlBuf *controlBuffer
    fc         *trInFlow
    
    // 流管理
    mu                     sync.Mutex
    activeStreams          map[uint32]*Stream
    nextID                 uint32
    maxConcurrentStreams   uint32
    streamQuota            int64
    streamsQuotaAvailable  chan struct{}
    waitingStreams         uint32
    
    // 状态管理
    state                  transportState
    goAway                 chan struct{}
    awakenKeepalive        chan struct{}
    
    // 统计信息
    czData                 *channelzData
    onClose                func(GoAwayReason)
    
    // 配置参数
    keepaliveEnabled       bool
    keepaliveParams        keepalive.ClientParameters
    initialWindowSize      int32
    bdpEst                 *bdpEstimator
}

// NewHTTP2Client 创建新的 HTTP/2 客户端传输
func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onClose func(GoAwayReason)) (_ ClientTransport, err error) {
    scheme := "http"
    ctx, cancel := context.WithCancel(ctx)
    defer func() {
        if err != nil {
            cancel()
        }
    }()

    conn, authInfo, err := newHTTP2Conn(connectCtx, addr, opts)
    if err != nil {
        return nil, connectionErrorf(isTemporary(err), err, "transport: error while dialing: %v", err)
    }

    // 发送连接 preface
    if err := conn.Write(clientPreface); err != nil {
        conn.Close()
        return nil, connectionErrorf(true, err, "transport: failed to write client preface: %v", err)
    }

    // 创建 framer
    framer := newFramer(conn, writeBufferSize, readBufferSize, opts.MaxHeaderListSize)
    
    // 创建 http2Client 实例
    t := &http2Client{
        ctx:                   ctx,
        ctxDone:               ctx.Done(),
        cancel:                cancel,
        userAgent:             opts.UserAgent,
        conn:                  conn,
        localAddr:             conn.LocalAddr(),
        remoteAddr:            conn.RemoteAddr(),
        authInfo:              authInfo,
        framer:                framer,
        fc:                    &trInFlow{limit: uint32(icwz)},
        sendQuotaPool:         newQuotaPool(defaultWindowSize),
        localSendQuota:        newQuotaPool(defaultWindowSize),
        scheme:                scheme,
        activeStreams:         make(map[uint32]*Stream),
        creds:                 opts.TransportCredentials,
        maxConcurrentStreams:  defaultMaxStreamsClient,
        streamQuota:           defaultMaxStreamsClient,
        streamsQuotaAvailable: make(chan struct{}, 1),
        czData:                new(channelzData),
        onClose:               onClose,
        nextID:                1,
        maxSendHeaderListSize: opts.MaxHeaderListSize,
        keepaliveEnabled:      keepaliveEnabled,
        keepaliveParams:       kp,
        initialWindowSize:     initialWindowSize,
        goAway:                make(chan struct{}),
        awakenKeepalive:       make(chan struct{}, 1),
        bdpEst:                bdpEst,
    }

    // 创建控制缓冲区
    t.controlBuf = newControlBuffer(t.ctxDone)

    // 发送初始设置
    if err := t.writeSettings(); err != nil {
        t.Close(err)
        return nil, err
    }

    // 更新窗口大小
    if delta := uint32(icwz - defaultWindowSize); delta > 0 {
        if err := t.framer.fr.WriteWindowUpdate(0, delta); err != nil {
            t.Close(err)
            return nil, err
        }
    }

    // 启动 reader 和 loopyWriter
    go t.reader()
    go t.loopy.run()
    
    if t.keepaliveEnabled {
        go t.keepalive()
    }

    return t, nil
}

// 位置：internal/transport/http2_server.go
type http2Server struct {
    ctx               context.Context
    done              *grpcsync.Event
    conn              net.Conn
    loopy             *loopyWriter
    readerDone        chan struct{}
    loopyWriterDone   chan struct{}
    
    // HTTP/2 相关
    framer            *framer
    hBuf              *bytes.Buffer
    hEnc              *hpack.Encoder
    
    // 流管理
    mu                sync.Mutex
    activeStreams     map[uint32]*ServerStream
    streamSendQuota   uint32
    
    // 控制相关
    controlBuf        *controlBuffer
    fc                *trInFlow
    sendQuotaPool     *quotaPool
    localSendQuota    *quotaPool
    
    // 配置参数
    maxStreams        uint32
    maxStreamMsgSize  int
    initialWindowSize int32
    keepaliveParams   keepalive.ServerParameters
    
    // 统计信息
    czData            *channelzData
    bufferPool        mem.BufferPool
    
    // 回调函数
    stats             []stats.Handler
    inTapHandle       tap.ServerInHandle
}

// NewServerTransport 创建新的服务器传输
func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error) {
    writeBufSize := config.WriteBufferSize
    readBufSize := config.ReadBufferSize
    maxHeaderListSize := defaultServerMaxHeaderListSize
    if config.MaxHeaderListSize != nil {
        maxHeaderListSize = *config.MaxHeaderListSize
    }

    framer := newFramer(conn, writeBufSize, readBufSize, maxHeaderListSize)
    
    // 发送服务器设置
    var settings []http2.Setting
    if config.MaxStreams != math.MaxUint32 {
        settings = append(settings, http2.Setting{
            ID:  http2.SettingMaxConcurrentStreams,
            Val: config.MaxStreams,
        })
    }
    
    // 其他设置...
    if err := framer.fr.WriteSettings(settings...); err != nil {
        return nil, connectionErrorf(false, err, "transport: %v", err)
    }

    // 创建 http2Server 实例
    t := &http2Server{
        ctx:               context.Background(),
        done:              grpcsync.NewEvent(),
        conn:              conn,
        framer:            framer,
        hBuf:              &bytes.Buffer{},
        hEnc:              hpack.NewEncoder(&bytes.Buffer{}),
        maxStreams:        maxStreams,
        inTapHandle:       config.InTapHandle,
        fc:                &trInFlow{limit: uint32(icwz)},
        sendQuotaPool:     newQuotaPool(defaultWindowSize),
        localSendQuota:    newQuotaPool(defaultWindowSize),
        activeStreams:     make(map[uint32]*ServerStream),
        streamSendQuota:   defaultWindowSize,
        stats:             config.StatsHandlers,
        keepaliveParams:   kp,
        maxStreamMsgSize:  config.MaxReceiveMessageSize,
        initialWindowSize: iwz,
        czData:            new(channelzData),
        bufferPool:        config.BufferPool,
    }

    t.controlBuf = newControlBuffer(t.done)
    
    // 启动 loopy writer
    t.loopy = newLoopyWriter(serverSide, t.framer, t.controlBuf, t.bdpEst, t.conn, t.logger, t.outgoingGoAwayHandler, t.keepaliveParams)
    go t.loopy.run()
    
    // 启动 keepalive
    t.kep.Start(t)
    
    return t, nil
}

graph TB
    subgraph "HTTP/2 流控制机制"
        subgraph "发送端"
            A1[应用数据]
            A2[检查连接级配额]
            A3[检查流级配额]
            A4[发送 DATA 帧]
            A5[更新配额计数]
        end
        
        subgraph "接收端"
            B1[接收 DATA 帧]
            B2[更新接收计数]
            B3[应用消费数据]
            B4[发送 WINDOW_UPDATE]
        end
        
        subgraph "配额管理"
            C1[连接级窗口]
            C2[流级窗口]
            C3[未确认字节数]
            C4[窗口更新阈值]
        end
    end
    
    A1 --> A2
    A2 --> A3
    A3 --> A4
    A4 --> A5
    A4 --> B1
    B1 --> B2
    B2 --> B3
    B3 --> B4
    B4 --> A2
    
    A2 --> C1
    A3 --> C2
    B2 --> C3
    B4 --> C4
    
    style A1 fill:#e3f2fd
    style B1 fill:#f3e5f5
    style C1 fill:#e8f5e8

// 位置：internal/transport/flowcontrol.go
type inFlow struct {
    mu sync.Mutex
    
    // 窗口大小限制
    limit uint32
    
    // 未确认的字节数（已接收但未发送 WINDOW_UPDATE）
    unacked uint32
    
    // 有效窗口大小
    effectiveWindowSize uint32
}

// onData 处理接收到的数据
func (f *inFlow) onData(n uint32) error {
    f.mu.Lock()
    defer f.mu.Unlock()
    
    f.unacked += n
    if f.unacked >= f.limit {
        return fmt.Errorf("received %d-bytes data exceeding the limit %d bytes", f.unacked, f.limit)
    }
    return nil
}

// onRead 处理应用读取数据
func (f *inFlow) onRead(n uint32) uint32 {
    f.mu.Lock()
    defer f.mu.Unlock()
    
    if f.unacked < n {
        n = f.unacked
    }
    f.unacked -= n
    
    // 如果未确认字节数超过阈值，返回需要更新的窗口大小
    if f.unacked >= f.limit/4 {
        wu := f.unacked
        f.unacked = 0
        return wu
    }
    return 0
}

// reset 重置流控制状态
func (f *inFlow) reset() uint32 {
    f.mu.Lock()
    defer f.mu.Unlock()
    
    wu := f.unacked
    f.unacked = 0
    return wu
}

// outFlow 发送流控制
type outFlow struct {
    mu sync.Mutex
    
    // 发送配额
    sendQuota uint32
    
    // 已发送但未确认的字节数
    bytesOutStanding uint32
    
    // 初始窗口大小
    initialWindowSize uint32
}

// add 增加发送配额
func (f *outFlow) add(n uint32) {
    f.mu.Lock()
    defer f.mu.Unlock()
    
    f.sendQuota += n
}

// consume 消费发送配额
func (f *outFlow) consume(n uint32) bool {
    f.mu.Lock()
    defer f.mu.Unlock()
    
    if f.sendQuota < n {
        return false
    }
    
    f.sendQuota -= n
    f.bytesOutStanding += n
    return true
}

// replenish 补充配额（收到 WINDOW_UPDATE）
func (f *outFlow) replenish(n uint32) {
    f.mu.Lock()
    defer f.mu.Unlock()
    
    if f.bytesOutStanding < n {
        n = f.bytesOutStanding
    }
    f.bytesOutStanding -= n
}

sequenceDiagram
    participant App as 应用层
    participant LW as loopyWriter
    participant OF as outFlow
    participant Net as 网络层
    participant IF as inFlow
    participant Peer as 对端
    
    Note over App,Peer: 数据发送流程
    App->>LW: 发送数据
    LW->>OF: consume(dataSize)
    
    alt 有足够配额
        OF-->>LW: true
        LW->>Net: 发送 DATA 帧
        Net->>Peer: HTTP/2 DATA 帧
        LW->>OF: 更新 bytesOutStanding
    else 配额不足
        OF-->>LW: false
        LW->>LW: 流进入等待状态
    end
    
    Note over App,Peer: 窗口更新流程
    Peer->>Net: WINDOW_UPDATE 帧
    Net->>IF: 处理窗口更新
    IF->>OF: replenish(increment)
    OF->>LW: 通知配额可用
    LW->>LW: 激活等待的流

// 位置：internal/transport/http2_client.go
type framer struct {
    fr *http2.Framer
    
    // 写入相关
    wmu       sync.Mutex
    wbuf      []byte
    
    // 读取相关
    rmu       sync.Mutex
    rbuf      []byte
}

// newFramer 创建新的帧处理器
func newFramer(conn net.Conn, writeBufferSize, readBufferSize int, maxHeaderListSize uint32) *framer {
    if writeBufferSize < 0 {
        writeBufferSize = 0
    }
    if readBufferSize <= 0 {
        readBufferSize = defaultReadBufferSize
    }
    
    r := bufio.NewReaderSize(conn, readBufferSize)
    w := bufio.NewWriterSize(conn, writeBufferSize)
    f := &framer{
        fr: http2.NewFramer(w, r),
    }
    
    f.fr.SetMaxReadFrameSize(http2MaxFrameLen)
    if maxHeaderListSize != 0 {
        f.fr.SetMaxHeaderListSize(maxHeaderListSize)
    }
    f.fr.ReadMetaHeaders = hpack.NewDecoder(http2InitHeaderTableSize, nil)
    
    return f
}

// 位置：internal/transport/http2_server.go
func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeadersFrame, handle func(*ServerStream)) error {
    streamID := frame.Header().StreamID
    
    // 验证流 ID
    if streamID%2 != 1 || streamID <= t.maxStreamID {
        return status.Errorf(codes.Internal, "transport: invalid stream id: %d", streamID)
    }
    t.maxStreamID = streamID
    
    // 检查并发流限制
    t.mu.Lock()
    if t.state != reachable {
        t.mu.Unlock()
        return ErrConnClosing
    }
    if uint32(len(t.activeStreams))+1 > t.maxStreams {
        t.mu.Unlock()
        t.controlBuf.put(&cleanupStream{
            streamID: streamID,
            rst:      true,
            rstCode:  http2.ErrCodeRefusedStream,
            onWrite:  func() {},
        })
        return nil
    }
    t.mu.Unlock()

    // 解析头部
    var (
        method      string
        path        string
        authority   string
        contentType string
        timeoutStr  string
        encoding    string
    )
    
    for _, hf := range frame.Fields {
        switch hf.Name {
        case ":method":
            method = hf.Value
        case ":path":
            path = hf.Value
        case ":authority":
            authority = hf.Value
        case "content-type":
            contentType = hf.Value
        case "grpc-timeout":
            timeoutStr = hf.Value
        case "grpc-encoding":
            encoding = hf.Value
        }
    }

    // 验证必需的头部
    if method != "POST" {
        return status.Errorf(codes.Internal, "transport: invalid method %q", method)
    }
    if path == "" {
        return status.Errorf(codes.Internal, "transport: missing :path header")
    }

    // 解析超时
    var timeout time.Duration
    if timeoutStr != "" {
        var err error
        if timeout, err = decodeTimeout(timeoutStr); err != nil {
            return status.Errorf(codes.Internal, "transport: malformed grpc-timeout: %v", err)
        }
    }

    // 创建服务器流
    s := &ServerStream{
        id:             streamID,
        st:             t,
        buf:            newRecvBuffer(),
        fc:             &inFlow{limit: uint32(t.initialWindowSize)},
        recvCompress:   encoding,
        method:         path,
        contentSubtype: contentSubtype(contentType),
    }

    if timeout != 0 {
        s.ctx, s.cancel = context.WithTimeout(ctx, timeout)
    } else {
        s.ctx, s.cancel = context.WithCancel(ctx)
    }

    // 注册流
    t.mu.Lock()
    if t.state != reachable {
        t.mu.Unlock()
        s.cancel()
        return ErrConnClosing
    }
    if t.activeStreams == nil {
        t.activeStreams = make(map[uint32]*ServerStream)
    }
    t.activeStreams[streamID] = s
    if len(t.activeStreams) == 1 {
        t.idle = time.Time{}
    }
    t.mu.Unlock()

    // 处理流
    if frame.StreamEnded() {
        s.compareAndSwapState(streamActive, streamReadDone)
    }
    
    handle(s)
    return nil
}

// 位置：internal/transport/http2_server.go
func (t *http2Server) handleData(f *http2.DataFrame) {
    size := f.Header().Length
    var sendBDPPing bool
    if t.bdpEst != nil {
        sendBDPPing = t.bdpEst.add(size)
    }
    
    // 解耦连接级流控制和应用读取
    // 连接级流控制的更新不应依赖于用户应用是否读取了数据
    if w := t.fc.onData(size); w > 0 {
        t.controlBuf.put(&outgoingWindowUpdate{
            streamID:  0,
            increment: w,
        })
    }
    
    if sendBDPPing {
        // 避免过度 ping 检测（例如在 L7 代理中）
        // 通过在 BDP ping 之前发送窗口更新来实现
        if w := t.fc.reset(); w > 0 {
            t.controlBuf.put(&outgoingWindowUpdate{
                streamID:  0,
                increment: w,
            })
        }
        t.controlBuf.put(bdpPing)
    }
    
    // 选择正确的流进行分发
    s, ok := t.getStream(f)
    if !ok {
        return
    }
    if s.getState() == streamReadDone {
        t.closeStream(s, true, http2.ErrCodeStreamClosed, false)
        return
    }
    
    if size > 0 {
        if err := s.fc.onData(size); err != nil {
            t.closeStream(s, true, http2.ErrCodeFlowControl, false)
            return
        }
        if f.Header().Flags.Has(http2.FlagDataPadded) {
            if w := s.fc.onRead(size - uint32(len(f.Data()))); w > 0 {
                t.controlBuf.put(&outgoingWindowUpdate{s.id, w})
            }
        }
        
        // 复制数据到流缓冲区
        if len(f.Data()) > 0 {
            pool := t.bufferPool
            if pool == nil {
                pool = mem.DefaultBufferPool()
            }
            s.write(recvMsg{buffer: mem.Copy(f.Data(), pool)})
        }
    }
    
    if f.StreamEnded() {
        // 从客户端接收到流结束
        s.compareAndSwapState(streamActive, streamReadDone)
        s.write(recvMsg{err: io.EOF})
    }
}

// 传输状态枚举
type transportState int

const (
    reachable transportState = iota
    closing
    draining
)

// 连接状态转换
func (t *http2Client) updateState(s transportState) {
    t.mu.Lock()
    defer t.mu.Unlock()
    
    if t.state == s {
        return
    }
    
    t.state = s
    switch s {
    case closing:
        // 开始关闭过程
        t.closeAllStreams()
    case draining:
        // 开始排空过程
        t.startDraining()
    }
}

// 位置：internal/transport/http2_client.go
func (t *http2Client) keepalive() {
    p := &ping{data: [8]byte{}}
    // 填充随机数据
    binary.LittleEndian.PutUint64(p.data[:], uint64(time.Now().UnixNano()))
    
    timer := time.NewTimer(t.keepaliveParams.Time)
    for {
        select {
        case <-timer.C:
            if atomic.CompareAndSwapUint32(&t.activity, 1, 0) {
                timer.Reset(t.keepaliveParams.Time)
                continue
            }
            // 发送 keepalive ping
            t.controlBuf.put(p)
            timer.Reset(t.keepaliveParams.Timeout)
        case <-t.awakenKeepalive:
            if !timer.Stop() {
                <-timer.C
            }
            timer.Reset(t.keepaliveParams.Time)
        case <-t.ctx.Done():
            if !timer.Stop() {
                <-timer.C
            }
            return
        }
    }
}

// 位置：internal/transport/http2_server.go
func (t *http2Server) drain(code http2.ErrCode, debugData []byte) {
    t.mu.Lock()
    defer t.mu.Unlock()
    if t.drainEvent != nil {
        return
    }
    t.drainEvent = grpcsync.NewEvent()
    t.controlBuf.put(&goAway{code: code, debugData: debugData, headsUp: true})
}

func (t *http2Server) Close(err error) {
    t.mu.Lock()
    if t.state == closing {
        t.mu.Unlock()
        return
    }
    t.state = closing
    streams := t.activeStreams
    t.activeStreams = nil
    t.mu.Unlock()
    
    t.controlBuf.finish()
    
    // 关闭所有活跃流
    for _, s := range streams {
        s.cancel()
    }
    
    if err != ErrConnClosing {
        // 通知所有等待的 goroutine
        t.conn.Close()
    }
}

// 位置：internal/transport/controlbuf.go
type controlBuffer struct {
    ch   chan struct{}
    done <-chan struct{}
    mu   sync.Mutex
    
    consumerWaiting bool
    list            *itemList
    err             error
    
    // 限流相关
    throttle func()
}

// 控制缓冲区项目接口
type cbItem interface {
    isTransportResponseFrame() bool
}

// 数据帧
type dataFrame struct {
    streamID    uint32
    endStream   bool
    h           []byte
    data        mem.BufferSlice
    processing  bool
    onEachWrite func()
}

// 头部帧
type headerFrame struct {
    streamID  uint32
    hf        []hpack.HeaderField
    endStream bool
    initStream func(uint32)
    onOrphaned func(error)
    onWrite    func()
    cleanup    *cleanupStream
}

// 窗口更新帧
type outgoingWindowUpdate struct {
    streamID  uint32
    increment uint32
}

// 位置：internal/transport/controlbuf.go
type loopyWriter struct {
    side            side
    cbuf            *controlBuffer
    sendQuota       uint32
    oiws            uint32
    estdStreams     map[uint32]*outStream
    activeStreams   *outStreamList
    framer          *framer
    hBuf            *bytes.Buffer
    hEnc            *hpack.Encoder
    bdpEst          *bdpEstimator
    draining        bool
    conn            net.Conn
    logger          *grpclog.PrefixLogger
    
    // 流状态管理
    streamQuota     int64
    maxFrameSize    uint32
}

// run 主循环
func (l *loopyWriter) run() (err error) {
    defer func() {
        if err == ErrConnClosing {
            err = nil
        }
    }()
    
    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
        }
        gosched := true
    hasData:
        for {
            it, err := l.cbuf.get(false)
            if err != nil {
                return err
            }
            if it != nil {
                if err = l.handle(it); err != nil {
                    return err
                }
                if _, err = l.processData(); err != nil {
                    return err
                }
                continue hasData
            }
            isEmpty, err := l.processData()
            if err != nil {
                return err
            }
            if !isEmpty {
                continue hasData
            }
            if gosched {
                gosched = false
                if l.framer.writer.offset < minBatchSize {
                    runtime.Gosched()
                    continue hasData
                }
            }
            l.framer.writer.Flush()
            break hasData
        }
    }
}

// processData 处理数据帧
func (l *loopyWriter) processData() (bool, error) {
    if l.sendQuota == 0 {
        return true, nil
    }
    str := l.activeStreams.dequeue() // 移除第一个流
    if str == nil {
        return true, nil
    }
    
    reader := str.reader
    dataItem := str.itl.peek().(*dataFrame) // 查看第一个数据项
    if !dataItem.processing {
        dataItem.processing = true
        str.reader.Reset(dataItem.data)
        dataItem.data.Free()
    }

    if len(dataItem.h) == 0 && reader.Remaining() == 0 { // 空数据帧
        // 客户端发送带有 endStream = true 的空数据帧
        if err := l.framer.fr.WriteData(dataItem.streamID, dataItem.endStream, nil); err != nil {
            return false, err
        }
        str.itl.dequeue() // 从流中移除空数据项
        _ = reader.Close()
        if str.itl.isEmpty() {
            str.state = empty
        } else if trailer, ok := str.itl.peek().(*headerFrame); ok { // 下一项是 trailers
            if err := l.writeHeader(trailer.streamID, trailer.endStream, trailer.hf, trailer.onWrite); err != nil {
                return false, err
            }
            if err := l.cleanupStreamHandler(trailer.cleanup); err != nil {
                return false, err
            }
        } else {
            l.activeStreams.enqueue(str)
        }
        return false, nil
    }

    // 计算最大发送大小
    maxSize := http2MaxFrameLen
    if strQuota := int(l.oiws) - str.bytesOutStanding; strQuota <= 0 { // 流级流控制
        str.state = waitingOnStreamQuota
        return false, nil
    } else if maxSize > strQuota {
        maxSize = strQuota
    }
    if maxSize > int(l.sendQuota) { // 连接级流控制
        maxSize = int(l.sendQuota)
    }
    
    // 计算头部和数据在配额和最大帧长度内可以发送多少
    hSize := min(maxSize, len(dataItem.h))
    dSize := min(maxSize-hSize, reader.Remaining())
    if hSize != 0 {
        if dSize == 0 {
            endStream := reader.Remaining() == 0 && dataItem.endStream
            if err := l.framer.fr.WriteData(dataItem.streamID, endStream, dataItem.h[:hSize]); err != nil {
                return false, err
            }
        } else {
            // 将头部和数据合并到一个帧中
            buf := make([]byte, hSize+dSize)
            copy(buf, dataItem.h[:hSize])
            reader.Read(buf[hSize:])
            endStream := reader.Remaining() == 0 && dataItem.endStream
            if err := l.framer.fr.WriteData(dataItem.streamID, endStream, buf); err != nil {
                return false, err
            }
        }
        dataItem.h = dataItem.h[hSize:]
    } else if dSize > 0 {
        buf := make([]byte, dSize)
        reader.Read(buf)
        endStream := reader.Remaining() == 0 && dataItem.endStream
        if err := l.framer.fr.WriteData(dataItem.streamID, endStream, buf); err != nil {
            return false, err
        }
    }
    
    // 更新配额
    size := hSize + dSize
    str.bytesOutStanding += size
    l.sendQuota -= uint32(size)
    
    if dataItem.onEachWrite != nil {
        dataItem.onEachWrite()
    }
    
    if reader.Remaining() == 0 && len(dataItem.h) == 0 {
        // 数据帧处理完成
        str.itl.dequeue()
        _ = reader.Close()
        if str.itl.isEmpty() {
            str.state = empty
        } else if trailer, ok := str.itl.peek().(*headerFrame); ok {
            if err := l.writeHeader(trailer.streamID, trailer.endStream, trailer.hf, trailer.onWrite); err != nil {
                return false, err
            }
            if err := l.cleanupStreamHandler(trailer.cleanup); err != nil {
                return false, err
            }
        } else {
            l.activeStreams.enqueue(str)
        }
    } else {
        l.activeStreams.enqueue(str)
    }
    return false, nil
}