golang实现HTTP2之协议处理

[BruceChen7]

参考资料

• golang 实现 HTTP2 之主流程

处理客户端发送

func (sc *serverConn) serve() {
    ....
    go sc.readFrame()
    ...
}
func (sc *serverConn) readFrames() {
    gate := make(gate)
    gateDone := gate.Done
    for {
        // 执行 IO 操作，并解析成一帧
        f, err := sc.framer.ReadFrame()
        select {
        case sc.readFrameCh <- readFrameResult{f, err, gateDone}:
        case <-sc.doneServing: // connection is finished
            return
        }
        select {
        case <-gate:
        case <-sc.doneServing:
            return
        }
        if terminalReadFrameError(err) {
            return
        }
    }
}

• readFrameCh 将解析获取的一帧数据返回给 main goroutine, 主要解析逻辑是 readFrame

  func (fr *Framer) ReadFrame() (Frame, error) {
      fr.errDetail = nil
      if fr.lastFrame != nil {
          fr.lastFrame.invalidate()
      }
      // 获取 frame header
      // fr.headerBuf[:]将数组转成 slice
      fh, err := readFrameHeader(fr.headerBuf[:], fr.r)
      if err != nil {
          return nil, err
      }
      // fheader 中的长度太大
      if fh.Length > fr.maxReadSize {
          return nil, ErrFrameTooLarge
      }
      // 获取 payload body
      payload := fr.getReadBuf(fh.Length)
      // 获取 payload
      if _, err := io.ReadFull(fr.r, payload); err != nil {
          return nil, err
      }
  
      // 根据 frame 解析器获取 frame
      f, err := typeFrameParser(fh.Type)(fr.frameCache, fh, payload)
      if err != nil {
          if ce, ok := err.(connError); ok {
              return nil, fr.connError(ce.Code, ce.Reason)
          }
          return nil, err
      }
      if err := fr.checkFrameOrder(f); err != nil {
          return nil, err
      }
      // 打印日志
      if fr.logReads {
          fr.debugReadLoggerf("http2: Framer %p: read %v", fr, summarizeFrame(f))
      }
      if fh.Type == FrameHeaders && fr.ReadMetaHeaders != nil {
          return fr.readMetaFrame(f.(*HeadersFrame))
      }
      return f, nil
  }
  // 读取 frameHeader
  // +-----------------------------------------------+
  // |                 Length (24)                   |
  // +---------------+---------------+---------------+
  // |   Type (8)    |   Flags (8)   |
  // +-+-------------+---------------+-------------------------------+
  // |R|                 Stream Identifier (31)                      |
  // +=+=============================================================+
  // |                   Frame Payload (0...)                      ...
  // +---------------------------------------------------------------+
  func readFrameHeader(buf []byte, r io.Reader) (FrameHeader, error) {
      // 读取 frame 长度，前 9 个字节
      _, err := io.ReadFull(r, buf[:frameHeaderLen])
      if err != nil {
          return FrameHeader{}, err
      }
      return FrameHeader{
          // frame 长度，低地址放高字节，属于网络字节序列
          Length: (uint32(buf[0])<<16 | uint32(buf[1])<<8 | uint32(buf[2])),
          // frame 类型
          Type:     FrameType(buf[3]),
          Flags:    Flags(buf[4]),
          StreamID: binary.BigEndian.Uint32(buf[5:]) & (1<<31 - 1),
          valid:    true,
      }, nil
  }

• 这里采用的大端字节序。
• ReadFrame 逻辑：
  • 获取 frame 的头，并且从头中获取 payload body 的长度
  • 然后获取 payload body，根据 header 的类型来解析成一个 frame

• 这里的类型包含如下

  var frameParsers = map[FrameType]frameParser{
      FrameData:         parseDataFrame,
      FrameHeaders:      parseHeadersFrame,
      FramePriority:     parsePriorityFrame,
      FrameRSTStream:    parseRSTStreamFrame,
      FrameSettings:     parseSettingsFrame,
      FramePushPromise:  parsePushPromise,
      FramePing:         parsePingFrame,
      FrameGoAway:       parseGoAwayFrame,
      FrameWindowUpdate: parseWindowUpdateFrame,
      FrameContinuation: parseContinuationFrame,
  }
  
  func typeFrameParser(t FrameType) frameParser {
      if f := frameParsers[t]; f != nil {
          return f
      }
      return parseUnknownFrame
  }

