tx_batch.go 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859
  1. package batch
  2. import "net/netip"
  3. const SendBatchCap = 128
  4. // batchWriter is the minimal subset of udp.Conn needed by SendBatch to flush.
  5. type batchWriter interface {
  6. WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error)
  7. }
  8. // SendBatch accumulates encrypted UDP packets and flushes them via WriteBatch.
  9. // Each reader goroutine owns its own SendBatch; no locking is needed.
  10. // Slots can borrow receive buffers or be backed by an Arena (see its docs).
  11. type SendBatch struct {
  12. out batchWriter
  13. bufs [][]byte
  14. dsts []netip.AddrPort
  15. arena *Arena
  16. }
  17. // NewSendBatch makes a SendBatch with batchCap slots and an arenaSize byte buffer for slices to back those slots
  18. func NewSendBatch(out batchWriter, batchCap, arenaSize int) *SendBatch {
  19. return &SendBatch{
  20. out: out,
  21. bufs: make([][]byte, 0, batchCap),
  22. dsts: make([]netip.AddrPort, 0, batchCap),
  23. arena: NewArena(arenaSize),
  24. }
  25. }
  26. func (b *SendBatch) Reserve(sz int) []byte {
  27. return b.arena.Reserve(sz)
  28. }
  29. // Len reports how many packets are queued for the next Flush. Callers use
  30. // it to flush incrementally once a full sendmmsg batch has accumulated,
  31. // bounding how long the first packet of a large read batch waits.
  32. func (b *SendBatch) Len() int { return len(b.bufs) }
  33. func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort) {
  34. b.bufs = append(b.bufs, pkt)
  35. b.dsts = append(b.dsts, dst)
  36. }
  37. // Flush writes every queued packet and reports how many actually went out. A short count means some destinations
  38. // were undeliverable; the batch is drained either way.
  39. func (b *SendBatch) Flush() (int, error) {
  40. var err error
  41. written := 0
  42. if len(b.bufs) > 0 {
  43. written, err = b.out.WriteBatch(b.bufs, b.dsts)
  44. }
  45. clear(b.bufs)
  46. b.bufs = b.bufs[:0]
  47. b.dsts = b.dsts[:0]
  48. b.arena.Reset()
  49. return written, err
  50. }