Untitled-1 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171
  1. package main
  2. import (
  3. "io"
  4. "net"
  5. "time"
  6. )
  7. const defaultBufSize = 4096
  8. const maxCachedBufSize = 256 * 1024
  9. // 一个用于读写的缓冲区。
  10. // 由于每个连接上的通信是同步的,因此可以这样做。
  11. // 换句话说,我们不能在同一个连接上同时进行读写。
  12. // 该缓冲区类似于 bufio.Reader/Writer,但具有零拷贝特性。
  13. // 还针对这种特定用例进行了高度优化。
  14. // 这个缓冲区由两个字节切片组成,使用双缓冲机制。
  15. type buffer struct {
  16. buf []byte // buf 是一个字节缓冲区,其长度和容量相等。
  17. nc net.Conn
  18. idx int
  19. length int
  20. timeout time.Duration
  21. dbuf [2][]byte // dbuf 是一个包含两个字节切片的数组,用于支持这个缓冲区
  22. flipcnt uint // flipcnt 是双缓冲机制中的当前缓冲区计数器
  23. }
  24. // newBuffer 分配并返回一个新的缓冲区。
  25. func newBuffer(nc net.Conn) buffer {
  26. fg := make([]byte, defaultBufSize)
  27. return buffer{
  28. buf: fg,
  29. nc: nc,
  30. dbuf: [2][]byte{fg, nil},
  31. }
  32. }
  33. // flip 用后台缓冲区替换活动缓冲区
  34. // 这是一个延迟的翻转操作,只是简单地增加缓冲区计数器;
  35. // 实际的翻转将在下次调用 `buffer.fill` 时执行
  36. func (b *buffer) flip() {
  37. b.flipcnt += 1
  38. }
  39. // fill 将数据读入缓冲区,直到至少包含 _need_ 字节的数据
  40. func (b *buffer) fill(need int) error {
  41. n := b.length
  42. // 将数据填充到双缓冲目标:如果我们在这个缓冲区上调用了 flip,
  43. // 我们将复制到后台缓冲区,然后用网络数据填充它;否则我们只需在填充前将当前缓冲区的内容移到前面
  44. dest := b.dbuf[b.flipcnt&1]
  45. // 如果需要,增加缓冲区的大小以适应整个数据包。
  46. if need > len(dest) {
  47. // 向上舍入到下一个默认大小的倍数
  48. dest = make([]byte, ((need/defaultBufSize)+1)*defaultBufSize)
  49. // 如果分配的缓冲区不太大,将其移至支持存储,以防止在执行大读操作的应用程序上进行额外的分配
  50. if len(dest) <= maxCachedBufSize {
  51. b.dbuf[b.flipcnt&1] = dest
  52. }
  53. }
  54. // 如果我们填充的是 fg 缓冲区,将现有数据移动到其起始位置。
  55. // 如果我们填充的是 bg 缓冲区,复制数据
  56. if n > 0 {
  57. copy(dest[:n], b.buf[b.idx:])
  58. }
  59. b.buf = dest
  60. b.idx = 0
  61. for {
  62. if b.timeout > 0 {
  63. if err := b.nc.SetReadDeadline(time.Now().Add(b.timeout)); err != nil {
  64. return err
  65. }
  66. }
  67. nn, err := b.nc.Read(b.buf[n:])
  68. n += nn
  69. switch err {
  70. case nil:
  71. if n < need {
  72. continue
  73. }
  74. b.length = n
  75. return nil
  76. case io.EOF:
  77. if n >= need {
  78. b.length = n
  79. return nil
  80. }
  81. return io.ErrUnexpectedEOF
  82. default:
  83. return err
  84. }
  85. }
  86. }
  87. // 返回缓冲区中的下一个 N 个字节。
  88. // 返回的切片仅保证在下一次读取之前有效
  89. func (b *buffer) readNext(need int) ([]byte, error) {
  90. if b.length < need {
  91. // 重新填充
  92. if err := b.fill(need); err != nil {
  93. return nil, err
  94. }
  95. }
  96. offset := b.idx
  97. b.idx += need
  98. b.length -= need
  99. return b.buf[offset:b.idx], nil
  100. }
  101. // takeBuffer 返回一个具有请求大小的缓冲区。
  102. // 如果可能,从现有缓冲区返回一个切片。
  103. // 否则,创建一个更大的缓冲区。
  104. // 只能同时使用一个缓冲区(总共)。
  105. func (b *buffer) takeBuffer(length int) ([]byte, error) {
  106. if b.length > 0 {
  107. return nil, ErrBusyBuffer
  108. }
  109. // 先测试(便宜的)一般情况
  110. if length <= cap(b.buf) {
  111. return b.buf[:length], nil
  112. }
  113. if length < maxPacketSize {
  114. b.buf = make([]byte, length)
  115. return b.buf, nil
  116. }
  117. // 缓冲区大于我们要存储的内容。
  118. return make([]byte, length), nil
  119. }
  120. // takeSmallBuffer 是一个快捷方式,如果已知长度小于 defaultBufSize,可以使用。
  121. // 只能同时使用一个缓冲区(总共)。
  122. func (b *buffer) takeSmallBuffer(length int) ([]byte, error) {
  123. if b.length > 0 {
  124. return nil, ErrBusyBuffer
  125. }
  126. return b.buf[:length], nil
  127. }
  128. // takeCompleteBuffer 返回完整的现有缓冲区。
  129. // 如果必要的缓冲区大小未知,可以使用此方法。
  130. // 返回的缓冲区的 cap 和 len 将相等。
  131. // 只能同时使用一个缓冲区(总共)。
  132. func (b *buffer) takeCompleteBuffer() ([]byte, error) {
  133. if b.length > 0 {
  134. return nil, ErrBusyBuffer
  135. }
  136. return b.buf, nil
  137. }
  138. // store 存储缓冲区 buf,如果适合的话。
  139. // 如果其大小适中,则更新缓冲区。
  140. func (b *buffer) store(buf []byte) error {
  141. if b.length > 0 {
  142. return ErrBusyBuffer
  143. } else if cap(buf) <= maxPacketSize && cap(buf) > cap(b.buf) {
  144. b.buf = buf[:cap(buf)]
  145. }
  146. return nil
  147. }