• readFrameCh 中获取的数据，通知到主 goroutine，然后开始来逻辑处理：

  func (sc *serverConn) serve() {
     ....
      case res := <-sc.readFrameCh:
          // 获取从客户端读取的一帧的结果
          if !sc.processFrameFromReader(res) {
              return
          }
          res.readMore()
          if settingsTimer != nil {
              settingsTimer.Stop()
              settingsTimer = nil
          }
      ...
  }
  func (sc *serverConn) processFrameFromReader(res readFrameResult) bool {
      sc.serveG.check()
      err := res.err
      if err != nil {
          // freame 太大
          if err == ErrFrameTooLarge {
              sc.goAway(ErrCodeFrameSize)
              return true // goAway will close the loop
          }
          clientGone := err == io.EOF || err == io.ErrUnexpectedEOF || isClosedConnError(err)
          if clientGone {
              // TODO: could we also get into this state if
              // the peer does a half close
              // (e.g. CloseWrite) because they're done
              // sending frames but they're still wanting
              // our open replies?  Investigate.
              // TODO: add CloseWrite to crypto/tls.Conn first
              // so we have a way to test this? I suppose
              // just for testing we could have a non-TLS mode.
              return false
          }
      } else {
          // 获取一帧
          f := res.f
          if VerboseLogs {
              // 获取一帧
              sc.vlogf("http2: server read frame %v", summarizeFrame(f))
          }
          // 处理一帧
          err = sc.processFrame(f)
          if err == nil {
              return true
          }
      }
  
      switch ev := err.(type) {
      case StreamError:
          // 重置 stream
          sc.resetStream(ev)
          return true
      case goAwayFlowError:
          sc.goAway(ErrCodeFlowControl)
          return true
      case ConnectionError:
          sc.logf("http2: server connection error from %v: %v", sc.conn.RemoteAddr(), ev)
          sc.goAway(ErrCode(ev))
          return true // goAway will handle shutdown
      default:
          if res.err != nil {
              sc.vlogf("http2: server closing client connection; error reading frame from client %s: %v", sc.conn.RemoteAddr(), err)
          } else {
              sc.logf("http2: server closing client connection: %v", err)
          }
          return false
      }
  }

• 查看当前的 frame 是否太大，太大直接发 goAway 帧，
• 出来客户端关闭写的情况，这里留下了未实现的部分

• 开始处理不同类型的帧

  // 处理各种类型的帧
  func (sc *serverConn) processFrame(f Frame) error {
      sc.serveG.check()
  
      // First frame received must be SETTINGS.
      if !sc.sawFirstSettings {
          if _, ok := f.(*SettingsFrame); !ok {
              return ConnectionError(ErrCodeProtocol)
          }
          // 第一帧必须是 setting 帧
          sc.sawFirstSettings = true
      }
  
      switch f := f.(type) {
      case *SettingsFrame:
          /// 处理 setting 帧
          return sc.processSettings(f)
      case *MetaHeadersFrame:
          return sc.processHeaders(f)
      case *WindowUpdateFrame:
          return sc.processWindowUpdate(f)
      case *PingFrame: // ping frame
          return sc.processPing(f)
      case *DataFrame: // data frame
          return sc.processData(f)
      case *RSTStreamFrame:
          return sc.processResetStream(f)
      case *PriorityFrame:
          return sc.processPriority(f)
          // 处理 GoAway，用来通知客户端，要重新建立链接，这个链接的 stream 满了
      case *GoAwayFrame:
          return sc.processGoAway(f)
      case *PushPromiseFrame:
          // A client cannot push. Thus, servers MUST treat the receipt of a PUSH_PROMISE
          // frame as a connection error (Section 5.4.1) of type PROTOCOL_ERROR.
          return ConnectionError(ErrCodeProtocol)
      default:
          sc.vlogf("http2: server ignoring frame: %v", f.Header())
          return nil
      }
  }

• 这个连接没有处理过第一帧，那么这一帧必须是 setting 帧
• 然后根据不同的帧类型来处理

处理 setting 帧

func (sc *serverConn) processSettings(f *SettingsFrame) error {
    sc.serveG.check()
    // 是客户端对 server 的 ack
    if f.IsAck() {
        sc.unackedSettings--
        if sc.unackedSettings < 0 {
            // Why is the peer ACKing settings we never sent?
            // The spec doesn't mention this case, but
            // hang up on them anyway.
            return ConnectionError(ErrCodeProtocol)
        }
        return nil
    }
    // 数量不对
    if f.NumSettings() > 100 || f.HasDuplicates() {
        // This isn't actually in the spec, but hang up on
        // suspiciously large settings frames or those with
        // duplicate entries.
        return ConnectionError(ErrCodeProtocol)
    }
    if err := f.ForeachSetting(sc.processSetting); err != nil {
        return err
    }
    // TODO: judging by RFC 7540, Section 6.5.3 each SETTINGS frame should be
    // acknowledged individually, even if multiple are received before the ACK.
    // 需要应达 setting 的 ack
    sc.needToSendSettingsAck = true
    sc.scheduleFrameWrite()
    return nil
}

type/golang #public