123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171 |
- package main
- import (
- "io"
- "net"
- "time"
- )
- const defaultBufSize = 4096
- const maxCachedBufSize = 256 * 1024
- // 一个用于读写的缓冲区。
- // 由于每个连接上的通信是同步的,因此可以这样做。
- // 换句话说,我们不能在同一个连接上同时进行读写。
- // 该缓冲区类似于 bufio.Reader/Writer,但具有零拷贝特性。
- // 还针对这种特定用例进行了高度优化。
- // 这个缓冲区由两个字节切片组成,使用双缓冲机制。
- type buffer struct {
- buf []byte // buf 是一个字节缓冲区,其长度和容量相等。
- nc net.Conn
- idx int
- length int
- timeout time.Duration
- dbuf [2][]byte // dbuf 是一个包含两个字节切片的数组,用于支持这个缓冲区
- flipcnt uint // flipcnt 是双缓冲机制中的当前缓冲区计数器
- }
- // newBuffer 分配并返回一个新的缓冲区。
- func newBuffer(nc net.Conn) buffer {
- fg := make([]byte, defaultBufSize)
- return buffer{
- buf: fg,
- nc: nc,
- dbuf: [2][]byte{fg, nil},
- }
- }
- // flip 用后台缓冲区替换活动缓冲区
- // 这是一个延迟的翻转操作,只是简单地增加缓冲区计数器;
- // 实际的翻转将在下次调用 `buffer.fill` 时执行
- func (b *buffer) flip() {
- b.flipcnt += 1
- }
- // fill 将数据读入缓冲区,直到至少包含 _need_ 字节的数据
- func (b *buffer) fill(need int) error {
- n := b.length
- // 将数据填充到双缓冲目标:如果我们在这个缓冲区上调用了 flip,
- // 我们将复制到后台缓冲区,然后用网络数据填充它;否则我们只需在填充前将当前缓冲区的内容移到前面
- dest := b.dbuf[b.flipcnt&1]
- // 如果需要,增加缓冲区的大小以适应整个数据包。
- if need > len(dest) {
- // 向上舍入到下一个默认大小的倍数
- dest = make([]byte, ((need/defaultBufSize)+1)*defaultBufSize)
- // 如果分配的缓冲区不太大,将其移至支持存储,以防止在执行大读操作的应用程序上进行额外的分配
- if len(dest) <= maxCachedBufSize {
- b.dbuf[b.flipcnt&1] = dest
- }
- }
- // 如果我们填充的是 fg 缓冲区,将现有数据移动到其起始位置。
- // 如果我们填充的是 bg 缓冲区,复制数据
- if n > 0 {
- copy(dest[:n], b.buf[b.idx:])
- }
- b.buf = dest
- b.idx = 0
- for {
- if b.timeout > 0 {
- if err := b.nc.SetReadDeadline(time.Now().Add(b.timeout)); err != nil {
- return err
- }
- }
- nn, err := b.nc.Read(b.buf[n:])
- n += nn
- switch err {
- case nil:
- if n < need {
- continue
- }
- b.length = n
- return nil
- case io.EOF:
- if n >= need {
- b.length = n
- return nil
- }
- return io.ErrUnexpectedEOF
- default:
- return err
- }
- }
- }
- // 返回缓冲区中的下一个 N 个字节。
- // 返回的切片仅保证在下一次读取之前有效
- func (b *buffer) readNext(need int) ([]byte, error) {
- if b.length < need {
- // 重新填充
- if err := b.fill(need); err != nil {
- return nil, err
- }
- }
- offset := b.idx
- b.idx += need
- b.length -= need
- return b.buf[offset:b.idx], nil
- }
- // takeBuffer 返回一个具有请求大小的缓冲区。
- // 如果可能,从现有缓冲区返回一个切片。
- // 否则,创建一个更大的缓冲区。
- // 只能同时使用一个缓冲区(总共)。
- func (b *buffer) takeBuffer(length int) ([]byte, error) {
- if b.length > 0 {
- return nil, ErrBusyBuffer
- }
- // 先测试(便宜的)一般情况
- if length <= cap(b.buf) {
- return b.buf[:length], nil
- }
- if length < maxPacketSize {
- b.buf = make([]byte, length)
- return b.buf, nil
- }
- // 缓冲区大于我们要存储的内容。
- return make([]byte, length), nil
- }
- // takeSmallBuffer 是一个快捷方式,如果已知长度小于 defaultBufSize,可以使用。
- // 只能同时使用一个缓冲区(总共)。
- func (b *buffer) takeSmallBuffer(length int) ([]byte, error) {
- if b.length > 0 {
- return nil, ErrBusyBuffer
- }
- return b.buf[:length], nil
- }
- // takeCompleteBuffer 返回完整的现有缓冲区。
- // 如果必要的缓冲区大小未知,可以使用此方法。
- // 返回的缓冲区的 cap 和 len 将相等。
- // 只能同时使用一个缓冲区(总共)。
- func (b *buffer) takeCompleteBuffer() ([]byte, error) {
- if b.length > 0 {
- return nil, ErrBusyBuffer
- }
- return b.buf, nil
- }
- // store 存储缓冲区 buf,如果适合的话。
- // 如果其大小适中,则更新缓冲区。
- func (b *buffer) store(buf []byte) error {
- if b.length > 0 {
- return ErrBusyBuffer
- } else if cap(buf) <= maxPacketSize && cap(buf) > cap(b.buf) {
- b.buf = buf[:cap(buf)]
- }
- return nil
- }
|